fn update_source_timestamps(
    id: &GlobalId,
    tx: &mpsc::UnboundedSender<coord::Message>,
    byo_consumer: &mut ByoTimestampConsumer,
) -> Result<(), anyhow::Error> {
    let messages = byo_query_source(byo_consumer)?;
    match byo_consumer.envelope {
        ConsistencyFormatting::DebeziumAvro => {
            for msg in messages {
                let ValueEncoding::Bytes(msg) = msg;
                // The first 5 bytes are reserved for the schema id/schema registry information
                let mut bytes = &msg[5..];
                let res = mz_avro::from_avro_datum(&DEBEZIUM_TRX_SCHEMA_VALUE, &mut bytes);
                match res {
                    Err(_) => {
                        // This was a key message, can safely ignore it
                        // TODO (#6671): raise error on failure to parse
                        continue;
                    }
                    Ok(record) => {
                        generate_ts_updates_from_debezium(&id, tx, byo_consumer, record)?;
                    }
                }
            }
        }
    }

    Ok(())
}
/// Implements the byo timestamping logic
///
/// If the partition count remains the same:
/// A new timestamp should be
/// 1) strictly greater than the last timestamp in this partition
/// 2) greater or equal to all the timestamps that have been assigned so far across all partitions
/// If the partition count increases:
/// A new timestamp should be:
/// 1) strictly greater than the last timestamp
/// This is necessary to guarantee that this timestamp *could not have been closed yet*
///
/// Supports two envelopes: None and Debezium. Currentlye compatible with Debezium format 1.1
Implements the byo timestamping logic
If the partition count remains the same:
A new timestamp should be
1) strictly greater than the last timestamp in this partition
2) greater or equal to all the timestamps that have been assigned so far across all partitions
If the partition count increases:
A new timestamp should be:
1) strictly greater than the last timestamp
This is necessary to guarantee that this timestamp *could not have been closed yet

Supports two envelopes: None and Debezium. Currentlye compatible with Debezium format 1.1
Implements the byo timestamping logic If the partition count remains the same : A new timestamp should be 1 ) strictly greater than the last timestamp in this partition 2 ) greater or equal to all the timestamps that have been assigned so far across all partitions If the partition count increases : A new timestamp should be : 1 ) strictly greater than the last timestamp This is necessary to guarantee that this timestamp * could not have been closed yet Supports two envelopes : None and Debezium . Currentlye compatible with Debezium format 1 . 1
fn update_source_timestamps(
    id: &GlobalId,
    tx: &mpsc::UnboundedSender<coord::Message>,
    byo_consumer: &mut ByoTimestampConsumer,
) -> Result<(), anyhow::Error> {
    let messages = byo_query_source(byo_consumer)?;
    match byo_consumer.envelope {
        ConsistencyFormatting::DebeziumAvro => {
            for msg in messages {
                let ValueEncoding::Bytes(msg) = msg;
                // The first 5 bytes are reserved for the schema id/schema registry information
                let mut bytes = &msg[5..];
                let res = mz_avro::from_avro_datum(&DEBEZIUM_TRX_SCHEMA_VALUE, &mut bytes);
                match res {
                    Err(_) => {
                        // This was a key message, can safely ignore it
                        // TODO (#6671): raise error on failure to parse
                        continue;
                    }
                    Ok(record) => {
                        generate_ts_updates_from_debezium(&id, tx, byo_consumer, record)?;
                    }
                }
            }
        }
    }
    Ok(())
}
fn update_source_timestamps ( id : & GlobalId , tx : & mpsc :: UnboundedSender < coord :: Message > , byo_consumer : & mut ByoTimestampConsumer , ) -> Result < ( ) , anyhow :: Error > { let messages = byo_query_source ( byo_consumer ) ? ; match byo_consumer . envelope { ConsistencyFormatting :: DebeziumAvro => { for msg in messages { let ValueEncoding :: Bytes ( msg ) = msg ; // The first 5 bytes are reserved for the schema id/schema registry information let mut bytes = & msg [ 5 .. ] ; let res = mz_avro :: from_avro_datum ( & DEBEZIUM_TRX_SCHEMA_VALUE , & mut bytes ) ; match res { Err ( _ ) => { // This was a key message, can safely ignore it // TODO (#6671): raise error on failure to parse continue ; } Ok ( record ) => { generate_ts_updates_from_debezium ( & id , tx , byo_consumer , record ) ? ; } } } } } Ok ( ( ) ) }
Implements the byo timestamping logic
If the partition count remains the same:
A new timestamp should be
1) strictly greater than the last timestamp in this partition
2) greater or equal to all the timestamps that have been assigned so far across all partitions
If the partition count increases:
A new timestamp should be:
1) strictly greater than the last timestamp
This is necessary to guarantee that this timestamp *could not have been closed yet
Implements the byo timestamping logic If the partition count remains the same : A new timestamp should be 1 ) strictly greater than the last timestamp in this partition 2 ) greater or equal to all the timestamps that have been assigned so far across all partitions If the partition count increases : A new timestamp should be : 1 ) strictly greater than the last timestamp This is necessary to guarantee that this timestamp * could not have been closed yet
id
&GlobalId
tx
&mpsc::UnboundedSender<coord::Message>
byo_consumer
&mut ByoTimestampConsumer
id
&GlobalId
tx
&mpsc::UnboundedSender<coord::Message>
byo_consumer
&mut ByoTimestampConsumer