Class: Karafka::Pro::Instrumentation::ConsumerGroups::LagCompensation::Registry

Inherits:
Object
  • Object
show all
Includes:
Singleton
Defined in:
lib/karafka/pro/instrumentation/consumer_groups/lag_compensation/registry.rb

Overview

Thread-safe storage for actively refreshed watermarks and lags of long-paused partitions. Written by the refresher on listener threads, read by the statistics decorator on the librdkafka callbacks thread.

No age based expiry is needed: entries are refreshed while a partition stays paused, kept (but no longer refreshed) once it resumes so the compensator can hand over to the live statistics, and dropped in bulk on rebalances. Within an assignment the stored set is therefore bounded by the partitions the client paused.

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Registry

Returns a new instance of Registry.



52
53
54
55
 
# File 'lib/karafka/pro/instrumentation/consumer_groups/lag_compensation/registry.rb', line 52

def initialize
  @mutex = Mutex.new
  @clients = {}
end

Instance Method Details

#evict(client_name) ⇒ Object

Removes all the data of a given client. Used on rebalances as partitions may no longer belong to the client that refreshed them.

Parameters:

  • client_name (String) —

    rdkafka client name



92
93
94
95
96
 
# File 'lib/karafka/pro/instrumentation/consumer_groups/lag_compensation/registry.rb', line 92

def evict(client_name)
  @mutex.synchronize do
    @clients.delete(client_name)
  end
end

#fetch(client_name) ⇒ Hash?

Returns refreshed data of a given client or nil when none.

Parameters:

  • client_name (String) —

    rdkafka client name

Returns:

  • (Hash, nil) —

    refreshed data of a given client or nil when none



82
83
84
85
86
 
# File 'lib/karafka/pro/instrumentation/consumer_groups/lag_compensation/registry.rb', line 82

def fetch(client_name)
  @mutex.synchronize do
    @clients[client_name]
  end
end

#update(client_name, data) ⇒ Object

Merges the freshly refreshed offsets into the stored data of a client, overwriting the values of the refreshed partitions and keeping the previously stored ones.

A refresh covers only the currently long-paused partitions, so a partition that resumed is absent here and its last value is kept rather than dropped: the compensator keeps overlaying it (its guard applies it only while still fresher than the live statistics) until librdkafka post-resume fetches catch up, avoiding a snap back to the frozen pre-resume values in the meantime. Stale entries are dropped in bulk on the next rebalance.

Parameters:

  • client_name (String) —

    rdkafka client name (matches statistics name field)

  • data (Hash{String => Hash{Integer => Integer}}) —

    topics with partitions with their end offsets



70
71
72
73
74
75
76
77
78
 
# File 'lib/karafka/pro/instrumentation/consumer_groups/lag_compensation/registry.rb', line 70

def update(client_name, data)
  @mutex.synchronize do
    stored = (@clients[client_name] ||= {})

    data.each do |topic, partitions|
      (stored[topic] ||= {}).merge!(partitions)
    end
  end
end