Module: Karafka::Pro::Processing::Strategies::Dlq::Lrj

Includes:
Default, Lrj::Default
Included in:
FtrLrj, LrjMom, LrjVp
Defined in:
lib/karafka/pro/processing/strategies/dlq/lrj.rb

Overview

DLQ enabled Long-Running Job enabled

Constant Summary collapse

FEATURES =

Features for this strategy

%i[
  dead_letter_queue
  long_running_job
].freeze

Constants included from Lrj::Default

Lrj::Default::MAX_PAUSE_TIME

Instance Method Summary collapse

Methods included from Lrj::Default

#handle_before_schedule_consume, #handle_revoked, #synchronize

Methods included from Karafka::Pro::Processing::Strategies::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_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, #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 Default

#apply_dlq_flow, #build_dlq_message, #dispatch_if_needed_and_mark_as_consumed, #dispatch_in_a_transaction?, #dispatch_to_dlq, #dispatch_to_dlq?, #find_skippable_message, #mark_after_dispatch?, #mark_as_consumed, #mark_as_consumed!, #mark_dispatched_to_dlq

Instance Method Details

#handle_after_consume ⇒ Object

LRJ standard flow after consumption with DLQ dispatch



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/dlq/lrj.rb', line 33

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

      return if coordinator.manual_pause?

      mark_as_consumed(last_group_message) unless revoked?
      # We should not overwrite user manual seel request with our seek
      seek(coordinator.seek_offset, false) unless revoked? || coordinator.manual_seek?

      resume
    else
      apply_dlq_flow do
        return resume if revoked?

        dispatch_if_needed_and_mark_as_consumed
      end
    end
  end
end