Module: Karafka::Pro::Processing::Strategies::Aj::LrjMomVp

Includes:
Default, Vp::Default
Defined in:
lib/karafka/pro/processing/strategies/aj/lrj_mom_vp.rb

Overview

ActiveJob enabled Long-Running Job enabled Manual offset management enabled Virtual Partitions enabled

Constant Summary collapse

FEATURES =

Features for this strategy

%i[
  active_job
  long_running_job
  manual_offset_management
  virtual_partitions
].freeze

Instance Method Summary collapse

Methods included from Vp::Default

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

Methods included from Default

#handle_before_consume, #handle_before_schedule_tick, #handle_consume, #handle_tick, #mark_as_consumed, #mark_as_consumed!, #mark_in_memory, #mark_in_transaction, #mark_with_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_shutdown, #handle_wrap, #mark_as_consumed, #mark_as_consumed!

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

#handle_before_consume, #handle_consume, #handle_idle, #handle_shutdown

Instance Method Details

#handle_after_consume ⇒ Object

Standard flow without any features



62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/karafka/pro/processing/strategies/aj/lrj_mom_vp.rb', line 62

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?

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

      resume
    else
      # If processing failed, we need to pause
      # For long running job this will overwrite the default never-ending pause and
      # will cause the processing to keep going after the error backoff
      retry_after_pause
    end
  end
end

#handle_before_schedule_consume ⇒ Object

No actions needed for the standard flow here



53
54
55
56
57
58
59
 
# File 'lib/karafka/pro/processing/strategies/aj/lrj_mom_vp.rb', line 53

def handle_before_schedule_consume
  super

  coordinator.on_enqueued do
    pause(:consecutive, Strategies::Lrj::Default::MAX_PAUSE_TIME, false)
  end
end

#handle_revoked ⇒ Object

LRJ cannot resume here. Only in handling the after consumption



84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/karafka/pro/processing/strategies/aj/lrj_mom_vp.rb', line 84

def handle_revoked
  coordinator.on_revoked do
    coordinator.revoke
  end

  monitor.instrument("consumer.revoke", caller: self)
  monitor.instrument("consumer.revoked", caller: self) do
    revoked
  end
ensure
  coordinator.decrement(:revoked)
end