Hi Lorenzo, The main failure scenario that recommendation is addressing is when the Flink job fails right after a checkpoint successfully completes, but before the KafkaSink subtasks receive from the JM the checkpoint completed RPC notification to commit the transactions. It is possible that during this window of opportunity, the Kafka txns can be aborted by Kafka due to timeout, leaving you with an inconsistency between Flink and Kafka.
This inconsistency can happen if the checkpoint duration is too close to the configured Kafka transaction timeout, or if the job fails just at the right time during that window of opportunity and remains down long enough for the transaction to timeout. I have a more detailed formulation of this in FLIP-319 [1], at the very end of the proposal in the Appendix section. Thanks, Gordon [1] https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=255071710 On Mon, Oct 2, 2023 at 2:47 PM Lorenzo Nicora <lorenzo.nic...@gmail.com> wrote: > Hi team > > In Kafka Sink docs [1], with EXACTLY_ONCE it is recommended to set: > transaction_timeout > maximum_checkpoint duration + > maximum_restart_duration. > > I understand transaction_timeout > maximum_checkpoint_duration > But why adding maximum_restart_duration? > > If the application recovers from a checkpoint, any uncommitted message > that was written after the last successful checkpoint will be > re-written regardless. > If a transaction times out during the recovery it doesn't matter. > > I would rather say: > transaction_timeout > maximum_checkpoint duration + checkpoint_interval > > Any thoughts? > > Regards > Lorenzo > > [1] > https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/datastream/kafka/#fault-tolerance >
