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

Inherits:
Object
  • Object
show all
Defined in:
lib/karafka/pro/processing/coordinators/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



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

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.



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

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



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

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



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

def applied?
  return false unless active?

  !applied.empty?
end

#apply!(messages) ⇒ Object

Parameters:



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

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:



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

def cursor
  return nil unless active?

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

#mark_as_consumed? ⇒ Boolean

Returns did any of the filters requested offset storage during filter application.

Returns:

  • (Boolean) —

    did any of the filters requested offset storage during filter application



94
95
96
97
98
 
# File 'lib/karafka/pro/processing/coordinators/filters_applier.rb', line 94

def mark_as_consumed?
  # We can manage filtering offset only when user wanted that and there is a cursor
  # to use
  applied.any?(&:mark_as_consumed?) && cursor
end

#marking_method ⇒ Symbol

Returns :mark_as_consumed or :mark_as_consumed!.

Returns:

  • (Symbol) —

    :mark_as_consumed or :mark_as_consumed!



101
102
103
104
105
106
107
 
# File 'lib/karafka/pro/processing/coordinators/filters_applier.rb', line 101

def marking_method
  candidates = applied.map(&:marking_method)

  return :mark_as_consumed! if candidates.include?(:mark_as_consumed!)

  :mark_as_consumed
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.



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

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