lib/kafka/cluster.rb
# Fetches the cluster metadata.
#
# This is used to update the partition leadership information, among other things.
# The methods will go through each node listed in `seed_brokers`, connecting to the
# first one that is available. This node will be queried for the cluster metadata.
#
# @raise [ConnectionError] if none of the nodes in `seed_brokers` are available.
# @return [Protocol::MetadataResponse] the cluster metadata.
Fetches the cluster metadata.
This is used to update the partition leadership information, among other things.
The methods will go through each node listed in `seed_brokers`, connecting to the
first one that is available. This node will be queried for the cluster metadata.
Fetches the cluster metadata . This is used to update the partition leadership information among other things . The methods will go through each node listed in ` seed_brokers ` connecting to the first one that is available . This node will be queried for the cluster metadata .
def fetch_cluster_info
      errors = []
      @seed_brokers.shuffle.each do |node|
        @logger.info "Fetching cluster metadata from #{node}"
        begin
          broker = @broker_pool.connect(node.hostname, node.port)
          cluster_info = broker.fetch_metadata(topics: @target_topics)
          @stale = false
          @logger.info "Discovered cluster metadata; nodes: #{cluster_info.brokers.join(', ')}"
          return cluster_info
        rescue Error => e
          @logger.error "Failed to fetch metadata from #{node}: #{e}"
          errors << [node, e]
        ensure
          broker.disconnect unless broker.nil?
        end
      end
      error_description = errors.map {|node, exception| "- #{node}: #{exception}" }.join("\n")
      raise ConnectionError, "Could not connect to any of the seed brokers:\n#{error_description}"
    end
def fetch_cluster_info errors = [ ] @seed_brokers . shuffle . each do | node | @logger . info "Fetching cluster metadata from #{node}" begin broker = @broker_pool . connect ( node . hostname , node . port ) cluster_info = broker . fetch_metadata ( topics : @target_topics ) @stale = false @logger . info "Discovered cluster metadata; nodes: #{cluster_info.brokers.join(', ')}" return cluster_info rescue Error => e @logger . error "Failed to fetch metadata from #{node}: #{e}" errors << [ node , e ] ensure broker . disconnect unless broker . nil? end end error_description = errors . map { | node , exception | "- #{node}: #{exception}" } . join ( "\n" ) raise ConnectionError , "Could not connect to any of the seed brokers:\n#{error_description}" end
Fetches the cluster metadata.
Fetches the cluster metadata .
the cluster metadata.
the cluster metadata .
Protocol::MetadataResponse
[ConnectionError] if none of the nodes in `seed_brokers` are available.
[ ConnectionError ] if none of the nodes in ` seed_brokers ` are available .