Module: Karafka::Pro::Processing::ConsumerGroups::Strategies::Dlq::Vp

Includes:
Default, Vp::Default
Included in:
Aj::DlqMomVp, FtrLrjMomVp, FtrLrjVp, FtrMomVp, FtrVp, LrjMomVp, LrjVp, MomVp
Defined in:
lib/karafka/pro/processing/consumer_groups/strategies/dlq/vp.rb

Overview

  • Dead Letter Queue enabled
  • Virtual Partitions enabled

In general because we collapse processing in virtual partitions to one on errors, there is no special action that needs to be taken because we warranty that even with VPson errors a retry collapses into a single state and from this single state we can mark as consumed the message that we are moving to the DLQ.

Constant Summary collapse

FEATURES =

Features for this strategy

%i[
  dead_letter_queue
  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 Karafka::Pro::Processing::ConsumerGroups::Strategies::Default

#handle_after_consume, #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_memory, #mark_in_transaction, #mark_with_transaction, #store_offset_metadata, #transaction

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

#commit_offsets, #commit_offsets!, #handle_after_consume, #handle_before_consume, #handle_consume, #handle_eofed, #handle_idle, #handle_initialized, #handle_revoked, #handle_shutdown, #handle_wrap, #mark_as_consumed, #mark_as_consumed!

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

#handle_after_consume, #handle_before_consume, #handle_consume, #handle_idle, #handle_revoked, #handle_shutdown

Methods included from Default

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

Instance Method Details

#apply_dlq_flow ⇒ Object

Runs the DLQ strategy and based on it it performs certain operations

In case of :skip and :dispatch will run the exact flow provided in a block In case of :retry always #retry_after_pause is applied



58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/karafka/pro/processing/consumer_groups/strategies/dlq/vp.rb', line 58

def apply_dlq_flow
  # Process-critical errors are never dispatched or skipped regardless of the
  # strategy outcome: the retry pause protects the partition during the critical
  # shutdown and the failed batch is redelivered after the restart.
  # We consult `errors_tracker.last` (not the per-consumer consumption cause used
  # by the OSS strategies, which have no tracker) because it is exactly what the
  # DLQ strategy callable below receives - this guard judges the same evidence as
  # the strategy it overrides. The tracker is cleared at attempt zero, so `last`
  # is always the most recent failure of the current failure streak
  if critical_error?(errors_tracker.last)
    retry_after_pause

    return
  end

  # With virtual partitions, a dispatch/skip decision is never made on a
  # non-collapsed (parallel) run. The deciding consumer operates only on its own
  # virtual partition subset there: the skippable message would be selected from
  # an arbitrary subset and the dispatch marking would commit offsets of messages
  # other virtual partitions never processed. The failure already requested a
  # collapse, so we retry and let the decision happen on the collapsed, linear
  # flow where it is deterministic
  if topic.virtual_partitions? && !collapsed?
    retry_after_pause

    return
  end

  flow, target_topic = topic.dead_letter_queue.strategy.call(errors_tracker, attempt)

  case flow
  when :retry
    retry_after_pause

    return
  when :skip
    @_dispatch_to_dlq = false
  when :dispatch
    @_dispatch_to_dlq = true
    # Use custom topic if it was returned from the strategy
    @_dispatch_to_dlq_topic = target_topic || topic.dead_letter_queue.topic
  else
    raise Karafka::UnsupportedCaseError, flow
  end

  yield

  # We reset the pause to indicate we will now consider it as "ok".
  coordinator.pause_tracker.reset

  # Always backoff after DLQ dispatch even on skip to prevent overloads on errors
  pause(seek_offset, nil, false)
ensure
  @_dispatch_to_dlq_topic = nil
end