/// One time initialize Sequencers of this Ingester
One time initialize Sequencers of this Ingester
One time initialize Sequencers of this Ingester
pub async fn initialize(ingester: &IngestHandlerImpl) -> Result<Self> {
        // Get sequencer ids from the catalog
        let sequencer_repro = ingester.iox_catalog.sequencers();
        let mut sequencers = BTreeMap::default();
        let topic = ingester.get_topic();
        for shard in ingester.get_kafka_partitions() {
            let sequencer = sequencer_repro
                .get_by_topic_id_and_partition(topic.id, shard)
                .await
                .context(ReadSequencerSnafu { id: shard })?
                .context(SequencerNotFoundSnafu {
                    kafka_topic_id: topic.id,
                    kafka_partition: shard,
                })?;
            // Create empty buffer for each sequencer
            sequencers.insert(sequencer.id, Arc::new(SequencerData::default()));
        }
        Ok(Self { data: sequencers })
    }
pub async fn initialize ( ingester : & IngestHandlerImpl ) -> Result < Self > { // Get sequencer ids from the catalog let sequencer_repro = ingester . iox_catalog . sequencers ( ) ; let mut sequencers = BTreeMap :: default ( ) ; let topic = ingester . get_topic ( ) ; for shard in ingester . get_kafka_partitions ( ) { let sequencer = sequencer_repro . get_by_topic_id_and_partition ( topic . id , shard ) . await . context ( ReadSequencerSnafu { id : shard } ) ? . context ( SequencerNotFoundSnafu { kafka_topic_id : topic . id , kafka_partition : shard , } ) ? ; // Create empty buffer for each sequencer sequencers . insert ( sequencer . id , Arc :: new ( SequencerData :: default ( ) ) ) ; } Ok ( Self { data : sequencers } ) }
One time initialize Sequencers of this Ingester
One time initialize Sequencers of this Ingester
ingester
&IngestHandlerImpl