Class: Karafka::Pro::Processing::FiltersApplier

Inherits:
Object
  • Object
show all
Defined in:
lib/karafka/pro/processing/filters_applier.rb

Overview

Applier for all filters we want to have. Whether related to limiting messages based on the payload or any other things.

From the outside world perspective, this encapsulates all the filters. This means that this is the API we expose as a single filter, allowing us to control the filtering via many filters easily.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(coordinator) ⇒ FiltersApplier

Returns a new instance of FiltersApplier.

Parameters:

  • coordinator (Pro::Coordinator) —

    pro coordinator



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/karafka/pro/processing/filters_applier.rb', line 29

def initialize(coordinator)
  # Builds filters out of their factories
  # We build it that way (providing topic and partition) because there may be a case where
  # someone wants to have a specific logic that is per topic or partition. Like for example
  # a case where there is a cache bypassing revocations for topic partition.
  #
  # We provide full Karafka routing topic here and not the name only, in case the filter
  # would be customized based on other topic settings (like VPs, etc)
  #
  # This setup allows for biggest flexibility also because topic object holds the reference
  # to the subscription group and consumer group
  @filters = coordinator.topic.filtering.factories.map do |factory|
    factory.call(coordinator.topic, coordinator.partition)
  end
end

Instance Attribute Details

#filters ⇒ Array (readonly)

Returns registered filters array. Useful if we want to inject internal context aware filters.

Returns:

  • (Array) —

    registered filters array. Useful if we want to inject internal context aware filters.



26
27
28
 
# File 'lib/karafka/pro/processing/filters_applier.rb', line 26

def filters
  @filters
end

Instance Method Details

#action ⇒ Symbol

Returns consumer post-filtering action that should be taken.

Returns:

  • (Symbol) —

    consumer post-filtering action that should be taken



61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/karafka/pro/processing/filters_applier.rb', line 61

def action
  return :skip unless applied?

  # The highest priority is on a potential backoff from any of the filters because it is
  # the less risky (delay and continue later)
  return :pause if applied.any? { |filter| filter.action == :pause }

  # If none of the filters wanted to pause, we can check for any that would want to seek
  # and if there is any, we can go with this strategy
  return :seek if applied.any? { |filter| filter.action == :seek }

  :skip
end

#applied? ⇒ Boolean

Returns did we filter out any messages during filtering run.

Returns:

  • (Boolean) —

    did we filter out any messages during filtering run



54
55
56
57
58
 
# File 'lib/karafka/pro/processing/filters_applier.rb', line 54

def applied?
  return false unless active?

  !applied.empty?
end

#apply!(messages) ⇒ Object

Parameters:



47
48
49
50
51
 
# File 'lib/karafka/pro/processing/filters_applier.rb', line 47

def apply!(messages)
  return unless active?

  @filters.each { |filter| filter.apply!(messages) }
end

#cursor ⇒ Karafka::Messages::Message?

Note:

Cursor message can also return the offset in the time format

The first message we do need to get next time we poll. We use the minimum not to jump accidentally by over any.

Returns:



85
86
87
88
89
 
# File 'lib/karafka/pro/processing/filters_applier.rb', line 85

def cursor
  return nil unless active?

  applied.map(&:cursor).compact.min_by(&:offset)
end

#timeout ⇒ Integer

Returns minimum timeout we need to pause. This is the minimum for all the filters to satisfy all of them.

Returns:

  • (Integer) —

    minimum timeout we need to pause. This is the minimum for all the filters to satisfy all of them.



77
78
79
 
# File 'lib/karafka/pro/processing/filters_applier.rb', line 77

def timeout
  applied.map(&:timeout).compact.min || 0
end