about Kafka sink and 2PC function

After reading about FlinkKafkaProducer011 and 2PC function in FLINK, I know, 

when snapshotState(),

• preCommit currentTransaction.
• add <currentTransaction, newTransaction> to the State.

when Checkpoint done and notifyCheckpointComplete(),

• producer will commit currentTransaction to brokers.

when initializeState(),

• restore from State.
• commit currentTransaction and abort newTransaction.

And I have one question, what happens if program fails after notifyCheckpointComplete() done?

As my opinion, when it recovers, it will re-commit what has committed which results duplicate.

---

[hidden email]

[Zhu Zhu]

According to FlinkKafkaProducer011#recoverAndCommit(..), a re-commit of committed transaction would result in an error which will be skipped.

So the duplicated commit would not take effect.

Thanks,

Zhu Zhu

[hidden email] <[hidden email]> 于2019年10月10日周四 下午1:16写道：

After reading about FlinkKafkaProducer011 and 2PC function in FLINK, I know, 

when snapshotState(),

• preCommit currentTransaction.
• add <currentTransaction, newTransaction> to the State.

when Checkpoint done and notifyCheckpointComplete(),

• producer will commit currentTransaction to brokers.

when initializeState(),

• restore from State.
• commit currentTransaction and abort newTransaction.

And I have one question, what happens if program fails after notifyCheckpointComplete() done?

As my opinion, when it recovers, it will re-commit what has committed which results duplicate.

---

[hidden email]