// Query Group Coordinator for offsets.
Query Group Coordinator for offsets.
Query Group Coordinator for offsets .
pub async fn offsets_fetch(
        &mut self,
        group_id: &str,
        topic_name: &str,
        tp_param: &TopicPartitionParam,
    ) -> Result<KfOffsetFetchResponse, KfSocketError> {
        let mut request = KfOffsetFetchRequest::default();
        // collect partition indexes
        let mut partition_indexes: Vec<i32> = vec![];
        for leader in &tp_param.leaders {
            for partition in &leader.partitions {
                partition_indexes.push(partition.partition_idx);
            }
        }
        // topics
        let topics = vec![OffsetFetchRequestTopic {
            name: topic_name.to_owned(),
            partition_indexes: partition_indexes,
        }];
        // request
        request.group_id = group_id.to_owned();
        request.topics = Some(topics);
        self.0.send_receive(request).await
    }
pub async fn offsets_fetch ( & mut self , group_id : & str , topic_name : & str , tp_param : & TopicPartitionParam , ) -> Result < KfOffsetFetchResponse , KfSocketError > { let mut request = KfOffsetFetchRequest :: default ( ) ; // collect partition indexes let mut partition_indexes : Vec < i32 > = vec ! [ ] ; for leader in & tp_param . leaders { for partition in & leader . partitions { partition_indexes . push ( partition . partition_idx ) ; } } // topics let topics = vec ! [ OffsetFetchRequestTopic { name topic_name to_owned ( ) partition_indexes partition_indexes } ] ; // request request . group_id = group_id . to_owned ( ) ; request . topics = Some ( topics ) ; self . 0 . send_receive ( request ) . await }
Query Group Coordinator for offsets.
Query Group Coordinator for offsets .
&mut self
group_id
&str
topic_name
&str
tp_param
&TopicPartitionParam