Module: Karafka::Pro::Processing::Strategies::Aj::DlqLrjMom

Includes:
LrjMom, Default, Dlq::Default
Defined in:
lib/karafka/pro/processing/strategies/aj/dlq_lrj_mom.rb

Overview

ActiveJob enabled DLQ enabled Long-Running Job enabled Manual offset management enabled

This case is a bit of special. Please see the ‘AjDlqMom` for explanation on how the offset management works in this case.

Constant Summary collapse

FEATURES =

Features for this strategy

%i[
  active_job
  dead_letter_queue
  long_running_job
  manual_offset_management
].freeze

Instance Method Summary collapse

Methods included from Lrj::Mom

#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 Dlq::Default

#build_dlq_message, #dispatch_to_dlq, #dispatch_to_dlq?, #find_skippable_message

Instance Method Details

#handle_after_consume ⇒ Object

We cannot use a VP version of this, because non-VP can early stop on shutdown



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/dlq_lrj_mom.rb', line 40

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

      # no need to check for manual seek because AJ consumer is internal and
      # fully controlled by us
      seek(coordinator.seek_offset, false) unless revoked?

      resume
    elsif coordinator.pause_tracker.attempt <= topic.dead_letter_queue.max_retries
      retry_after_pause
    else
      coordinator.pause_tracker.reset
      skippable_message, = find_skippable_message
      dispatch_to_dlq(skippable_message) if dispatch_to_dlq?
      mark_as_consumed(skippable_message)
      pause(coordinator.seek_offset, nil, false)
    end
  end
end