Class: Karafka::Pro::Instrumentation::ConsumerGroups::LagCompensation::Compensator

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

Overview

Compensates stale offsets and lags of long-paused partitions in the raw librdkafka statistics, using the actively refreshed values from the registry. When there is no refreshed data for a given client, topic or partition, it does nothing, so values from librdkafka remain untouched.

It is pure and instant: all the broker queries happen in the refresher on the listener threads, never on the librdkafka callbacks path.

Instance Method Summary collapse

Instance Method Details

#call(statistics) ⇒ Object

Compensates the statistics in place, so the standard decoration (deltas, freeze durations) that runs afterwards operates on the compensated values as well

Parameters:

  • statistics (Hash) —

    raw librdkafka statistics



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/karafka/pro/instrumentation/consumer_groups/lag_compensation/compensator.rb', line 48

def call(statistics)
  data = Registry.instance.fetch(statistics["name"])

  return unless data

  topics = statistics["topics"]

  return unless topics

  data.each do |topic, partitions|
    t_stats = topics[topic]

    next unless t_stats

    partitions.each do |partition, end_offset|
      p_stats = t_stats["partitions"][partition.to_s]

      next unless p_stats

      compensate(p_stats, end_offset)
    end
  end
end