[FLINK-22902] Port KafkaSink to new Unified Sink API (FLIP-143)

• Type: Improvement
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.14.0
• Fix Version/s: 1.14.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available

Release Note

`KafkaSink` supersedes `FlinkKafkaProducer` and provides efficient exactly-once and at-least-once writing with the new unified sink interface supporting both batch and streaming mode of DataStream API. To upgrade, please stop with savepoint. `KafkaSink` needs a user-configured and unique transaction prefix, such that transactions of different applications do not interfere with each other.

• Assignee: Fabian Paul
• Reporter: Arvid Heise
• Created: 07/Jun/21 12:43
• Updated: 04/Jul/22 07:03
• Resolved: 10/Sep/21 13:21

Description

As one of the most commonly connectors, Kafka sink should be ported to the new interfaces as quickly as possible such that Flink would support batch queries on Kafka.

The implementation should probably follow the current implementation as close as possible.

Issue Links

• relates to: FLINK-28370 Add close method for KafkaRecordSerializationSchema
• links to: GitHub Pull Request #16676

Sub-Tasks

• Implement at-least-once Kafka Sink: Resolved
• Implement exactly-once Kafka Sink: Resolved
• Migrate Table API to new KafkaSink: Resolved
• Create a KafkaRecordSerializationSchemas valueOnly helper: Resolved
• Write documentation for new KafkaSink: Resolved
• Move sink to org.apache.kafka.conntor.kafka.sink package: Resolved
• Migrate BufferedUpsertSinkFunction to FLIP-143: Resolved
• Test FLIP-143 KafkaSink: Closed
• Add FLIP-33 metrics to new KafkaSink: Resolved