fn maybe_emit_progress(
        &mut self,
        input_frontier: AntichainRef<Timestamp>,
    ) -> Result<(), anyhow::Error> {
        // 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 self.consistency.is_some() {
                    if self.transactional {
                        self.producer.begin_transaction()?
                    }

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

                    if self.transactional {
                        self.producer.commit_transaction(self.txn_timeout)?
                    }
                }
                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(())
    }
/// 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 .
fn maybe_emit_progress(
        &mut self,
        input_frontier: AntichainRef<Timestamp>,
    ) -> Result<(), anyhow::Error> {
        // 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 self.consistency.is_some() {
                    if self.transactional {
                        self.producer.begin_transaction()?
                    }
                    self.send_consistency_record(&min_frontier.to_string(), "END", None)
                        .map_err(|_e| anyhow::anyhow!("Error sending write frontier update."))?;
                    if self.transactional {
                        self.producer.commit_transaction(self.txn_timeout)?
                    }
                }
                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(())
    }
fn maybe_emit_progress ( & mut self , input_frontier : AntichainRef < Timestamp > , ) -> Result < ( ) , anyhow :: Error > { // 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 self . consistency . is_some ( ) { if self . transactional { self . producer . begin_transaction ( ) ? } self . send_consistency_record ( & min_frontier . to_string ( ) , " " , None ) . map_err ( | _e | anyhow :: anyhow ! ( " " ) ) ? ; if self . transactional { self . producer . commit_transaction ( self . txn_timeout ) ? } } 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 ( ( ) ) }
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<Timestamp>
&mut self
input_frontier
AntichainRef<Timestamp>