Krzysztof Chmielewski created FLINK-29589:
---------------------------------------------
Summary: 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.6, 1.14.5, 1.14.4, 1.14.3, 1.14.2, 1.14.0
Reporter: Krzysztof Chmielewski
Flink's Sink's 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 for sinks.
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:
1. Streaming source emitting constant number of events per checkpoint (20
events for 5 commits)
2. Sink with parallelism > 1 with committer and GlobalCommitter elements.
3. Commitaers processed committables for checkpointId 2.
3. 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).
4. Streaming source ends
5. 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 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].
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)
