Module: Karafka::Pro::Processing::Strategies::Aj::DlqFtrMomVp

Includes:
DlqFtrMom, DlqMomVp
Defined in:
lib/karafka/pro/processing/strategies/aj/dlq_ftr_mom_vp.rb

Overview

  • Aj

  • Dlq

  • Ftr

  • Mom

  • VP

Constant Summary collapse

FEATURES =

Features for this strategy

%i[
  active_job
  dead_letter_queue
  filtering
  manual_offset_management
  virtual_partitions
].freeze

Instance Method Summary collapse

Methods included from Dlq::FtrMom

#mark_after_dispatch?

Methods included from Dlq::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

Methods included from 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 Ftr::Default

#handle_idle, #handle_post_filtering

Methods included from Vp::Default

#collapse_until!, #collapsed?, #failing?, #mark_as_consumed, #mark_as_consumed!, #mark_in_transaction, #synchronize

Instance Method Details

#handle_after_consume ⇒ Object

AJ VP does not early stop on shutdown, hence here we can mark as consumed at the end of all VPs



39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/karafka/pro/processing/strategies/aj/dlq_ftr_mom_vp.rb', line 39

def handle_after_consume
  coordinator.on_finished do |last_group_message|
    return if revoked?

    if coordinator.success?
      coordinator.pause_tracker.reset

      return if coordinator.manual_pause?

      mark_as_consumed(last_group_message)

      handle_post_filtering
    else
      apply_dlq_flow do
        skippable_message, = find_skippable_message
        dispatch_to_dlq(skippable_message) if dispatch_to_dlq?
        # We can commit the offset here because we know that we skip it "forever" and
        # since AJ consumer commits the offset after each job, we also know that the
        # previous job was successful
        mark_dispatched_to_dlq(skippable_message)
      end
    end
  end
end