Robert Hoyt created FLINK-33170: ----------------------------------- Summary: 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.17.1, 1.16.2, 1.15.4, 1.19.0, 1.18.1 Reporter: Robert Hoyt
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. -- This message was sent by Atlassian Jira (v8.20.10#820010)