pub(crate) async fn progresses(
        &self,
        partitions: Vec<KafkaPartition>,
    ) -> BTreeMap<KafkaPartition, SequencerProgress> {
        let mut progresses = BTreeMap::new();
        for kafka_partition in partitions {
            let sequencer_data = self
                .sequencers
                .iter()
                .map(|(_, sequencer_data)| sequencer_data)
                .find(|sequencer_data| sequencer_data.kafka_partition == kafka_partition);

            let progress = match sequencer_data {
                Some(sequencer_data) => sequencer_data.progress().await,
                None => SequencerProgress::new(), // don't know about this sequencer
            };

            progresses.insert(kafka_partition, progress);
        }
        progresses
    }
/// Return the ingestion progress for the specified kafka
/// partitions. Returns an empty `SequencerProgress` for any kafka
/// partitions that this ingester doesn't know about.
Return the ingestion progress for the specified kafka
partitions. Returns an empty `SequencerProgress` for any kafka
partitions that this ingester doesn't know about.
Return the ingestion progress for the specified kafka partitions . Returns an empty ` SequencerProgress ` for any kafka partitions that this ingester doesn ' t know about .
pub(crate) async fn progresses(
        &self,
        partitions: Vec<KafkaPartition>,
    ) -> BTreeMap<KafkaPartition, SequencerProgress> {
        let mut progresses = BTreeMap::new();
        for kafka_partition in partitions {
            let sequencer_data = self
                .sequencers
                .iter()
                .map(|(_, sequencer_data)| sequencer_data)
                .find(|sequencer_data| sequencer_data.kafka_partition == kafka_partition);
            let progress = match sequencer_data {
                Some(sequencer_data) => sequencer_data.progress().await,
                None => SequencerProgress::new(), // don't know about this sequencer
            };
            progresses.insert(kafka_partition, progress);
        }
        progresses
    }
pub ( crate ) async fn progresses ( & self , partitions : Vec < KafkaPartition > , ) -> BTreeMap < KafkaPartition , SequencerProgress > { let mut progresses = BTreeMap :: new ( ) ; for kafka_partition in partitions { let sequencer_data = self . sequencers . iter ( ) . map ( | ( _ , sequencer_data ) | sequencer_data ) . find ( | sequencer_data | sequencer_data . kafka_partition == kafka_partition ) ; let progress = match sequencer_data { Some ( sequencer_data ) => sequencer_data . progress ( ) . await , None => SequencerProgress :: new ( ) , // don't know about this sequencer } ; progresses . insert ( kafka_partition , progress ) ; } progresses }
Return the ingestion progress for the specified kafka
partitions.
Return the ingestion progress for the specified kafka partitions .
&self
partitions
Vec<KafkaPartition>
&self
partitions
Vec<KafkaPartition>