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

Lrj::Default::MAX_PAUSE_TIME

Instance Method Summary collapse

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 |last_group_message|
    if coordinator.success?
      coordinator.pause_tracker.reset

      mark_as_consumed(last_group_message) 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