Module: Karafka::Pro::Processing::Strategies::Mom::Ftr

Includes:
Ftr::Default, Default
Included in:
Aj::FtrMom, FtrVp
Defined in:
lib/karafka/pro/processing/strategies/mom/ftr.rb

Overview

Filtering support for MoM

Constant Summary collapse

FEATURES =

MoM + Ftr

%i[
  filtering
  manual_offset_management
].freeze

Instance Method Summary collapse

Methods included from Default

#handle_before_consume, #handle_before_schedule_consume, #handle_before_schedule_tick, #handle_consume, #handle_revoked, #handle_tick, #mark_as_consumed, #mark_as_consumed!, #mark_in_memory, #mark_in_transaction, #mark_with_transaction, #store_offset_metadata, #transaction

Methods included from Karafka::Processing::Strategies::Default

#commit_offsets, #commit_offsets!, #handle_before_consume, #handle_consume, #handle_eofed, #handle_idle, #handle_initialized, #handle_revoked, #handle_shutdown, #handle_wrap, #mark_as_consumed, #mark_as_consumed!

Methods included from Karafka::Processing::Strategies::Base

#handle_before_consume, #handle_consume, #handle_idle, #handle_revoked, #handle_shutdown

Methods included from Ftr::Default

#handle_idle, #handle_post_filtering

Instance Method Details

#handle_after_consume ⇒ Object

When mom is enabled, we do not mark messages as consumed after processing but we also need to keep in mind throttling here

[View source]

24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/karafka/pro/processing/strategies/mom/ftr.rb', line 24

def handle_after_consume
  coordinator.on_finished do
    return if revoked?

    if coordinator.success?
      coordinator.pause_tracker.reset

      # Do not throttle if paused
      return if coordinator.manual_pause?

      handle_post_filtering
    else
      retry_after_pause
    end
  end
end