Module: Karafka::Pro::Processing::Strategies::Aj::FtrLrjMomVp
- Includes:
- Lrj::FtrMom, Vp::Default
- Defined in:
- lib/karafka/pro/processing/strategies/aj/ftr_lrj_mom_vp.rb
Overview
ActiveJob enabled Filtering enabled Long-Running Job enabled Manual offset management enabled Virtual Partitions enabled
Constant Summary collapse
- FEATURES =
Features for this strategy
%i[ active_job filtering long_running_job manual_offset_management virtual_partitions ].freeze
Constants included from Lrj::Default
Instance Method Summary collapse
-
#handle_after_consume ⇒ Object
AJ MOM VP does not do intermediate marking, hence we need to make sure we mark as consumed here.
Methods included from Lrj::Default
#handle_before_enqueue, #handle_revoked
Methods included from Default
#handle_before_consume, #handle_before_enqueue, #handle_consume, #handle_revoked
Methods included from Karafka::Processing::Strategies::Default
#commit_offsets, #commit_offsets!, #handle_before_consume, #handle_before_enqueue, #handle_consume, #handle_idle, #handle_revoked, #handle_shutdown, #mark_as_consumed, #mark_as_consumed!
Methods included from Karafka::Processing::Strategies::Base
#handle_before_consume, #handle_before_enqueue, #handle_consume, #handle_idle, #handle_revoked, #handle_shutdown
Methods included from Ftr::Default
#handle_idle, #handle_post_filtering
Methods included from Vp::Default
#collapsed?, #failing?, #mark_as_consumed, #mark_as_consumed!
Instance Method Details
#handle_after_consume ⇒ Object
AJ MOM VP does not do intermediate marking, hence we need to make sure we mark as consumed here.
39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 |
# File 'lib/karafka/pro/processing/strategies/aj/ftr_lrj_mom_vp.rb', line 39 def handle_after_consume coordinator.on_finished do || if coordinator.success? coordinator.pause_tracker.reset mark_as_consumed() unless revoked? if coordinator.filtered? && !revoked? handle_post_filtering elsif !revoked? # no need to check for manual seek because AJ consumer is internal and # fully controlled by us seek(coordinator.seek_offset, false) resume else resume end else retry_after_pause end end end |