[ https://issues.apache.org/jira/browse/FLINK-29589?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17616307#comment-17616307 ]
Krzysztof Chmielewski commented on FLINK-29589: ----------------------------------------------- Hi [~chesnay] V2 on 1.15, 1.16 and 1.17 has its own issues that we have found and actually we are working to fix them with Fabian Paul. https://issues.apache.org/jira/browse/FLINK-29509 https://issues.apache.org/jira/browse/FLINK-29583 https://issues.apache.org/jira/browse/FLINK-29512 With those stil on the plate we cant really tell if there is a data loss on V2 since Task manager is failing to start during recovery when running Sink with global committer. > 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 > Reporter: Krzysztof Chmielewski > Priority: Blocker > > 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 build 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. > The test reproduces the issue in more than 90% of runs. > 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)