/// Find address of the SPU leader for a topic/partition
Find address of the SPU leader for a topic/partition
Find address of the SPU leader for a topic / partition
async fn find_leader_for_topic_partition(
        &mut self,
        topic: &str,
        partition: i32,
    ) -> Result<Self::Leader, ClientError> {
        let topic_comp_resp = self.get_topic_composition(topic).await?;
        let mut topics_resp = topic_comp_resp.topics;
        let spus_resp = topic_comp_resp.spus;
        // there must be one topic in reply
        if topics_resp.len() != 1 {
            return Err(ClientError::IoError(IoError::new(
                ErrorKind::InvalidData,
                format!("topic error: expected 1 topic, found {}", topics_resp.len()),
            )));
        }
        // check for errors
        let topic_resp = topics_resp.remove(0);
        if topic_resp.error_code != FlvErrorCode::None {
            return Err(ClientError::IoError(IoError::new(
                ErrorKind::InvalidData,
                format!("topic error: {}", topic_resp.error_code.to_sentence()),
            )));
        }
        // lookup leader
        for partition_resp in topic_resp.partitions {
            if partition_resp.partition_idx == partition {
                // check for errors
                if partition_resp.error_code != FlvErrorCode::None {
                    return Err(ClientError::IoError(IoError::new(
                        ErrorKind::InvalidData,
                        format!(
                            "topic-composition partition error: {}",
                            topic_resp.error_code.to_sentence()
                        ),
                    )));
                }
                // traverse spus and find leader
                let leader_id = partition_resp.leader_id;
                for spu_resp in &spus_resp {
                    if spu_resp.spu_id == leader_id {
                        // check for errors
                        if spu_resp.error_code != FlvErrorCode::None {
                            return Err(ClientError::IoError(IoError::new(
                                ErrorKind::InvalidData,
                                format!(
                                    "topic-composition spu error: {}",
                                    topic_resp.error_code.to_sentence()
                                ),
                            )));
                        }
                        debug!("spu {}/{}: is leader", spu_resp.host, spu_resp.port);
                        let leader_config =
                            LeaderConfig::new(spu_resp.into(), topic.to_owned(), partition)
                                .spu_id(leader_id)
                                .client_id(self.0.client_id());
                        return SpuLeader::connect(leader_config).await;
                    }
                }
            }
        }
        Err(ClientError::IoError(IoError::new(
            ErrorKind::Other,
            format!(
                "topic-composition '{}/{}': unknown topic or partition",
                topic, partition
            ),
        )))
    }
async fn find_leader_for_topic_partition ( & mut self , topic : & str , partition : i32 , ) -> Result < Self :: Leader , ClientError > { let topic_comp_resp = self . get_topic_composition ( topic ) . await ? ; let mut topics_resp = topic_comp_resp . topics ; let spus_resp = topic_comp_resp . spus ; // there must be one topic in reply if topics_resp . len ( ) != 1 { return Err ( ClientError :: IoError ( IoError :: new ( ErrorKind :: InvalidData , format ! ( " " topics_resp len ( ) ) , ) ) ) ; } // check for errors let topic_resp = topics_resp . remove ( 0 ) ; if topic_resp . error_code != FlvErrorCode :: None { return Err ( ClientError :: IoError ( IoError :: new ( ErrorKind :: InvalidData , format ! ( " " topic_resp error_code to_sentence ( ) ) , ) ) ) ; } // lookup leader for partition_resp in topic_resp . partitions { if partition_resp . partition_idx == partition { // check for errors if partition_resp . error_code != FlvErrorCode :: None { return Err ( ClientError :: IoError ( IoError :: new ( ErrorKind :: InvalidData , format ! ( " " topic_resp error_code to_sentence ( ) ) , ) ) ) ; } // traverse spus and find leader let leader_id = partition_resp . leader_id ; for spu_resp in & spus_resp { if spu_resp . spu_id == leader_id { // check for errors if spu_resp . error_code != FlvErrorCode :: None { return Err ( ClientError :: IoError ( IoError :: new ( ErrorKind :: InvalidData , format ! ( " " topic_resp error_code to_sentence ( ) ) , ) ) ) ; } debug ! ( " " spu_resp host spu_resp port ) ; let leader_config = LeaderConfig :: new ( spu_resp . into ( ) , topic . to_owned ( ) , partition ) . spu_id ( leader_id ) . client_id ( self . 0 . client_id ( ) ) ; return SpuLeader :: connect ( leader_config ) . await ; } } } } Err ( ClientError :: IoError ( IoError :: new ( ErrorKind :: Other , format ! ( " " topic partition ) , ) ) ) }
Find address of the SPU leader for a topic/partition
Find address of the SPU leader for a topic / partition
&mut self
topic
&str
partition
i32