Module: Karafka::Pro::Processing::Strategies::AjMomVp

Includes:
Default
Defined in:
lib/karafka/pro/processing/strategies/aj_mom_vp.rb

Overview

ActiveJob enabled Manual offset management enabled Virtual Partitions enabled

Constant Summary collapse

FEATURES =

Features for this strategy

%i[
  active_job
  manual_offset_management
  virtual_partitions
].freeze

Instance Method Summary collapse

Methods included from Default

#handle_before_consume, #handle_before_enqueue, #handle_consume, #handle_revoked

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

#handle_before_consume, #handle_before_enqueue, #handle_consume, #handle_revoked, #handle_shutdown

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

#handle_before_consume, #handle_before_enqueue, #handle_consume, #handle_revoked, #handle_shutdown

Instance Method Details

#handle_after_consume ⇒ Object

Standard flow without any features



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/karafka/pro/processing/strategies/aj_mom_vp.rb', line 32

def handle_after_consume
  coordinator.on_finished do |last_group_message|
    if coordinator.success?
      coordinator.pause_tracker.reset

      # When this is an ActiveJob running via Pro with virtual partitions, we cannot mark
      # intermediate jobs as processed not to mess up with the ordering.
      # Only when all the jobs are processed and we did not loose the partition
      # assignment and we are not stopping (Pro ActiveJob has an early break) we can
      # commit offsets on this as only then we can be sure, that all the jobs were
      # processed.
      # For a non virtual partitions case, the flow is regular and state is marked after
      # each successfully processed job
      return if revoked?
      return if Karafka::App.stopping?

      mark_as_consumed(last_group_message)
    else
      retry_after_pause
    end
  end
end