Module: Karafka::Pro::Processing::Strategies::LrjMom

Includes:
Default
Included in:
AjLrjMom
Defined in:
lib/karafka/pro/processing/strategies/lrj_mom.rb

Overview

Long-Running Job enabled Manual offset management enabled

Constant Summary collapse

FEATURES =

Features for this strategy

%i[
  long_running_job
  manual_offset_management
].freeze

Instance Method Summary collapse

Methods included from Default

#handle_before_consume, #handle_consume

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

#handle_before_consume, #handle_consume, #handle_shutdown

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

#handle_before_consume, #handle_consume, #handle_shutdown

Instance Method Details

#handle_after_consume ⇒ Object

No offset management, aside from that typical LRJ



44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/karafka/pro/processing/strategies/lrj_mom.rb', line 44

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

      seek(coordinator.seek_offset) unless revoked?

      resume
    else
      retry_after_pause
    end
  end
end

#handle_before_enqueue ⇒ Object

We always need to pause prior to doing any jobs for LRJ



30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/karafka/pro/processing/strategies/lrj_mom.rb', line 30

def handle_before_enqueue
  # This ensures that when running LRJ with VP, things operate as expected run only once
  # for all the virtual partitions collectively
  coordinator.on_enqueued do
    # Pause at the first message in a batch. That way in case of a crash, we will not
    # loose any messages.
    #
    # For VP it applies the same way and since VP cannot be used with MOM we should not
    # have any edge cases here.
    pause(coordinator.seek_offset, Lrj::MAX_PAUSE_TIME, false)
  end
end

#handle_revoked ⇒ Object

We do not un-pause on revokations for LRJ



59
60
61
62
63
64
65
66
67
68
 
# File 'lib/karafka/pro/processing/strategies/lrj_mom.rb', line 59

def handle_revoked
  coordinator.on_revoked do
    coordinator.revoke
  end

  Karafka.monitor.instrument('consumer.revoke', caller: self)
  Karafka.monitor.instrument('consumer.revoked', caller: self) do
    revoked
  end
end