Class: Karafka::Pro::Processing::ConsumerGroups::Filters::Expirer

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

Overview

Expirer for removing too old messages. It never moves offsets in any way and does not impact the processing flow. It always runs the Actions.skip action.

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(ttl) ⇒ Expirer

Returns a new instance of Expirer.

Parameters:

  • ttl (Integer) —

    maximum age of a message (in ms)



41
42
43
44
45
 
# File 'lib/karafka/pro/processing/consumer_groups/filters/expirer.rb', line 41

def initialize(ttl)
  super()

  @ttl = ttl
end

Instance Method Details

#apply!(messages) ⇒ Object

Removes too old messages

Parameters:



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/karafka/pro/processing/consumer_groups/filters/expirer.rb', line 50

def apply!(messages)
  @applied = false

  # Time on message is in seconds with ms precision, so we need to convert the ttl that
  # is in ms to this format
  border = Time.now.utc - (@ttl / 1_000.to_f)

  messages.delete_if do |message|
    too_old = message.timestamp < border

    @applied = true if too_old

    too_old
  end
end

#timeout ⇒ nil

Returns this filter does not deal with timeouts.

Returns:

  • (nil) —

    this filter does not deal with timeouts



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

def timeout
  nil
end