lib/kafka/client.rb
# Enumerate all messages in a topic.
#
# @param topic [String] the topic to consume messages from.
#
# @param start_from_beginning [Boolean] whether to start from the beginning
#   of the topic or just subscribe to new messages being produced.
#
# @param max_wait_time [Integer] the maximum amount of time to wait before
#   the server responds, in seconds.
#
# @param min_bytes [Integer] the minimum number of bytes to wait for. If set to
#   zero, the broker will respond immediately, but the response may be empty.
#   The default is 1 byte, which means that the broker will respond as soon as
#   a message is written to the partition.
#
# @param max_bytes [Integer] the maximum number of bytes to include in the
#   response message set. Default is 1 MB. You need to set this higher if you
#   expect messages to be larger than this.
#
# @return [nil]
Enumerate all messages in a topic.
Enumerate all messages in a topic .
def each_message(topic:, start_from_beginning: true, max_wait_time: 5, min_bytes: 1, max_bytes: 1048576, &block)
      default_offset ||= start_from_beginning ? :earliest : :latest
      offsets = Hash.new { default_offset }
      loop do
        operation = FetchOperation.new(
          cluster: @cluster,
          logger: @logger,
          min_bytes: min_bytes,
          max_wait_time: max_wait_time,
        )
        @cluster.partitions_for(topic).map(&:partition_id).each do |partition|
          partition_offset = offsets[partition]
          operation.fetch_from_partition(topic, partition, offset: partition_offset, max_bytes: max_bytes)
        end
        batches = operation.execute
        batches.each do |batch|
          batch.messages.each(&block)
          offsets[batch.partition] = batch.last_offset + 1 unless batch.unknown_last_offset?
        end
      end
    end
def each_message ( topic : , start_from_beginning : true , max_wait_time : 5 , min_bytes : 1 , max_bytes : 1048576 , & block ) default_offset ||= start_from_beginning ? :earliest : :latest offsets = Hash . new { default_offset } loop do operation = FetchOperation . new ( cluster : @cluster , logger : @logger , min_bytes : min_bytes , max_wait_time : max_wait_time , ) @cluster . partitions_for ( topic ) . map ( & :partition_id ) . each do | partition | partition_offset = offsets [ partition ] operation . fetch_from_partition ( topic , partition , offset : partition_offset , max_bytes : max_bytes ) end batches = operation . execute batches . each do | batch | batch . messages . each ( & block ) offsets [ batch . partition ] = batch . last_offset + 1 unless batch . unknown_last_offset? end end end
Enumerate all messages in a topic.
Enumerate all messages in a topic .
None
nil
topic
the topic to consume messages from.
the topic to consume messages from .
start_from_beginning
whether to start from the beginning of the topic or just subscribe to new messages being produced.
whether to start from the beginning of the topic or just subscribe to new messages being produced .
max_wait_time
the maximum amount of time to wait before the server responds, in seconds.
the maximum amount of time to wait before the server responds in seconds .
min_bytes
the minimum number of bytes to wait for. If set to zero, the broker will respond immediately, but the response may be empty. The default is 1 byte, which means that the broker will respond as soon as a message is written to the partition.
the minimum number of bytes to wait for . If set to zero the broker will respond immediately but the response may be empty . The default is 1 byte which means that the broker will respond as soon as a message is written to the partition .
max_bytes
the maximum number of bytes to include in the response message set. Default is 1 MB. You need to set this higher if you expect messages to be larger than this.
the maximum number of bytes to include in the response message set . Default is 1 MB . You need to set this higher if you expect messages to be larger than this .
block
<not_specific>