[jira] [Updated] (FLINK-31620) ReducingUpsertWriter does not flush the wrapped writer

["Tzu-Li (Gordon) Tai (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-31620?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Tzu-Li (Gordon) Tai updated FLINK-31620:
----------------------------------------
    Fix Version/s: kafka-3.0.0

> ReducingUpsertWriter does not flush the wrapped writer
> ------------------------------------------------------
>
>                 Key: FLINK-31620
>                 URL: https://issues.apache.org/jira/browse/FLINK-31620
>             Project: Flink
>          Issue Type: Bug
>          Components: Connectors / Kafka
>    Affects Versions: 1.17.0, 1.16.1
>            Reporter: Alex Sorokoumov
>            Assignee: Alex Sorokoumov
>            Priority: Critical
>              Labels: pull-request-available
>             Fix For: kafka-3.0.0
>
>
> According to {{SinkWriter#flush}} [javadoc|https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/connector/sink2/SinkWriter.java#L43-L47], 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.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)