Contents

# frozen_string_literal: true

module Karafka
  module Processing
    module Strategies
      # Same as pure dead letter queue but we do not marked failed message as consumed
      module DlqMom
        include Dlq

        # Apply strategy when dlq is on with manual offset management
        FEATURES = %i[
          dead_letter_queue
          manual_offset_management
        ].freeze

        # When manual offset management is on, we do not mark anything as consumed automatically
        # and we rely on the user to figure things out
        def handle_after_consume
          return if revoked?

          if coordinator.success?
            coordinator.pause_tracker.reset
          elsif coordinator.pause_tracker.attempt <= topic.dead_letter_queue.max_retries
            retry_after_pause
          # If we've reached number of retries that we could, we need to skip the first message
          # that was not marked as consumed, pause and continue, while also moving this message
          # to the dead topic
          else
            # We reset the pause to indicate we will now consider it as "ok".
            coordinator.pause_tracker.reset

            skippable_message, marked = find_skippable_message

            dispatch_to_dlq(skippable_message)

            # Backoff and move forward
            if marked
              pause(coordinator.seek_offset, nil, false)
            else
              pause(skippable_message.offset + 1, nil, false)
            end
          end
        end
      end
    end
  end
end

Version data entries

3 entries across 3 versions & 1 rubygems

Version	Path
karafka-2.0.41	lib/karafka/processing/strategies/dlq_mom.rb
karafka-2.0.40	lib/karafka/processing/strategies/dlq_mom.rb
karafka-2.0.39	lib/karafka/processing/strategies/dlq_mom.rb