[FLINK-23839] Unclear severity of Kafka transaction recommit warning in logs

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.11.4, 1.12.5, 1.13.2
• Fix Version/s: 1.14.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: David Morávek
• Reporter: Nico Kruber
• Created: 17/Aug/21 15:09
• Updated: 06/Sep/21 15:06
• Resolved: 06/Sep/21 10:46

Description

In a transactional Kafka sink, after recovery, all transactions from the recovered checkpoint are recommitted even though they may have already been committed before because this is not part of the checkpoint.

This second commit can lead to a number of WARN entries in the logs coming from KafkaCommitter or FlinkKafkaProducer.

Examples:

WARN [org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer] ... Encountered error org.apache.kafka.common.errors.InvalidTxnStateException: The producer attempted a transactional operation in an invalid state. while recovering transaction KafkaTransactionState [transactionalId=..., producerId=12345, epoch=123]. Presumably this transaction has been already committed before.

WARN [org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer] ... Encountered error org.apache.kafka.common.errors.ProducerFencedException: Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId, or the producer's transaction has been expired by the broker. while recovering transaction KafkaTransactionState [transactionalId=..., producerId=12345, epoch=12345]. Presumably this transaction has been already committed before

It sounds to me like the second exception is useful and indicates that the transaction timeout is too short. The first exception, however, seems superfluous and rather alerts the user more than it helps. Or what would you do with it?

Can we instead filter out superfluous exceptions and at least put these onto DEBUG logs instead?

Issue Links

• relates to: FLINK-23896 The new KafkaSink drops data if job fails between checkpoint and transaction commit.
• links to: GitHub Pull Request #16941