Module: Karafka::Pro::Processing::Strategies::Lrj::Default

Includes:
Default
Included in:
Dlq::Lrj, Ftr, Vp
Defined in:
lib/karafka/pro/processing/strategies/lrj/default.rb

Overview

Long-Running Job enabled

Constant Summary collapse

MAX_PAUSE_TIME =

Pause for tops 31 years

1_000_000_000_000
FEATURES =

Features for this strategy

%i[
  long_running_job
].freeze

Instance Method Summary collapse

Methods included from Default

#handle_before_consume, #handle_before_schedule_tick, #handle_consume, #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_shutdown, #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

LRJ standard flow after consumption



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/karafka/pro/processing/strategies/lrj/default.rb', line 46

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?
      seek(coordinator.seek_offset, false) unless revoked? || coordinator.manual_seek?

      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

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



32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/karafka/pro/processing/strategies/lrj/default.rb', line 32

def handle_before_schedule_consume
  super

  # 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 and continue with another batch in case of a regular resume.
    # In case of an error, the `#retry_after_pause` will move the offset to the first
    # message out of this batch.
    pause(:consecutive, MAX_PAUSE_TIME, false)
  end
end

#handle_revoked ⇒ Object

We do not un-pause on revokations for LRJ



67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/karafka/pro/processing/strategies/lrj/default.rb', line 67

def handle_revoked
  coordinator.on_revoked do
    # We do not want to resume on revocation in case of a LRJ.
    # For LRJ we resume after the successful processing or do a backoff pause in case
    # of a failure. Double non-blocking resume could cause problems in coordination.
    coordinator.revoke
  end

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

#synchronize(&block) ⇒ Object

Allows for LRJ to synchronize its work. It may be needed because LRJ can run lifecycle events like revocation while the LRJ work is running and there may be a need for a critical section.

Parameters:

  • block (Proc) —

    block we want to run in a mutex to prevent race-conditions



88
89
90
 
# File 'lib/karafka/pro/processing/strategies/lrj/default.rb', line 88

def synchronize(&block)
  coordinator.shared_mutex.synchronize(&block)
end