[
https://issues.apache.org/jira/browse/FLINK-29589?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Krzysztof Chmielewski updated FLINK-29589:
------------------------------------------
Description:
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 for 5 commits)
# 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.
was:
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 for 5 commits)
# 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].
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.
> 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 for 5 commits)
> # 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)
