Hey 🙂 I’m trying to set up a flink pipeline to dedup Kafka events - and send the deduped events to another kafka topic, with exactly once guarantee. Looks like I’m going wrong about the issue and it cannot be done. Can someone explain what I’m doing wrong?

It definitely can be done, what issues / errors are you facing?

@sap1ens I tried several things: 1. Using TableAPI I’m doing a

SELECT DISTINCT

from a Kafka topic, and then write the results to a

upsert-kafka

sink: during recovery all messages since the checkpoint are resubmitted 2. Using TableAPI I’m trying to use the

kafka

sink, but it keeps saying that it

doesn't support consuming update changes which is produced by node GroupAggregate

I need exactly-once guarantee on the outgoing events - but I can’t figure out how to do it.

Have you looked at https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sql/queries/deduplication/#deduplication?

Thanks! I’ll try both 🙂 Since you’ve raised it, I have another question - I’m trying to wrap my head around state, persistence and day 2 operations: can I store key-value state in Flink - regardless of the operator? Is there some guarantee on which pipeline updates will retain operator state and which won’t? Is there some sort of a ‘state migration’ logic I can do when updating a pipeline? Basically I’m trying to figure out if the key space for deduplication will be retained - even if the deduplication logic will change (within reason, obviously)…

btw the deduplication didn’t work either because of the same reason:

kafka

sink

doesn't support consuming update and delete changes which is produced by node Deduplicate

Are you using

upsert-kafka

sink?

no, because it doesn’t guarantee exactly once… Oh, wait. It’s not suppose to emit changes 🤦

Regarding the other questions, might be better to ask them separately in the channel, it’s really hard to give you a succinct answer here.

Thanks! Will do 🙂

ok, so it seems like the Table API is not going to work for your use-case

Having a definitive answer is a way forward too 🙂 In DataStream is there a conventional way to force 2-phase commit?

Yep! https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/datastream/kafka/#fault-tolerance

Thank you so very much for the help and patience!