Class: Kafka::Protocol::MetadataResponse

Inherits:
Object
  • Object
show all
Defined in:
lib/kafka/protocol/metadata_response.rb

Overview

A response to a MetadataRequest.

The response contains information on the brokers, topics, and partitions in the cluster.

  • For each broker a node id, host, and port is provided.
  • For each topic partition the node id of the broker acting as partition leader, as well as a list of node ids for the set of replicas, are given. The isr list is the subset of replicas that are "in sync", i.e. have fully caught up with the leader.

API Specification

MetadataResponse => [Broker][TopicMetadata] Broker => NodeId Host Port (any number of brokers may be returned) NodeId => int32 Host => string Port => int32 TopicMetadata => TopicErrorCode TopicName [PartitionMetadata] TopicErrorCode => int16 PartitionMetadata => PartitionErrorCode PartitionId Leader Replicas Isr PartitionErrorCode => int16 PartitionId => int32 Leader => int32 Replicas => [int32] Isr => [int32]

Defined Under Namespace

Classes: PartitionMetadata, TopicMetadata

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(brokers:, controller_id:, topics:) ⇒ MetadataResponse

Returns a new instance of MetadataResponse.

 75 76 77 78 79
 
# File 'lib/kafka/protocol/metadata_response.rb', line 75 def initialize(brokers:, controller_id:, topics:) @brokers = brokers @controller_id = controller_id @topics = topics end

Instance Attribute Details

#brokers ⇒ Array<Kafka::BrokerInfo> (readonly)

Returns the list of brokers in the cluster.

Returns:

 67 68 69
 
# File 'lib/kafka/protocol/metadata_response.rb', line 67 def brokers @brokers end

#controller_id ⇒ Integer (readonly)

Returns The broker id of the controller broker.

Returns:

  • (Integer) —

    The broker id of the controller broker.

 73 74 75
 
# File 'lib/kafka/protocol/metadata_response.rb', line 73 def controller_id @controller_id end

#topics ⇒ Array<TopicMetadata> (readonly)

Returns the list of topics in the cluster.

Returns:

 70 71 72
 
# File 'lib/kafka/protocol/metadata_response.rb', line 70 def topics @topics end

Class Method Details

.decode(decoder) ⇒ MetadataResponse

Decodes a MetadataResponse from a Decoder containing response data.

Parameters:

Returns:

 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182
 
# File 'lib/kafka/protocol/metadata_response.rb', line 143 def self.decode(decoder) brokers = decoder.array do node_id = decoder.int32 host = decoder.string port = decoder.int32 _rack = decoder.string BrokerInfo.new( node_id: node_id, host: host, port: port ) end controller_id = decoder.int32 topics = decoder.array do topic_error_code = decoder.int16 topic_name = decoder.string _is_internal = decoder.boolean partitions = decoder.array do PartitionMetadata.new( partition_error_code: decoder.int16, partition_id: decoder.int32, leader: decoder.int32, replicas: decoder.array { decoder.int32 }, isr: decoder.array { decoder.int32 }, ) end TopicMetadata.new( topic_error_code: topic_error_code, topic_name: topic_name, partitions: partitions, ) end new(brokers: brokers, controller_id: controller_id, topics: topics) end

Instance Method Details

#controller_broker ⇒ Object 

 123 124 125
 
# File 'lib/kafka/protocol/metadata_response.rb', line 123 def controller_broker find_broker(controller_id) end

#find_broker(node_id) ⇒ Kafka::BrokerInfo

Finds the broker info for the given node id.

Parameters:

  • node_id (Integer) —

    the node id of the broker.

Returns:

Raises:

 115 116 117 118 119 120 121
 
# File 'lib/kafka/protocol/metadata_response.rb', line 115 def find_broker(node_id) broker = @brokers.find {|b| b.node_id == node_id } raise Kafka::NoSuchBroker, "No broker with id #{node_id}" if broker.nil? broker end

#find_leader_id(topic, partition) ⇒ Integer

Finds the node id of the broker that is acting as leader for the given topic and partition per this metadata.

Parameters:

  • topic (String) —

    the name of the topic.

  • partition (Integer) —

    the partition number.

Returns:

  • (Integer) —

    the node id of the leader.

 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109
 
# File 'lib/kafka/protocol/metadata_response.rb', line 87 def find_leader_id(topic, partition) topic_info = @topics.find {|t| t.topic_name == topic } if topic_info.nil? raise UnknownTopicOrPartition, "no topic #{topic}" end Protocol.handle_error(topic_info.topic_error_code) partition_info = topic_info.partitions.find {|p| p.partition_id == partition } if partition_info.nil? raise UnknownTopicOrPartition, "no partition #{partition} in topic #{topic}" end begin Protocol.handle_error(partition_info.partition_error_code) rescue ReplicaNotAvailable # This error can be safely ignored per the protocol specification.  end partition_info.leader end

#partitions_for(topic_name) ⇒ Object 

 127 128 129 130 131 132 133 134 135 136 137
 
# File 'lib/kafka/protocol/metadata_response.rb', line 127 def partitions_for(topic_name) topic = @topics.find {|t| t.topic_name == topic_name } if topic.nil? raise UnknownTopicOrPartition, "unknown topic #{topic_name}" end Protocol.handle_error(topic.topic_error_code) topic.partitions end