[FLINK-23025] sink-buffer-max-rows and sink-buffer-flush-interval options produce a lot of duplicates

• Type: Bug
• Status: Closed
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 1.13.1
• Fix Version/s: 1.14.0, 1.13.2
• Component/s: Connectors / Kafka, Table SQL / Ecosystem
• Labels: pull-request-available
• Assignee: Shengkai Fang
• Reporter: Joe Moser
• Created: 17/Jun/21 18:31
• Updated: 24/Jun/21 12:23
• Resolved: 24/Jun/21 12:23

Description

Using the sink-buffer-flush-max-rows and sink-buffer-flush-interval options for a kafka sink produces a lot of duplicate key/values in the target kafka topic. Maybe the BufferedUpsertSinkFunction should clone the buffered key/value RowData objects, but it doesn’t. Seems like in line 134 the condition should be negated or the ternary operator results swapped:

this.valueCopier =
 getRuntimeContext().getExecutionConfig().isObjectReuseEnabled()
 ? Function.identity()
 : typeSerializer::copy;

(in the jdbc sink the same logic is done but the ternary operator results swapped)

 

Issue Links

• links to: GitHub Pull Request #16242
• links to: GitHub Pull Request #16257