Sha256: 614aaf849066de5ecd21fb14731c2698bc8135aaf7c118ad59491f75a8f91c6b

Contents?: true

Size: 1020 Bytes

Versions: 8

Compression:

Stored size: 1020 Bytes

Contents

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] the offset of the most recent message in the partition.
    attr_reader :highwater_mark_offset

    # @return [Array<Kafka::FetchedMessage>]
    attr_reader :messages

    def initialize(topic:, partition:, highwater_mark_offset:, messages:)
      @topic = topic
      @partition = partition
      @highwater_mark_offset = highwater_mark_offset
      @messages = messages
    end

    def empty?
      @messages.empty?
    end

    def first_offset
      if empty?
        nil
      else
        messages.first.offset
      end
    end

    def last_offset
      if empty?
        highwater_mark_offset - 1
      else
        messages.last.offset
      end
    end

    def offset_lag
      if empty?
        0
      else
        highwater_mark_offset - last_offset
      end
    end
  end
end

Version data entries

8 entries across 8 versions & 1 rubygems

Version Path
ruby-kafka-0.3.16 lib/kafka/fetched_batch.rb
ruby-kafka-0.3.16.beta2 lib/kafka/fetched_batch.rb
ruby-kafka-0.3.16.beta1 lib/kafka/fetched_batch.rb
ruby-kafka-0.3.15 lib/kafka/fetched_batch.rb
ruby-kafka-0.3.15.beta3 lib/kafka/fetched_batch.rb
ruby-kafka-0.3.15.beta2 lib/kafka/fetched_batch.rb
ruby-kafka-0.3.15.beta1 lib/kafka/fetched_batch.rb
ruby-kafka-0.3.14 lib/kafka/fetched_batch.rb