async fn maybe_emit_progress<'a>(
        &mut self,
        input_frontier: AntichainRef<'a, Timestamp>,
    ) -> Result<bool, anyhow::Error> {
        let mut progress_emitted = false;
        // This only looks at the first entry of the antichain.
        // If we ever have multi-dimensional time, this is not correct
        // anymore. There might not even be progress in the first dimension.
        // We panic, so that future developers introducing multi-dimensional
        // time in Materialize will notice.
        let input_frontier = input_frontier
            .iter()
            .at_most_one()
            .expect("more than one element in the frontier")
            .cloned();

        let min_pending_ts = self.pending_rows.keys().min().cloned();

        let min_frontier = input_frontier.into_iter().chain(min_pending_ts).min();

        if let Some(min_frontier) = min_frontier {
            // a frontier of `t` means we still might receive updates with `t`.
            // The progress frontier we emit is a strict frontier, so subtract `1`.
            let min_frontier = min_frontier.saturating_sub(1);

            if min_frontier > self.latest_progress_ts {
                // record the write frontier in the consistency topic.
                if let Some(consistency_state) = self.sink_state.unwrap_running() {
                    if self.transactional {
                        self.retry_on_txn_error(|p| p.begin_transaction()).await?;
                    }

                    self.send_consistency_record(
                        &min_frontier.to_string(),
                        "END",
                        None,
                        consistency_state,
                    )
                    .await
                    .map_err(|_| anyhow::anyhow!("Error sending write frontier update."))?;

                    if self.transactional {
                        self.retry_on_txn_error(|p| p.commit_transaction()).await?;
                    }
                    progress_emitted = true;
                }
                self.latest_progress_ts = min_frontier;
            }

            let mut write_frontier = self.write_frontier.borrow_mut();

            // make sure we don't regress
            assert!(write_frontier.less_equal(&min_frontier));
            write_frontier.clear();
            write_frontier.insert(min_frontier);
        } else {
            // If there's no longer an input frontier, we will no longer receive any data forever and, therefore, will
            // never output more data
            self.write_frontier.borrow_mut().clear();
        }

        Ok(progress_emitted)
    }
/// Updates the latest progress update timestamp based on the given
/// input frontier and pending rows.
///
/// This will emit an `END` record to the consistency topic if the frontier
/// advanced and advance the maintained write frontier, which will in turn
/// unblock compaction of timestamp bindings in sources.
///
/// *NOTE*: `END` records will only be emitted when
/// `KafkaSinkConnector.consistency` points to a consistency topic. The
/// write frontier will be advanced regardless.
Updates the latest progress update timestamp based on the given
input frontier and pending rows.

This will emit an `END` record to the consistency topic if the frontier
advanced and advance the maintained write frontier, which will in turn
unblock compaction of timestamp bindings in sources.

