pub async fn run(mut self, shutdown: CancellationToken) {
        let shutdown_fut = shutdown.cancelled().fuse();
        pin_mut!(shutdown_fut);

        let mut stream = self.write_buffer_stream_handler.stream().await;
        let mut sequence_number_before_reset: Option<SequenceNumber> = None;

        loop {
            // Wait for a DML operation from the sequencer, or a graceful stop signal.
            let maybe_op = futures::select!(
                next = stream.next().fuse() => next,
                _ = shutdown_fut => {
                    info!(
                        kafka_topic=%self.kafka_topic_name,
                        kafka_partition=%self.kafka_partition,
                        "stream handler shutdown",
                    );
                    return;
                }
            );

            // Read a DML op from the write buffer, logging and emitting metrics
            // for any potential errors to enable alerting on potential data
            // loss.
            //
            // If this evaluation results in no viable DML op to apply to the
            // DmlSink, return None rather than continuing the loop to ensure
            // ingest pauses are respected.
            let maybe_op = match maybe_op {
                Some(Ok(op)) => {
                    if let Some(sequence_number) = op.meta().sequence().map(|s| s.sequence_number) {
                        let sequence_number = SequenceNumber::new(sequence_number as i64);
                        if let Some(before_reset) = sequence_number_before_reset {
                            // We've requested the stream to be reset and we've skipped this many
                            // sequence numbers. Store in a metric once.
                            if before_reset != sequence_number {
                                let difference = sequence_number.get() - before_reset.get();
                                self.skipped_sequence_number_amount.inc(difference as u64);
                            }
                            sequence_number_before_reset = None;
                        }
                        self.current_sequence_number = sequence_number;
                    }

                    Some(op)
                }
                Some(Err(e)) if e.kind() == WriteBufferErrorKind::UnknownSequenceNumber => {
                    // If we get an unknown sequence number, and we're fine potentially having
                    // missed writes that were too old to be retained, try resetting the stream
                    // once and getting the next operation again.
                    // Keep the current sequence number to compare with the sequence number
                    if self.skip_to_oldest_available && sequence_number_before_reset.is_none() {
                        sequence_number_before_reset = Some(self.current_sequence_number);
                        self.write_buffer_stream_handler.reset_to_earliest();
                        stream = self.write_buffer_stream_handler.stream().await;
                        continue;
                    } else {
                        error!(
                            error=%e,
                            kafka_topic=%self.kafka_topic_name,
                            kafka_partition=%self.kafka_partition,
                            potential_data_loss=true,
                            "unable to read from desired sequencer offset"
                        );
                        self.seq_unknown_sequence_number_count.inc(1);
                        None
                    }
                }
                Some(Err(e)) if e.kind() == WriteBufferErrorKind::IO => {
                    warn!(
                        error=%e,
                        kafka_topic=%self.kafka_topic_name,
                        kafka_partition=%self.kafka_partition,
                        "I/O error reading from sequencer"
                    );
                    tokio::time::sleep(Duration::from_secs(1)).await;
                    None
                }
                Some(Err(e)) if e.kind() == WriteBufferErrorKind::InvalidData => {
                    // The DmlOperation could not be de-serialized from the
                    // kafka message.
                    //
                    // This is almost certainly data loss as the write will not
                    // be applied/persisted.
                    error!(
                        error=%e,
                        kafka_topic=%self.kafka_topic_name,
                        kafka_partition=%self.kafka_partition,
                        potential_data_loss=true,
                        "unable to deserialize dml operation"
                    );

                    self.seq_invalid_data_count.inc(1);
                    None
                }
                Some(Err(e)) => {
                    error!(
                        error=%e,
                        kafka_topic=%self.kafka_topic_name,
                        kafka_partition=%self.kafka_partition,
                        potential_data_loss=true,
                        "unhandled error converting write buffer data to DmlOperation",
                    );
                    self.seq_unknown_error_count.inc(1);
                    tokio::time::sleep(Duration::from_secs(1)).await;
                    None
                }
                None => {
                    panic!(
                        "sequencer {:?} stream for topic {} ended without graceful shutdown",
                        self.kafka_partition, self.kafka_topic_name
                    );
                }
            };

            // If a DML operation was successfully decoded, push it into the
            // DmlSink.
            self.maybe_apply_op(maybe_op).await;
        }
    }
/// Run the stream handler, consuming items from the stream provided by the
/// [`WriteBufferStreamHandler`] and applying them to the [`DmlSink`].
///
/// This method blocks until gracefully shutdown by cancelling the
/// `shutdown` [`CancellationToken`]. Once cancelled, this handler will
/// complete the current operation it is processing before this method
/// returns.
///
/// # Panics
///
/// This method panics if the input stream ends (yields a `None`).
Run the stream handler, consuming items from the stream provided by the
[`WriteBufferStreamHandler`] and applying them to the [`DmlSink`].

