[
https://issues.apache.org/jira/browse/FLINK-33170?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Robert Hoyt updated FLINK-33170:
--------------------------------
Description:
Possibly related to FLINK-27916.
In all versions since 1.15.x there's a subtle bug in
HybridSourceSplitEnumerator when determining if it's time to move on to the
next source:
{code:java}
finishedReaders.add(subtaskId);
if (finishedReaders.size() == context.currentParallelism()) {
// move on to the next source if it exists
{code}
This snippet is correct, but when changing to the next source,
`finishedReaders` is never cleared. So when processing the second source, the
`finishedReaders.size()` check will return true when the _first_ subtask
finishes. The hybrid source moves on to the next source if one exists, so any
unsent records in other subtasks will get dropped.
Concrete example with three sources, two subtasks each:
# subtask 0 finishes with the first source. `finishedReaders` has size 1
# subtask 1 finishes with the first source. `finishedReaders` has size 2 now,
and moves on to the second source
# subtask 1 finishes with the first source. `finishedReaders.add(1)` doesn't
change the set; `finishedReaders` still has size 2. So the hybrid source moves
on to the third source.
# subtask 0 wasn't finished with the second source, but receives the
notification to move on. Any unsent records are lost. *Data loss.*
# this continues to the last source. The source doesn't change over if at the
last source so the race condition in step 3 never happens
So step 3 results in the race condition that will drop records indeterminately
for all but the first source and last source.
In production this issue caused the loss of GBs to TBs of data depending on the
sources. We fixed it in a private fork by clearing the `finishedReaders` set
when changing to the next source.
was:
Possibly related to FLINK-27916.
In all versions since 1.15.x there's a subtle bug
`HybridSourceSplitEnumerator`'s when determining if it's time to move on to the
next source:
```
finishedReaders.add(subtaskId);
if (finishedReaders.size() == context.currentParallelism()) {
// move on to the next source if it exists
```
This snippet is correct, but when changing to the next source,
`finishedReaders` is never cleared. So when processing the second source, the
`finishedReaders.size()` check will return true when the _first_ subtask
finishes. The hybrid source moves on to the next source if one exists, so any
unsent records in other subtasks will get dropped.
Concrete example with three sources, two subtasks each:
# subtask 0 finishes with the first source. `finishedReaders` has size 1
# subtask 1 finishes with the first source. `finishedReaders` has size 2 now,
and moves on to the second source
# subtask 1 finishes with the first source. `finishedReaders.add(1)` doesn't
change the set; `finishedReaders` still has size 2. So the hybrid source moves
on to the third source.
# subtask 0 wasn't finished with the second source, but receives the
notification to move on. Any unsent records are lost. *Data loss.*
# this continues to the last source. The source doesn't change over if at the
last source so the race condition in step 3 never happens
So step 3 results in the race condition that will drop records indeterminately
for all but the first source and last source.
In production this issue caused the loss of GBs to TBs of data depending on the
sources. We fixed it in a private fork by clearing the `finishedReaders` set
when changing to the next source.
> HybridSourceSplitEnumerator causes dropped records in Hybrid Sources with 3+
> sources
> ------------------------------------------------------------------------------------
>
> Key: FLINK-33170
> URL: https://issues.apache.org/jira/browse/FLINK-33170
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Common
> Affects Versions: 1.15.4, 1.16.2, 1.17.1, 1.19.0, 1.18.1
> Reporter: Robert Hoyt
> Priority: Critical
> Original Estimate: 24h
> Remaining Estimate: 24h
>
> Possibly related to FLINK-27916.
>
> In all versions since 1.15.x there's a subtle bug in
> HybridSourceSplitEnumerator when determining if it's time to move on to the
> next source:
> {code:java}
> finishedReaders.add(subtaskId);
> if (finishedReaders.size() == context.currentParallelism()) {
> // move on to the next source if it exists
> {code}
> This snippet is correct, but when changing to the next source,
> `finishedReaders` is never cleared. So when processing the second source, the
> `finishedReaders.size()` check will return true when the _first_ subtask
> finishes. The hybrid source moves on to the next source if one exists, so any
> unsent records in other subtasks will get dropped.
>
> Concrete example with three sources, two subtasks each:
> # subtask 0 finishes with the first source. `finishedReaders` has size 1
> # subtask 1 finishes with the first source. `finishedReaders` has size 2
> now, and moves on to the second source
> # subtask 1 finishes with the first source. `finishedReaders.add(1)` doesn't
> change the set; `finishedReaders` still has size 2. So the hybrid source
> moves on to the third source.
> # subtask 0 wasn't finished with the second source, but receives the
> notification to move on. Any unsent records are lost. *Data loss.*
> # this continues to the last source. The source doesn't change over if at
> the last source so the race condition in step 3 never happens
>
> So step 3 results in the race condition that will drop records
> indeterminately for all but the first source and last source.
> In production this issue caused the loss of GBs to TBs of data depending on
> the sources. We fixed it in a private fork by clearing the `finishedReaders`
> set when changing to the next source.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
