fn generate_ts_updates_from_debezium(
    id: &GlobalId,
    tx: &mpsc::UnboundedSender<coord::Message>,
    byo_consumer: &mut ByoTimestampConsumer,
    value: Value,
) -> Result<(), anyhow::Error> {
    if let Value::Record(record) = value {
        let results =
            parse_debezium(record).with_context(|| format!("Failed to parse debezium record"))?;

        // Results are only returned when the record's status type is END
        if let Some(results) = results {
            byo_consumer.last_ts += 1;
            for (topic, count) in results {
                // Debezium topics are formatted as "server_name.database.topic", but
                // entries in data_collection do not contain server_name
                // so we discard it before doing the comparison.
                // We check for both here
                // TODO(): possible performance issue here?
                let parsed_source_name = byo_consumer.source_name.split('.').skip(1).join(".");
                if byo_consumer.source_name == topic.trim() || parsed_source_name == topic.trim() {
                    byo_consumer.last_offset.offset += count;
                    // Debezium consistency topic should only work for single-partition
                    // topics
                    tx.send(coord::Message::AdvanceSourceTimestamp(
                        coord::AdvanceSourceTimestamp {
                            id: *id,
                            update: TimestampSourceUpdate::BringYourOwn(
                                match byo_consumer.connector {
                                    ByoTimestampConnector::Kafka(_) => PartitionId::Kafka(0),
                                },
                                byo_consumer.last_ts,
                                byo_consumer.last_offset,
                            ),
                        },
                    ))
                    .expect("Failed to send update to coordinator");
                }
            }
        }
    } else {
        bail!("Expected record type for value, got {:?}", value);
    }

    Ok(())
}
/// Extracts Materialize timestamp updates from a Debezium consistency record.
Extracts Materialize timestamp updates from a Debezium consistency record.
Extracts Materialize timestamp updates from a Debezium consistency record .
fn generate_ts_updates_from_debezium(
    id: &GlobalId,
    tx: &mpsc::UnboundedSender<coord::Message>,
    byo_consumer: &mut ByoTimestampConsumer,
    value: Value,
) -> Result<(), anyhow::Error> {
    if let Value::Record(record) = value {
        let results =
            parse_debezium(record).with_context(|| format!("Failed to parse debezium record"))?;
        // Results are only returned when the record's status type is END
        if let Some(results) = results {
            byo_consumer.last_ts += 1;
            for (topic, count) in results {
                // Debezium topics are formatted as "server_name.database.topic", but
                // entries in data_collection do not contain server_name
                // so we discard it before doing the comparison.
                // We check for both here
                // TODO(): possible performance issue here?
                let parsed_source_name = byo_consumer.source_name.split('.').skip(1).join(".");
                if byo_consumer.source_name == topic.trim() || parsed_source_name == topic.trim() {
                    byo_consumer.last_offset.offset += count;
                    // Debezium consistency topic should only work for single-partition
                    // topics
                    tx.send(coord::Message::AdvanceSourceTimestamp(
                        coord::AdvanceSourceTimestamp {
                            id: *id,
                            update: TimestampSourceUpdate::BringYourOwn(
                                match byo_consumer.connector {
                                    ByoTimestampConnector::Kafka(_) => PartitionId::Kafka(0),
                                },
                                byo_consumer.last_ts,
                                byo_consumer.last_offset,
                            ),
                        },
                    ))
                    .expect("Failed to send update to coordinator");
                }
            }
        }
    } else {
        bail!("Expected record type for value, got {:?}", value);
    }
    Ok(())
}
fn generate_ts_updates_from_debezium ( id : & GlobalId , tx : & mpsc :: UnboundedSender < coord :: Message > , byo_consumer : & mut ByoTimestampConsumer , value : Value , ) -> Result < ( ) , anyhow :: Error > { if let Value :: Record ( record ) = value { let results = parse_debezium ( record ) . with_context ( | | format ! ( " " ) ) ? ; // Results are only returned when the record's status type is END if let Some ( results ) = results { byo_consumer . last_ts += 1 ; for ( topic , count ) in results { // Debezium topics are formatted as "server_name.database.topic", but // entries in data_collection do not contain server_name // so we discard it before doing the comparison. // We check for both here // TODO(): possible performance issue here? let parsed_source_name = byo_consumer . source_name . split ( '.' ) . skip ( 1 ) . join ( " " ) ; if byo_consumer . source_name == topic . trim ( ) || parsed_source_name == topic . trim ( ) { byo_consumer . last_offset . offset += count ; // Debezium consistency topic should only work for single-partition // topics tx . send ( coord :: Message :: AdvanceSourceTimestamp ( coord :: AdvanceSourceTimestamp { id : * id , update : TimestampSourceUpdate :: BringYourOwn ( match byo_consumer . connector { ByoTimestampConnector :: Kafka ( _ ) => PartitionId :: Kafka ( 0 ) , } , byo_consumer . last_ts , byo_consumer . last_offset , ) , } , ) ) . expect ( " " ) ; } } } } else { bail ! ( " " value ) ; } Ok ( ( ) ) }
Extracts Materialize timestamp updates from a Debezium consistency record.
Extracts Materialize timestamp updates from a Debezium consistency record .
id
&GlobalId
tx
&mpsc::UnboundedSender<coord::Message>
byo_consumer
&mut ByoTimestampConsumer
value
Value
id
&GlobalId
tx
&mpsc::UnboundedSender<coord::Message>
byo_consumer
&mut ByoTimestampConsumer
value
Value