Hi Kevin, Perhaps the easiest way to answer your question, is to go through how the exactly-once FlinkKafkaProducer using a 2PC implementation on top of Flink's checkpointing mechanism.
The phases can be broken down as follows (simplified assuming max 1 concurrent checkpoint and that checkpoint completion notifications are never late): 1. BEGIN_TXN: In between each Flink checkpoint, each FlinkKafkaProducer sink operator creates a new Kafka transaction. You can assume that on startup, a new Kafka transaction is created immediately for records that occur before the first checkpoint. 2. PRE_COMMIT: Once a FlinkKafkaProducer sink operator receives Flink's checkpoint barrier, it flushes pending records to the current open transaction, and opens a new one for future records, which belongs to the next checkpoint and thus should be written to the next transaction. Once flushed, the sink operator acknowledges it has completed its checkpoint. 3. COMMIT: Once all sinks acknowledge checkpoint completion, the Flink checkpoint is considered complete (containing state of all operators + consumer offsets). Once that happens, Flink notifies each sink operator of the completion, and only upon receiving this notification, can the sink operator commit the previous transaction. There are some edge cases that is handled, e.g. a checkpoint is considered complete, but before all sinks receive the completion notification and commit their transactions, the job fails (that's why txn ids are written into the checkpoint as well, to make sure all txns belonging to that checkpoint is still eventually committed after restore). The general takeaway is that each parallel sink operator can commit the Kafka transactions only after all participants in the 2PC (i.e. all Flink operators and sinks) acknowledge that they are ready to commit. In Flink terms, the JM is the coordinator, and an operator / sink completing their checkpoint is acknowledging that they are ready for committing. >From an end-to-end point of view, downstream consumers of the output Kafka topic will not see records (assuming they are consuming in Kafka's read.commited mode) until the upstream Flink application sink commits the open Kafka transactions. This boils down to, the read latency for downstream applications is at least the upstream Flink app's checkpoint interval. Hope this helps! Cheers, Gordon On Wed, Mar 10, 2021 at 5:20 PM Kevin Kwon <[email protected]> wrote: > Hi team, I just have a bit of confusion where Two Phase Commit and Kafka's > transaction aware producer using transaction.id and enable.autocommit > plays together > > what I understand of Flink checkpoint (correct me if I'm wrong) is that it > saves the transaction ID as well as the consumer's commit offsets, so when > application fails and restarts, it will reprocess everything from the last > checkpoint and data will be idempotently processed in the Kafka side. > (exactly-once processing rather than exactly-once delivery) > > the question is where does 2 phase commit play a role here? >
