Class: Karafka::Pro::Instrumentation::ConsumerGroups::LagCompensation::Refresher

Inherits:
Object
  • Object
show all
Includes:
Core::Helpers::Time
Defined in:
lib/karafka/pro/instrumentation/consumer_groups/lag_compensation/refresher.rb

Overview

Refreshes offsets and lags data of partitions that stayed paused for at least the configured pause age, at most once per interval.

Runs on the listener threads via the client.events_poll event, so the fetching happens on the thread that owns the client connection within its group identity. Pause state is tracked via the client.pause and client.resume instrumentation events, which fire on the same thread.

On resume or revocation the partition data is dropped immediately, so stale values never overlay live statistics and revoked partitions are never queried again. Errors never propagate and are only instrumented: the next attempt happens on the next interval anyway.

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Refresher

Returns a new instance of Refresher.



56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/karafka/pro/instrumentation/consumer_groups/lag_compensation/refresher.rb', line 56

def initialize
  @fetcher = Fetcher.new
  @mutex = Mutex.new
  @states = Hash.new do |states, subscription_group_id|
    states[subscription_group_id] = {
      # Gates the refreshing to run at most once per interval
      runner: Helpers::IntervalRunner.new(interval: interval) do |client, state|
        refresh_tick(client, state)
      end,
      paused: Hash.new { |paused, topic| paused[topic] = {} },
      client_name: nil
    }
  end
end

Instance Method Details

#on_client_events_poll(event) ⇒ Object

Once per interval refreshes the data of all the partitions that stayed paused for at least the configured pause age

Parameters:

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

    event with the client



104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/karafka/pro/instrumentation/consumer_groups/lag_compensation/refresher.rb', line 104

def on_client_events_poll(event)
  # Never run during shutdown or quieting: blocking broker queries at that time
  # would eat into the shutdown time budget for no benefit
  return if Karafka::App.done?

  state = state_for(event[:subscription_group].id)

  state[:runner].call(event[:caller], state)
rescue => e
  # Refreshing is a best-effort operation: values stay stale until the next
  # successful run on one of the following intervals. Errors are still
  # instrumented, so persistent failures (missing ACLs, removed topics, etc)
  # remain observable
  monitor.instrument(
    "error.occurred",
    caller: self,
    error: e,
    type: "lag_compensation.refresher.error"
  )
end

#on_client_pause(event) ⇒ Object

Tracks when a given partition got paused

Parameters:

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

    pause event



74
75
76
77
78
79
 
# File 'lib/karafka/pro/instrumentation/consumer_groups/lag_compensation/refresher.rb', line 74

def on_client_pause(event)
  state = state_for(event[:subscription_group].id)

  state[:paused][event[:topic]][event[:partition]] = monotonic_now
  state[:client_name] = event[:caller].name
end

#on_client_resume(event) ⇒ Object

Stops refreshing a resumed partition but keeps its last refreshed values in the registry. The compensator keeps overlaying them (its guard applies them only while still fresher than the live statistics), so the emitted statistics do not snap back to the frozen pre-resume values during the gap before librdkafka post-resume fetches update the offsets. Once the live values catch up the compensator stops using them, and the kept entries are dropped in bulk on the next rebalance.

Parameters:

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

    resume event



89
90
91
92
93
94
95
96
97
98
 
# File 'lib/karafka/pro/instrumentation/consumer_groups/lag_compensation/refresher.rb', line 89

def on_client_resume(event)
  state = state_for(event[:subscription_group].id)
  # Fetch with a fallback not to auto-build tracking of an untracked topic
  partitions = state[:paused].fetch(event[:topic], nil)

  return unless partitions
  return unless partitions.delete(event[:partition])

  state[:paused].delete(event[:topic]) if partitions.empty?
end

#on_rebalance_partitions_revoked(event) ⇒ Object

Expires all the refreshed data and pause tracking of a client on a rebalance. Its paused partitions may no longer belong to it and tracking will refill only via real pause events of partitions this client still owns.

Parameters:

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

    revocation event



130
131
132
133
134
135
136
137
138
 
# File 'lib/karafka/pro/instrumentation/consumer_groups/lag_compensation/refresher.rb', line 130

def on_rebalance_partitions_revoked(event)
  state = state_for(event[:subscription_group].id)
  # Clear (instead of a reassignment) preserves the auto-building behavior
  state[:paused].clear

  client_name = state[:client_name]

  Registry.instance.evict(client_name) if client_name
end