[FLINK-13535] Do not abort transactions twice during KafkaProducer startup

• Type: Improvement
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.8.1, 1.9.0
• Fix Version/s: 1.10.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: Nico Kruber
• Reporter: Nico Kruber
• Created: 01/Aug/19 12:37
• Updated: 02/Aug/19 16:43
• Resolved: 02/Aug/19 16:43

Description

During startup of a transactional Kafka producer from previous state, we recover in two steps:

1. in TwoPhaseCommitSinkFunction, we commit pending commit-transactions and abort pending transactions and then call into finishRecoveringContext()
2. in FlinkKafkaProducer#finishRecoveringContext() we iterate over all recovered transaction IDs and abort them.

This may lead to some transactions being worked on twice. Since this is quite some expensive operation, we unnecessarily slow down the job startup but could easily give finishRecoveringContext() a set of transactions that TwoPhaseCommitSinkFunction already covered instead.

Issue Links

• links to: GitHub Pull Request #9323