Class: Karafka::Pro::Processing::ConsumerGroups::Filters::VirtualLimiter

Inherits:
Base
  • Object
show all
Defined in:
lib/karafka/pro/processing/consumer_groups/filters/virtual_limiter.rb

Overview

Note:

It should be registered only when VPs are used

Removes messages that are already marked as consumed in the virtual offset manager This should operate only when using virtual partitions.

This cleaner prevents us from duplicated processing of messages that were virtually marked as consumed even if we could not mark them as consumed in Kafka. This allows us to limit reprocessing when errors occur drastically when operating with virtual partitions

Constant Summary

Constants included from Actions

Actions::ALL

Instance Attribute Summary

Attributes inherited from Base

#cursor

Instance Method Summary collapse

Methods inherited from Base

#action, #applied?, #mark_as_consumed?, #marking_cursor, #marking_method

Methods included from Actions

pause, #pause?, seek, #seek?, skip, #skip?

Constructor Details

#initialize(manager, collapser) ⇒ VirtualLimiter

Returns a new instance of VirtualLimiter.

Parameters:



48
49
50
51
52
53
 
# File 'lib/karafka/pro/processing/consumer_groups/filters/virtual_limiter.rb', line 48

def initialize(manager, collapser)
  @manager = manager
  @collapser = collapser

  super()
end

Instance Method Details

#apply!(messages) ⇒ Object

Remove messages that we already marked as virtually consumed. Does nothing if not in the collapsed mode.

Parameters:



59
60
61
62
63
64
65
 
# File 'lib/karafka/pro/processing/consumer_groups/filters/virtual_limiter.rb', line 59

def apply!(messages)
  return unless @collapser.collapsed?

  marked = @manager.marked

  messages.delete_if { |message| marked.include?(message.offset) }
end

#timeout ⇒ nil

Returns This filter does not deal with pausing, so timeout is always nil.

Returns:

  • (nil) —

    This filter does not deal with pausing, so timeout is always nil



68
69
70
 
# File 'lib/karafka/pro/processing/consumer_groups/filters/virtual_limiter.rb', line 68

def timeout
  nil
end