async fn find_leader_for_topic_partition(
        &mut self,
        topic: &str,
        partition: i32,
    ) -> Result<Self::Leader, ClientError> {

        
        let kf_metadata = self.query_metadata(Some(vec![topic.to_owned()])).await?;
        let brokers = &kf_metadata.brokers;
        let topics = kf_metadata.topics;
        
        for response_topic in topics {
            
            if response_topic.name == topic {
                for response_partition in response_topic.partitions {
                    if response_partition.partition_index == partition {
                        let leader_id = response_partition.leader_id;

                        // traverse brokers and find leader
                        for broker in brokers {
                            if broker.node_id == leader_id {
                                debug!("broker {}/{} is leader", broker.host, broker.port);
                                let config = LeaderConfig::new(broker.into(),topic.to_owned(),partition)
                                        .spu_id(leader_id)
                                        .client_id(self.0.client_id());
                                return KfLeader::connect(config).await
                            }
                        }
                    }
                }
            }
            
        }
        
    
        Err(ClientError::IoError(IoError::new(
            ErrorKind::Other,
            format!(
                "topic '{}/{}': unknown topic or partition",
                topic, partition
            ),
        )))
        
    }
/// Find address of the Broker leader for a topic/partition
Find address of the Broker leader for a topic/partition
Find address of the Broker leader for a topic / partition
async fn find_leader_for_topic_partition(
        &mut self,
        topic: &str,
        partition: i32,
    ) -> Result<Self::Leader, ClientError> {
        let kf_metadata = self.query_metadata(Some(vec![topic.to_owned()])).await?;
        let brokers = &kf_metadata.brokers;
        let topics = kf_metadata.topics;
        for response_topic in topics {
            if response_topic.name == topic {
                for response_partition in response_topic.partitions {
                    if response_partition.partition_index == partition {
                        let leader_id = response_partition.leader_id;
                        // traverse brokers and find leader
                        for broker in brokers {
                            if broker.node_id == leader_id {
                                debug!("broker {}/{} is leader", broker.host, broker.port);
                                let config = LeaderConfig::new(broker.into(),topic.to_owned(),partition)
                                        .spu_id(leader_id)
                                        .client_id(self.0.client_id());
                                return KfLeader::connect(config).await
                            }
                        }
                    }
                }
            }
        }
        Err(ClientError::IoError(IoError::new(
            ErrorKind::Other,
            format!(
                "topic '{}/{}': unknown topic or partition",
                topic, partition
            ),
        )))
    }
async fn find_leader_for_topic_partition ( & mut self , topic : & str , partition : i32 , ) -> Result < Self :: Leader , ClientError > { let kf_metadata = self . query_metadata ( Some ( vec ! [ topic to_owned ( ) ] ) ) . await ? ; let brokers = & kf_metadata . brokers ; let topics = kf_metadata . topics ; for response_topic in topics { if response_topic . name == topic { for response_partition in response_topic . partitions { if response_partition . partition_index == partition { let leader_id = response_partition . leader_id ; // traverse brokers and find leader for broker in brokers { if broker . node_id == leader_id { debug ! ( " " broker host broker port ) ; let config = LeaderConfig :: new ( broker . into ( ) , topic . to_owned ( ) , partition ) . spu_id ( leader_id ) . client_id ( self . 0 . client_id ( ) ) ; return KfLeader :: connect ( config ) . await } } } } } } Err ( ClientError :: IoError ( IoError :: new ( ErrorKind :: Other , format ! ( " " topic partition ) , ) ) ) }
Find address of the Broker leader for a topic/partition
Find address of the Broker leader for a topic / partition
&mut self
topic
&str
partition
i32
&mut self
topic
&str
partition
i32