fn maybe_decode_consistency_end_record(
    bytes: &[u8],
    consistency_topic: &str,
) -> Result<Option<Timestamp>, anyhow::Error> {
    // The first 5 bytes are reserved for the schema id/schema registry information
    let mut bytes = &bytes[5..];
    let record = mz_avro::from_avro_datum(get_debezium_transaction_schema(), &mut bytes)
        .context("Failed to decode consistency topic message")?;

    if let Value::Record(ref r) = record {
        let m: HashMap<String, Value> = r.clone().into_iter().collect();
        let status = m.get("status");
        let id = m.get("id");
        match (status, id) {
            (Some(Value::String(status)), Some(Value::String(id))) if status == "END" => {
                if let Ok(ts) = id.parse::<u64>() {
                    Ok(Some(ts))
                } else {
                    bail!(
                        "Malformed consistency record, failed to parse timestamp {} in topic {}",
                        id,
                        consistency_topic
                    );
                }
            }
            _ => Ok(None),
        }
    } else {
        Ok(None)
    }
}
// There may be arbitrary messages in this topic that we cannot decode.  We only
// return an error when we know we've found an END message but cannot decode it.
There may be arbitrary messages in this topic that we cannot decode.  We only
return an error when we know we've found an END message but cannot decode it.
There may be arbitrary messages in this topic that we cannot decode . We only return an error when we know we ' ve found an END message but cannot decode it .
fn maybe_decode_consistency_end_record(
    bytes: &[u8],
    consistency_topic: &str,
) -> Result<Option<Timestamp>, anyhow::Error> {
    // The first 5 bytes are reserved for the schema id/schema registry information
    let mut bytes = &bytes[5..];
    let record = mz_avro::from_avro_datum(get_debezium_transaction_schema(), &mut bytes)
        .context("Failed to decode consistency topic message")?;
    if let Value::Record(ref r) = record {
        let m: HashMap<String, Value> = r.clone().into_iter().collect();
        let status = m.get("status");
        let id = m.get("id");
        match (status, id) {
            (Some(Value::String(status)), Some(Value::String(id))) if status == "END" => {
                if let Ok(ts) = id.parse::<u64>() {
                    Ok(Some(ts))
                } else {
                    bail!(
                        "Malformed consistency record, failed to parse timestamp {} in topic {}",
                        id,
                        consistency_topic
                    );
                }
            }
            _ => Ok(None),
        }
    } else {
        Ok(None)
    }
}
fn maybe_decode_consistency_end_record ( bytes : & [ u8 ] , consistency_topic : & str , ) -> Result < Option < Timestamp > , anyhow :: Error > { // The first 5 bytes are reserved for the schema id/schema registry information let mut bytes = & bytes [ 5 .. ] ; let record = mz_avro :: from_avro_datum ( get_debezium_transaction_schema ( ) , & mut bytes ) . context ( " " ) ? ; if let Value :: Record ( ref r ) = record { let m : HashMap < String , Value > = r . clone ( ) . into_iter ( ) . collect ( ) ; let status = m . get ( " " ) ; let id = m . get ( " " ) ; match ( status , id ) { ( Some ( Value :: String ( status ) ) , Some ( Value :: String ( id ) ) ) if status == " " => { if let Ok ( ts ) = id . parse :: < u64 > ( ) { Ok ( Some ( ts ) ) } else { bail ! ( " " id consistency_topic ) ; } } _ => Ok ( None ) , } } else { Ok ( None ) } }
There may be arbitrary messages in this topic that we cannot decode.
There may be arbitrary messages in this topic that we cannot decode .
bytes
&[u8]
consistency_topic
&str
bytes
&[u8]
consistency_topic
&str