[FLINK-31620] ReducingUpsertWriter does not flush the wrapped writer

• Type: Bug
• Status: Closed
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 1.17.0, 1.16.1
• Fix Version/s: kafka-3.0.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: Alex Sorokoumov
• Reporter: Alex Sorokoumov
• Created: 26/Mar/23 22:33
• Updated: 12/Apr/23 17:11
• Resolved: 12/Apr/23 17:05

Description

According to SinkWriter#flush javadoc, the writer must flush its records to guarantee AT_LEAST_ONCE.

upsert-kafka's ReducingUpsertWriter inserts buffered records into the wrapped writer, but does not flush it:
1. SinkWriter#flush implementation - https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriter.java#L88-L91.
2. The actual flush code - https://github.com/apache/flink/blob/f3c653ed2e4264315ed83a5b4b2494a7dcc41474/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriter.java#L143-L150.

Issue Links

• links to: GitHub Pull Request #17