This method blocks until gracefully shutdown by cancelling the
`shutdown` [`CancellationToken`]. Once cancelled, this handler will
complete the current operation it is processing before this method
returns.
Run the stream handler consuming items from the stream provided by the [ ` WriteBufferStreamHandler ` ] and applying them to the [ ` DmlSink ` ] . This method blocks until gracefully shutdown by cancelling the ` shutdown ` [ ` CancellationToken ` ] . Once cancelled this handler will complete the current operation it is processing before this method returns .
pub async fn run(mut self, shutdown: CancellationToken) {
        let shutdown_fut = shutdown.cancelled().fuse();
        pin_mut!(shutdown_fut);
        let mut stream = self.write_buffer_stream_handler.stream().await;
        let mut sequence_number_before_reset: Option<SequenceNumber> = None;
        loop {
            // Wait for a DML operation from the sequencer, or a graceful stop signal.
            let maybe_op = futures::select!(
                next = stream.next().fuse() => next,
                _ = shutdown_fut => {
                    info!(
                        kafka_topic=%self.kafka_topic_name,
                        kafka_partition=%self.kafka_partition,
                        "stream handler shutdown",
                    );
                    return;
                }
            );
            // Read a DML op from the write buffer, logging and emitting metrics
            // for any potential errors to enable alerting on potential data
            // loss.
            //
            // If this evaluation results in no viable DML op to apply to the
            // DmlSink, return None rather than continuing the loop to ensure
            // ingest pauses are respected.
            let maybe_op = match maybe_op {
                Some(Ok(op)) => {
                    if let Some(sequence_number) = op.meta().sequence().map(|s| s.sequence_number) {
                        let sequence_number = SequenceNumber::new(sequence_number as i64);
                        if let Some(before_reset) = sequence_number_before_reset {
                            // We've requested the stream to be reset and we've skipped this many
                            // sequence numbers. Store in a metric once.
                            if before_reset != sequence_number {
                                let difference = sequence_number.get() - before_reset.get();
                                self.skipped_sequence_number_amount.inc(difference as u64);
                            }
                            sequence_number_before_reset = None;
                        }
                        self.current_sequence_number = sequence_number;
                    }
                    Some(op)
                }
                Some(Err(e)) if e.kind() == WriteBufferErrorKind::UnknownSequenceNumber => {
                    // If we get an unknown sequence number, and we're fine potentially having
                    // missed writes that were too old to be retained, try resetting the stream
                    // once and getting the next operation again.
                    // Keep the current sequence number to compare with the sequence number
                    if self.skip_to_oldest_available && sequence_number_before_reset.is_none() {
                        sequence_number_before_reset = Some(self.current_sequence_number);
                        self.write_buffer_stream_handler.reset_to_earliest();
                        stream = self.write_buffer_stream_handler.stream().await;
                        continue;
                    } else {
                        error!(
                            error=%e,
                            kafka_topic=%self.kafka_topic_name,
                            kafka_partition=%self.kafka_partition,
                            potential_data_loss=true,
                            "unable to read from desired sequencer offset"
                        );
                        self.seq_unknown_sequence_number_count.inc(1);
                        None
                    }
                }
                Some(Err(e)) if e.kind() == WriteBufferErrorKind::IO => {
                    warn!(
                        error=%e,
                        kafka_topic=%self.kafka_topic_name,
                        kafka_partition=%self.kafka_partition,
                        "I/O error reading from sequencer"
                    );
                    tokio::time::sleep(Duration::from_secs(1)).await;
                    None
                }
                Some(Err(e)) if e.kind() == WriteBufferErrorKind::InvalidData => {
                    // The DmlOperation could not be de-serialized from the
                    // kafka message.
                    //
                    // This is almost certainly data loss as the write will not
                    // be applied/persisted.
                    error!(
                        error=%e,
                        kafka_topic=%self.kafka_topic_name,
                        kafka_partition=%self.kafka_partition,
                        potential_data_loss=true,
                        "unable to deserialize dml operation"
                    );
                    self.seq_invalid_data_count.inc(1);
                    None
                }
                Some(Err(e)) => {
                    error!(
                        error=%e,
                        kafka_topic=%self.kafka_topic_name,
                        kafka_partition=%self.kafka_partition,
                        potential_data_loss=true,
                        "unhandled error converting write buffer data to DmlOperation",
                    );
                    self.seq_unknown_error_count.inc(1);
                    tokio::time::sleep(Duration::from_secs(1)).await;
                    None
                }
                None => {
                    panic!(
                        "sequencer {:?} stream for topic {} ended without graceful shutdown",
                        self.kafka_partition, self.kafka_topic_name
                    );
                }
            };
            // If a DML operation was successfully decoded, push it into the
            // DmlSink.
            self.maybe_apply_op(maybe_op).await;
        }
    }
