Sha256: 9d1df22857bc8ae23fc7f40f36ffd2c622902210985a6aa9f4d33be3da34f4b3
Contents?: true
Size: 1.07 KB
Versions: 15
Compression:
Stored size: 1.07 KB
Contents
# frozen_string_literal: true
module Kafka
# An ordered sequence of messages fetched from a Kafka partition.
class FetchedBatch
# @return [String]
attr_reader :topic
# @return [Integer]
attr_reader :partition
# @return [Integer]
attr_reader :last_offset
# @return [Integer] the offset of the most recent message in the partition.
attr_reader :highwater_mark_offset
# @return [Array<Kafka::FetchedMessage>]
attr_accessor :messages
def initialize(topic:, partition:, highwater_mark_offset:, messages:, last_offset: nil)
@topic = topic
@partition = partition
@highwater_mark_offset = highwater_mark_offset
@messages = messages
@last_offset = last_offset
end
def empty?
@messages.empty?
end
def unknown_last_offset?
@last_offset.nil?
end
def first_offset
if empty?
nil
else
messages.first.offset
end
end
def offset_lag
if empty?
0
else
(highwater_mark_offset - 1) - last_offset
end
end
end
end
Version data entries
15 entries across 15 versions & 2 rubygems