Class: Karafka::Instrumentation::AssignmentsTracker

Inherits:
Object
  • Object
show all
Includes:
Singleton
Defined in:
lib/karafka/instrumentation/assignments_tracker.rb

Overview

Keeps track of active assignments and materializes them by returning the routing topics with appropriate partitions that are assigned at a given moment

It is auto-subscribed as part of Karafka itself.

It is not heavy from the computational point of view, as it only operates during rebalances.

We keep assignments as flat topics structure because we can go from topics to both subscription and consumer groups if needed.

Constant Summary collapse

GENERATIONS_TOPICS_LIMIT =

Upper bound on the number of distinct topics tracked in @generations. Generations are intentionally retained after revocation (so they can keep incrementing across reassignments), which means the map only ever grows. With static topologies that is bounded by the routing topic count, but pattern subscriptions inject a fresh, permanent Routing::Topic per discovered name, so without a bound the map (and the frozen deep copy built in #generations) would grow for the whole process lifetime. When the bound is exceeded we evict the least-recently-assigned topics. It is high enough never to be reached by realistic static or even large dynamic topologies.

100_000

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ AssignmentsTracker

Initializes the assignments tracker with empty assignments



50
51
52
53
54
 
# File 'lib/karafka/instrumentation/assignments_tracker.rb', line 50

def initialize
  @mutex = Mutex.new
  @assignments = Hash.new { |hash, key| hash[key] = [] }
  @generations = Hash.new { |h, k| h[k] = {} }
end

Class Method Details

.current ⇒ Hash{Karafka::Routing::Topic => Array<Integer>}

Returns:

See Also:



30
31
32
 
# File 'lib/karafka/instrumentation/assignments_tracker.rb', line 30

def current
  instance.current
end

.generation(topic, partition) ⇒ Integer

Parameters:

Returns:

  • (Integer)

See Also:



44
45
46
 
# File 'lib/karafka/instrumentation/assignments_tracker.rb', line 44

def generation(topic, partition)
  instance.generation(topic, partition)
end

.generations ⇒ Hash{Karafka::Routing::Topic => Hash{Integer => Integer}}

Returns:

See Also:



36
37
38
 
# File 'lib/karafka/instrumentation/assignments_tracker.rb', line 36

def generations
  instance.generations
end

Instance Method Details

#clear ⇒ Object

Clears all the assignments and generations



115
116
117
118
119
120
 
# File 'lib/karafka/instrumentation/assignments_tracker.rb', line 115

def clear
  @mutex.synchronize do
    @assignments.clear
    @generations.clear
  end
end

#current ⇒ Hash{Karafka::Routing::Topic => Array<Integer>}

Note:

Keep in mind, that those assignments can change any time, especially when working with multiple consumer groups or subscription groups.

Note:

We return a copy because we modify internals and we do not want user to tamper with the data accidentally

Returns all the active/current assignments of this given process

Returns:



65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/karafka/instrumentation/assignments_tracker.rb', line 65

def current
  assignments = {}

  # Since the `@assignments` state can change during a rebalance, if we would iterate over
  # it exactly during state change, we would end up with the following error:
  #   RuntimeError: can't add a new key into hash during iteration
  @mutex.synchronize do
    @assignments.each do |topic, partitions|
      assignments[topic] = partitions.dup.freeze
    end
  end

  assignments.freeze
end

#generation(topic, partition) ⇒ Integer

Returns the generation count for a specific topic-partition

Parameters:

Returns:

  • (Integer) —

    generation count (0 if never assigned, 1+ otherwise)



108
109
110
111
112
 
# File 'lib/karafka/instrumentation/assignments_tracker.rb', line 108

def generation(topic, partition)
  @mutex.synchronize do
    @generations.dig(topic, partition) || 0
  end
end

#generations ⇒ Hash{Karafka::Routing::Topic => Hash{Integer => Integer}}

Note:

Returns a frozen deep copy to prevent external mutation

Note:

The map is bounded to GENERATIONS_TOPICS_LIMIT topics; once exceeded, the least-recently-assigned topics are evicted. This only ever matters for very large dynamic (pattern-subscribed) topologies - static ones never reach the bound.

Returns the generation counts for all partitions that have ever been assigned

Returns:

  • (Hash{Karafka::Routing::Topic => Hash{Integer => Integer}}) —

    topic to partition generation mapping. Generation starts at 1 on first assignment and increments on each reassignment. Revoked partitions remain in the hash with their last generation value.



91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/karafka/instrumentation/assignments_tracker.rb', line 91

def generations
  result = {}

  @mutex.synchronize do
    @generations.each do |topic, partitions|
      result[topic] = partitions.dup.freeze
    end
  end

  result.freeze
end

#inspect ⇒ String

Returns thread-safe and lock-safe inspect implementation.

Returns:

  • (String) —

    thread-safe and lock-safe inspect implementation



123
124
125
126
127
128
129
130
131
132
133
134
135
136
 
# File 'lib/karafka/instrumentation/assignments_tracker.rb', line 123

def inspect
  info = if @mutex.try_lock
    begin
      assignments = @assignments.dup.transform_keys(&:name).inspect
      "assignments=#{assignments}"
    ensure
      @mutex.unlock
    end
  else
    "busy"
  end

  "#<#{self.class.name} #{info}>"
end

#on_client_events_poll(event) ⇒ Object

Note:

We can run the #assignment_lost? on each events poll because they happen once every 5 seconds during processing plus prior to each messages poll. It takes 0.6 microseconds per call.

Handles events_poll notification to detect assignment loss This is called regularly (every tick_interval) so we check if assignment was lost

Parameters:

  • event (Karafka::Core::Monitoring::Event) 


158
159
160
161
162
163
164
165
166
 
# File 'lib/karafka/instrumentation/assignments_tracker.rb', line 158

def on_client_events_poll(event)
  client = event[:caller]

  # Only clear assignments if they were actually lost
  return unless client.assignment_lost?

  # Cleaning happens the same way as with the consumer reset
  on_client_reset(event)
end

#on_client_reset(event) ⇒ Object

When client is under reset due to critical issues, remove all of its assignments as we will get a new set of assignments

Parameters:

  • event (Karafka::Core::Monitoring::Event) 


141
142
143
144
145
146
147
148
149
 
# File 'lib/karafka/instrumentation/assignments_tracker.rb', line 141

def on_client_reset(event)
  sg = event[:subscription_group]

  @mutex.synchronize do
    @assignments.delete_if do |topic, _partitions|
      topic.subscription_group.id == sg.id
    end
  end
end

#on_rebalance_partitions_assigned(event) ⇒ Object

Adds partitions to the current assignments hash

Parameters:

  • event (Karafka::Core::Monitoring::Event) 


189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
 
# File 'lib/karafka/instrumentation/assignments_tracker.rb', line 189

def on_rebalance_partitions_assigned(event)
  sg = event[:subscription_group]

  @mutex.synchronize do
    event[:tpl].to_h.each do |topic, partitions|
      topic = sg.topics.find(topic)

      partition_ids = []

      partitions.each do |partition|
        partition_id = partition.partition
        partition_ids << partition_id
        @generations[topic][partition_id] ||= 0
        @generations[topic][partition_id] += 1
      end

      # Move this topic to the end of the insertion-ordered map so it counts as
      # most-recently-assigned for the bounded eviction below (keeps actively reassigned
      # topics, drops genuinely stale ones first).
      @generations[topic] = @generations.delete(topic)

      @assignments[topic] += partition_ids
      @assignments[topic].sort!
    end

    evict_stale_generations
  end
end

#on_rebalance_partitions_revoked(event) ⇒ Object

Removes partitions from the current assignments hash

Parameters:

  • event (Karafka::Core::Monitoring::Event) 


171
172
173
174
175
176
177
178
179
180
181
182
183
184
 
# File 'lib/karafka/instrumentation/assignments_tracker.rb', line 171

def on_rebalance_partitions_revoked(event)
  sg = event[:subscription_group]

  @mutex.synchronize do
    event[:tpl].to_h.each do |topic, partitions|
      topic = sg.topics.find(topic)

      @assignments[topic] -= partitions.map(&:partition)
      @assignments[topic].sort!
      # Remove completely topics for which we do not have any assignments left
      @assignments.delete_if { |_topic, cur_partitions| cur_partitions.empty? }
    end
  end
end