pub async fn run ( mut self , shutdown : CancellationToken ) { let shutdown_fut = shutdown . cancelled ( ) . fuse ( ) ; pin_mut ! ( shutdown_fut ) ; let mut stream = self . write_buffer_stream_handler . stream ( ) . await ; let mut sequence_number_before_reset : Option < SequenceNumber > = None ; loop { // Wait for a DML operation from the sequencer, or a graceful stop signal. let maybe_op = futures :: select ! ( next stream next ( ) fuse ( ) next shutdown_fut { info ( kafka_topic self kafka_topic_name kafka_partition self kafka_partition " " ) return } ) ; // Read a DML op from the write buffer, logging and emitting metrics // for any potential errors to enable alerting on potential data // loss. // // If this evaluation results in no viable DML op to apply to the // DmlSink, return None rather than continuing the loop to ensure // ingest pauses are respected. let maybe_op = match maybe_op { Some ( Ok ( op ) ) => { if let Some ( sequence_number ) = op . meta ( ) . sequence ( ) . map ( | s | s . sequence_number ) { let sequence_number = SequenceNumber :: new ( sequence_number as i64 ) ; if let Some ( before_reset ) = sequence_number_before_reset { // We've requested the stream to be reset and we've skipped this many // sequence numbers. Store in a metric once. if before_reset != sequence_number { let difference = sequence_number . get ( ) - before_reset . get ( ) ; self . skipped_sequence_number_amount . inc ( difference as u64 ) ; } sequence_number_before_reset = None ; } self . current_sequence_number = sequence_number ; } Some ( op ) } Some ( Err ( e ) ) if e . kind ( ) == WriteBufferErrorKind :: UnknownSequenceNumber => { // If we get an unknown sequence number, and we're fine potentially having // missed writes that were too old to be retained, try resetting the stream // once and getting the next operation again. // Keep the current sequence number to compare with the sequence number if self . skip_to_oldest_available && sequence_number_before_reset . is_none ( ) { sequence_number_before_reset = Some ( self . current_sequence_number ) ; self . write_buffer_stream_handler . reset_to_earliest ( ) ; stream = self . write_buffer_stream_handler . stream ( ) . await ; continue ; } else { error ! ( error e kafka_topic self kafka_topic_name kafka_partition self kafka_partition potential_data_loss true " " ) ; self . seq_unknown_sequence_number_count . inc ( 1 ) ; None } } Some ( Err ( e ) ) if e . kind ( ) == WriteBufferErrorKind :: IO => { warn ! ( error e kafka_topic self kafka_topic_name kafka_partition self kafka_partition " " ) ; tokio :: time :: sleep ( Duration :: from_secs ( 1 ) ) . await ; None } Some ( Err ( e ) ) if e . kind ( ) == WriteBufferErrorKind :: InvalidData => { // The DmlOperation could not be de-serialized from the // kafka message. // // This is almost certainly data loss as the write will not // be applied/persisted. error ! ( error e kafka_topic self kafka_topic_name kafka_partition self kafka_partition potential_data_loss true " " ) ; self . seq_invalid_data_count . inc ( 1 ) ; None } Some ( Err ( e ) ) => { error ! ( error e kafka_topic self kafka_topic_name kafka_partition self kafka_partition potential_data_loss true " " ) ; self . seq_unknown_error_count . inc ( 1 ) ; tokio :: time :: sleep ( Duration :: from_secs ( 1 ) ) . await ; None } None => { panic ! ( " " self kafka_partition self kafka_topic_name ) ; } } ; // If a DML operation was successfully decoded, push it into the // DmlSink. self . maybe_apply_op ( maybe_op ) . await ; } }
Run the stream handler, consuming items from the stream provided by the
[`WriteBufferStreamHandler`] and applying them to the [`DmlSink`].
Run the stream handler consuming items from the stream provided by the [ ` WriteBufferStreamHandler ` ] and applying them to the [ ` DmlSink ` ] .
mut self
shutdown
CancellationToken
mut self
shutdown
CancellationToken
panics
This method panics if the input stream ends (yields a `None`).
This method panics if the input stream ends ( yields a ` None ` ) .