lib/kafka/cluster.rb
def resolve_offsets(topic, partitions, offset)
      add_target_topics([topic])
      refresh_metadata_if_necessary!

      partitions_by_broker = partitions.each_with_object({}) {|partition, hsh|
        broker = get_leader(topic, partition)

        hsh[broker] ||= []
        hsh[broker] << partition
      }

      if offset == :earliest
        offset = -2
      elsif offset == :latest
        offset = -1
      end

      offsets = {}

      partitions_by_broker.each do |broker, broker_partitions|
        response = broker.list_offsets(
          topics: {
            topic => broker_partitions.map {|partition|
              {
                partition: partition,
                time: offset,
                max_offsets: 1,
              }
            }
          }
        )

        broker_partitions.each do |partition|
          offsets[partition] = response.offset_for(topic, partition)
        end
      end

      offsets
    rescue Kafka::ProtocolError
      mark_as_stale!
      raise
    end
# Initializes a Cluster with a set of seed brokers.
#
# The cluster will try to fetch cluster metadata from one of the brokers.
#
# @param seed_brokers [Array<URI>]
# @param broker_pool [Kafka::BrokerPool]
# @param logger [Logger]
Initializes a Cluster with a set of seed brokers.
The cluster will try to fetch cluster metadata from one of the brokers.
Initializes a Cluster with a set of seed brokers . The cluster will try to fetch cluster metadata from one of the brokers .
def resolve_offsets(topic, partitions, offset)
      add_target_topics([topic])
      refresh_metadata_if_necessary!
      partitions_by_broker = partitions.each_with_object({}) {|partition, hsh|
        broker = get_leader(topic, partition)
        hsh[broker] ||= []
        hsh[broker] << partition
      }
      if offset == :earliest
        offset = -2
      elsif offset == :latest
        offset = -1
      end
      offsets = {}
      partitions_by_broker.each do |broker, broker_partitions|
        response = broker.list_offsets(
          topics: {
            topic => broker_partitions.map {|partition|
              {
                partition: partition,
                time: offset,
                max_offsets: 1,
              }
            }
          }
        )
        broker_partitions.each do |partition|
          offsets[partition] = response.offset_for(topic, partition)
        end
      end
      offsets
    rescue Kafka::ProtocolError
      mark_as_stale!
      raise
    end
def resolve_offsets ( topic , partitions , offset ) add_target_topics ( [ topic ] ) refresh_metadata_if_necessary! partitions_by_broker = partitions . each_with_object ( { } ) { | partition , hsh | broker = get_leader ( topic , partition ) hsh [ broker ] ||= [ ] hsh [ broker ] << partition } if offset == :earliest offset = - 2 elsif offset == :latest offset = - 1 end offsets = { } partitions_by_broker . each do | broker , broker_partitions | response = broker . list_offsets ( topics : { topic => broker_partitions . map { | partition | { partition : partition , time : offset , max_offsets : 1 , } } } ) broker_partitions . each do | partition | offsets [ partition ] = response . offset_for ( topic , partition ) end end offsets rescue Kafka :: ProtocolError mark_as_stale! raise end
Initializes a Cluster with a set of seed brokers.
Initializes a Cluster with a set of seed brokers .
topic
partitions
offset
None