Class: Karafka::Pro::Instrumentation::ConsumerGroups::LagCompensation::Fetcher

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

Overview

Fetches the end offsets of given partitions via the client own connection: no dedicated instances and no extra Kafka connections are ever created.

All partitions are resolved with one batched request that librdkafka fans out to the involved partition leaders internally, so the cost does not grow with the number of paused partitions.

End offsets are the only thing the compensation needs from the broker: committed and stored offsets of paused partitions are maintained by the client commits (not by fetches), so their statistics values stay accurate while paused and lags can be derived from them at compensation time.

The consumer isolation level is forwarded to the query, so on topics without in-flight transactions the end offset matches the reference the native consumer lag derives from.

Known edge case (transactional topics): the batched ListOffsets this uses resolves :latest to the high watermark regardless of the forwarded isolation level, unlike the per-partition query_watermark_offsets it replaced, which returned the last stable offset for a read_committed consumer. So while a transaction is IN FLIGHT on a paused partition, the compensated end offset (and the lag derived from it) reflects the high watermark and can overstate the read_committed lag by the number of uncommitted messages. It self-corrects once the transaction commits or aborts and the last stable offset advances. Non-transactional topics are unaffected (LSO == HWM there).

Instance Method Summary collapse

Instance Method Details

#call(client, paused) ⇒ Hash{String => Hash{Integer => Integer}}

Returns end offsets of the requested partitions.

Parameters:

Returns:

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

    end offsets of the requested partitions



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/karafka/pro/instrumentation/consumer_groups/lag_compensation/fetcher.rb', line 65

def call(client, paused)
  request = paused.transform_values do |partitions|
    partitions.map { |partition| { partition: partition, offset: :latest } }
  end

  return {} if request.empty?

  # Auto-initializes the per-topic partitions hash on first access, same idiom as
  # the refresher state. Only ever written to and iterated here, never read by a
  # missing key, so the default block cannot insert a spurious entry.
  data = Hash.new { |topics, topic| topics[topic] = {} }

  client.read_partition_offsets(request).each do |result|
    data[result[:topic]][result[:partition]] = result[:offset]
  end

  data
end