NOTE*: `END` records will only be emitted when
`KafkaSinkConnector.consistency` points to a consistency topic. The
write frontier will be advanced regardless.
Updates the latest progress update timestamp based on the given input frontier and pending rows . This will emit an ` END ` record to the consistency topic if the frontier advanced and advance the maintained write frontier which will in turn unblock compaction of timestamp bindings in sources . NOTE * : ` END ` records will only be emitted when ` KafkaSinkConnector . consistency ` points to a consistency topic . The write frontier will be advanced regardless .
async fn maybe_emit_progress<'a>(
        &mut self,
        input_frontier: AntichainRef<'a, Timestamp>,
    ) -> Result<bool, anyhow::Error> {
        let mut progress_emitted = false;
        // This only looks at the first entry of the antichain.
        // If we ever have multi-dimensional time, this is not correct
        // anymore. There might not even be progress in the first dimension.
        // We panic, so that future developers introducing multi-dimensional
        // time in Materialize will notice.
        let input_frontier = input_frontier
            .iter()
            .at_most_one()
            .expect("more than one element in the frontier")
            .cloned();
        let min_pending_ts = self.pending_rows.keys().min().cloned();
        let min_frontier = input_frontier.into_iter().chain(min_pending_ts).min();
        if let Some(min_frontier) = min_frontier {
            // a frontier of `t` means we still might receive updates with `t`.
            // The progress frontier we emit is a strict frontier, so subtract `1`.
            let min_frontier = min_frontier.saturating_sub(1);
            if min_frontier > self.latest_progress_ts {
                // record the write frontier in the consistency topic.
                if let Some(consistency_state) = self.sink_state.unwrap_running() {
                    if self.transactional {
                        self.retry_on_txn_error(|p| p.begin_transaction()).await?;
                    }
                    self.send_consistency_record(
                        &min_frontier.to_string(),
                        "END",
                        None,
                        consistency_state,
                    )
                    .await
                    .map_err(|_| anyhow::anyhow!("Error sending write frontier update."))?;
                    if self.transactional {
                        self.retry_on_txn_error(|p| p.commit_transaction()).await?;
                    }
                    progress_emitted = true;
                }
                self.latest_progress_ts = min_frontier;
            }
            let mut write_frontier = self.write_frontier.borrow_mut();
            // make sure we don't regress
            assert!(write_frontier.less_equal(&min_frontier));
            write_frontier.clear();
            write_frontier.insert(min_frontier);
        } else {
            // If there's no longer an input frontier, we will no longer receive any data forever and, therefore, will
            // never output more data
            self.write_frontier.borrow_mut().clear();
        }
        Ok(progress_emitted)
    }
async fn maybe_emit_progress < ' a > ( & mut self , input_frontier : AntichainRef < ' a , Timestamp > , ) -> Result < bool , anyhow :: Error > { let mut progress_emitted = false ; // This only looks at the first entry of the antichain. // If we ever have multi-dimensional time, this is not correct // anymore. There might not even be progress in the first dimension. // We panic, so that future developers introducing multi-dimensional // time in Materialize will notice. let input_frontier = input_frontier . iter ( ) . at_most_one ( ) . expect ( " " ) . cloned ( ) ; let min_pending_ts = self . pending_rows . keys ( ) . min ( ) . cloned ( ) ; let min_frontier = input_frontier . into_iter ( ) . chain ( min_pending_ts ) . min ( ) ; if let Some ( min_frontier ) = min_frontier { // a frontier of `t` means we still might receive updates with `t`. // The progress frontier we emit is a strict frontier, so subtract `1`. let min_frontier = min_frontier . saturating_sub ( 1 ) ; if min_frontier > self . latest_progress_ts { // record the write frontier in the consistency topic. if let Some ( consistency_state ) = self . sink_state . unwrap_running ( ) { if self . transactional { self . retry_on_txn_error ( | p | p . begin_transaction ( ) ) . await ? ; } self . send_consistency_record ( & min_frontier . to_string ( ) , " " , None , consistency_state , ) . await . map_err ( | _ | anyhow :: anyhow ! ( " " ) ) ? ; if self . transactional { self . retry_on_txn_error ( | p | p . commit_transaction ( ) ) . await ? ; } progress_emitted = true ; } self . latest_progress_ts = min_frontier ; } let mut write_frontier = self . write_frontier . borrow_mut ( ) ; // make sure we don't regress assert ! ( write_frontier less_equal ( min_frontier ) ) ; write_frontier . clear ( ) ; write_frontier . insert ( min_frontier ) ; } else { // If there's no longer an input frontier, we will no longer receive any data forever and, therefore, will // never output more data self . write_frontier . borrow_mut ( ) . clear ( ) ; } Ok ( progress_emitted ) }
Updates the latest progress update timestamp based on the given
input frontier and pending rows.
Updates the latest progress update timestamp based on the given input frontier and pending rows .
&mut self
input_frontier
AntichainRef<'a, Timestamp>
&mut self
input_frontier
AntichainRef<'a, Timestamp>