[jira] [Commented] (FLINK-29589) Data Loss in Sink GlobalCommitter during Task Manager recovery

["Martijn Visser (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-29589?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17616076#comment-17616076 ] 

Martijn Visser commented on FLINK-29589:
----------------------------------------

If there's indeed a reproducible case with data loss I think we should consider making this a blocker

> Data Loss in Sink GlobalCommitter during Task Manager recovery
> --------------------------------------------------------------
>
>                 Key: FLINK-29589
>                 URL: https://issues.apache.org/jira/browse/FLINK-29589
>             Project: Flink
>          Issue Type: Bug
>    Affects Versions: 1.14.0, 1.14.2, 1.14.3, 1.14.4, 1.14.5, 1.14.6
>            Reporter: Krzysztof Chmielewski
>            Priority: Major
>
> Flink's Sink architecture with global committer seems to be vulnerable for data loss during Task Manager recovery. The entire checkpoint can be lost by GlobalCommitter resulting with data loss.
> Issue was observed in Delta Sink connector on a real 1.14.x cluster and was replicated using Flink's 1.14.6 Test Utils classes.
> Scenario:
>  #  Streaming source emitting constant number of events per checkpoint (20 events per commit for 5 commits in total, that gives 100 records).
>  #  Sink with parallelism > 1 with committer and GlobalCommitter elements.
>  #  Commiters processed committables for checkpointId 2.
>  #  GlobalCommitter throws exception (desired exception) during checkpointId 2 (third commit) while processing data from checkpoint 1 (it is expected to global committer architecture lag one commit behind in reference to rest of the pipeline).
>  # Task Manager recovery, source resumes sending data.
>  # Streaming source ends.
>  #  We are missing 20 records (one checkpoint).
> What is happening is that during recovery, committers are performing "retry" on committables for checkpointId 2, however those committables, reprocessed from "retry" task are not emit downstream to the global committer. 
> The issue can be reproduced using Junit Test builded with Flink's TestSink.
> The test was implemented [here|https://github.com/kristoffSC/flink/blob/Flink_1.14_DataLoss_SinkGlobalCommitter/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkITCase.java#:~:text=testGlobalCommitterMissingRecordsDuringRecovery] and it is based on other tests from SinkITCase.java class.
> I believe that problem is somewhere around `SinkOperator::notifyCheckpointComplete` method. In there we see that Retry async task is scheduled however its result is never emitted downstream like it is done for regular flow one line above.



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