/// Fetch log records from a target server
Fetch log records from a target server
Fetch log records from a target server
pub async fn kf_fetch_logs(
        &mut self,
        fetch_log_param: &FetchLogsParam,
    ) -> Result<DefaultKfFetchResponse, KfSocketError > {
        let mut fetch_partitions = vec![];
        for partition_param in &fetch_log_param.partitions {
            let mut fetch_part = FetchPartition::default();
            fetch_part.partition_index = partition_param.partition_idx;
            fetch_part.current_leader_epoch = partition_param.epoch;
            fetch_part.fetch_offset = partition_param.offset;
            fetch_part.log_start_offset = -1;
            fetch_part.max_bytes = fetch_log_param.max_bytes;
            fetch_partitions.push(fetch_part);
        }
        let mut topic_request = FetchableTopic::default();
        topic_request.name = fetch_log_param.topic.clone();
        topic_request.fetch_partitions = fetch_partitions;
        let mut request: DefaultKfFetchRequest = KfFetchRequest::default();
        request.replica_id = -1;
        request.max_wait = 500;
        request.min_bytes = 1;
        request.max_bytes = fetch_log_param.max_bytes;
        request.isolation_level = Isolation::ReadCommitted;
        request.session_id = 0;
        request.epoch = -1;
        request.topics.push(topic_request);
        self.0.send_receive(request).await
    }
pub async fn kf_fetch_logs ( & mut self , fetch_log_param : & FetchLogsParam , ) -> Result < DefaultKfFetchResponse , KfSocketError > { let mut fetch_partitions = vec ! [ ] ; for partition_param in & fetch_log_param . partitions { let mut fetch_part = FetchPartition :: default ( ) ; fetch_part . partition_index = partition_param . partition_idx ; fetch_part . current_leader_epoch = partition_param . epoch ; fetch_part . fetch_offset = partition_param . offset ; fetch_part . log_start_offset = - 1 ; fetch_part . max_bytes = fetch_log_param . max_bytes ; fetch_partitions . push ( fetch_part ) ; } let mut topic_request = FetchableTopic :: default ( ) ; topic_request . name = fetch_log_param . topic . clone ( ) ; topic_request . fetch_partitions = fetch_partitions ; let mut request : DefaultKfFetchRequest = KfFetchRequest :: default ( ) ; request . replica_id = - 1 ; request . max_wait = 500 ; request . min_bytes = 1 ; request . max_bytes = fetch_log_param . max_bytes ; request . isolation_level = Isolation :: ReadCommitted ; request . session_id = 0 ; request . epoch = - 1 ; request . topics . push ( topic_request ) ; self . 0 . send_receive ( request ) . await }
Fetch log records from a target server
Fetch log records from a target server
&mut self
fetch_log_param
&FetchLogsParam