Hello team! I have a scenario which is plaguing me. I am using flink sql. I'm sourcing 2 tables from kafka, using debezium-avro-confluent format. I inner join with a filter (the filter is on purpose for this test). I'm sending the results back to kafka, using debezium-avro-confluent because I want all the cdc messages. Meaning, when a message is no longer in the result set I want to see an op 'd'. When a message is updated, I want to see 2 messages, op 'd' and op 'c'. For the most part, this is working, except i'm finding that it's not working 100% of time. In my current test, I'm updating the filtered field so that results will be removed, and out of the 5 that I expect to emit delete messages I only receive 2 or 3 (it's not consistent). The same for when I'm updating any other fields, I expect to receive 5 delete messages followed by 5 create messages. But most of the time I'll only receive anywhere between 2-4 delete messages, whereas I always receive the create messages. After our data gets sent back to kafka, our client is using clickhouse as their data-warehouse. It doesn't currently support deletes from a stream (meaning someone would have to manually manage the data to delete anything). So, we are attempting to use clickhouse's collapsing merge tree. This is why the messages from flink back to kafka must replicate their cdc exactly. I am grateful for any help you may provide, including hints on how to solve our problem another way.