[FLINK-6988] Add Apache Kafka 0.11 connector

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.3.1
• Fix Version/s: 1.4.0
• Component/s: Connectors / Kafka
• Assignee: Piotr Nowojski
• Reporter: Piotr Nowojski
• Created: 22/Jun/17 17:08
• Updated: 28/May/21 11:50
• Resolved: 09/Oct/17 17:01

Description

Kafka 0.11 (it will be released very soon) add supports for transactions. Thanks to that, Flink might be able to implement Kafka sink supporting "exactly-once" semantic. API changes and whole transactions support is described in KIP-98.

The goal is to mimic implementation of existing BucketingSink. New FlinkKafkaProducer011 would

• upon creation begin transaction, store transaction identifiers into the state and would write all incoming data to an output Kafka topic using that transaction
• on `snapshotState` call, it would flush the data and write in state information that current transaction is pending to be committed
• on `notifyCheckpointComplete` we would commit this pending transaction
• in case of crash between `snapshotState` and `notifyCheckpointComplete` we either abort this pending transaction (if not every participant successfully saved the snapshot) or restore and commit it.

Issue Links

• causes: FLINK-22800 EXACTLY_ONCE KAFKA consumer (FLINK-6988) has not set isolation.level
• links to: GitHub Pull Request #4239
• links to: GitHub Pull Request #4779