[jira] [Commented] (FLINK-39639) Split assignment accepted after source finished reading leads to silent data loss

["Spoorthi Basu (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-39639?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18088105#comment-18088105 ] 

Spoorthi Basu commented on FLINK-39639:
---------------------------------------

Hi [~pnowojski] , I reproduced this on current master with the test in the description. The root cause is that {{handleAddSplitsEvent}} has no {{DATA_FINISHED}} guard: a late {{AddSplitEvent}} reaches {{sourceReader.addSplits()}} but is never polled, since {{emitNext}} no longer drives the reader. The coordinator gets no signal and records the assignment as successful, which is the silent loss.

My fix rejects the assignment in {{DATA_FINISHED}} by failing the task. This doesn't lose data: the split is recorded in the assignment tracker before the event is sent ({{{}SourceCoordinatorContext.assignSplits{}}}), so failover returns it via {{subtaskReset}} -> {{getAndRemoveUncheckpointedAssignment}} -> {{addSplitsBack}} and it's reassigned.

It does change one existing test, {{testSourceCheckpointLastUnaligned}} from FLINK-18906: it drives the chained {{MockSource}} to {{END_OF_INPUT}} (no splits, so the reader finishes immediately) and asserts that a later split's records are dropped. A real reader only reaches {{END_OF_INPUT}} after no-more-splits, so that drop looks like the mock surfacing this same bug rather than intended behavior, in which case the test should expect the rejection. Before I change it: is the post-finish drop ever intentional, or is rejecting at {{DATA_FINISHED}} the right call across the board?

Happy to open a PR once that's settled.

> Split assignment accepted after source finished reading leads to silent data loss
> ---------------------------------------------------------------------------------
>
>                 Key: FLINK-39639
>                 URL: https://issues.apache.org/jira/browse/FLINK-39639
>             Project: Flink
>          Issue Type: Bug
>          Components: Runtime / Task
>    Affects Versions: 2.0.1
>            Reporter: Piotr Nowojski
>            Priority: Major
>
> h2. Summary
> When an {{OperatorCoordinator}} sends a split assignment ({{{}AddSplitEvent{}}}) to a {{SourceOperator}} that has concurrently finished reading (reached {{{}END_OF_INPUT{}}}), the event is silently 
> accepted and the splits are added to the {{SourceReader}} — but they will never be processed. The coordinator receives no error signal, so it believes the assignment succeeded. This is silent
> data loss.
> h2. Root Cause
> There is a window between the source operator finishing and the mailbox being quiesced where coordinator events are still accepted and processed:
> {{SourceReader.pollNext()}} returns {{END_OF_INPUT}}
> {{StreamTask.processInput()}} calls {{endData(DRAIN)}} → {{operatorChain.finishOperators()}} → {{SourceOperator.finish()}}
> The mailbox loop suspends, but the mailbox remains in {{OPEN}} state
> A coordinator event (split assignment) arrives via RPC and is queued into the mailbox via {{StreamTask.dispatchOperatorEvent()}}
> {{afterInvoke()}} runs a second mailbox loop ({{{}StreamTask.java{}}} line 1074) that processes the queued event
> {{SourceOperator.handleOperatorEvent()}} calls {{sourceReader.addSplits()}} with no guard on the operator's {{operatingMode}} — the split is accepted even though the operator is in 
> {{DATA_FINISHED}} mode
> The only protection against late events is the {{RejectedExecutionException}} catch in {{{}StreamTask.dispatchOperatorEvent(){}}}, but that only fires after {{mailboxProcessor.prepareClose()}} 
> (line 1085), which happens after the second mailbox loop has already drained.
> h2. Impact
> In batch (bounded) execution, if a {{SplitEnumerator}} assigns splits concurrently with a reader reaching end-of-input, the assigned splits are lost without any error. The enumerator believes 
> the assignment succeeded. This can occur when:
>  - Multiple subtasks finish at different times and the enumerator redistributes work
>  - The enumerator performs lazy/deferred split discovery and assigns splits just as the reader finishes its current work
> h2. Reproducer
> A unit test using {{StreamTaskMailboxTestHarness}} demonstrates the issue without any timing dependencies:
> {code:java}
> /**
>  * Verifies that a coordinator event (split assignment) dispatched after the source has finished
>  * reading (END_OF_INPUT) is silently accepted and the splits are added to the reader, even
>  * though they will never be processed — demonstrating a race condition between coordinator event
>  * delivery and operator lifecycle.
>  */
> @Test
> void testSplitAssignmentAfterSourceFinished() throws Exception {
>     SplitAssignmentTrackingSource testSource =
>             new SplitAssignmentTrackingSource(Boundedness.BOUNDED);
>     SourceOperatorFactory<Integer> sourceOperatorFactory =
>             new SourceOperatorFactory<>(testSource, WatermarkStrategy.noWatermarks());
>     try (StreamTaskMailboxTestHarness<Integer> testHarness =
>             new StreamTaskMailboxTestHarnessBuilder<>(
>                             SourceOperatorStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
>                     .setCollectNetworkEvents()
>                     .setupOutputForSingletonOperatorChain(sourceOperatorFactory, OPERATOR_ID)
>                     .build()) {
>         SplitAssignmentTrackingReader reader =
>                 (SplitAssignmentTrackingReader)
>                         ((SourceOperator) testHarness.getStreamTask().mainOperator)
>                                 .getSourceReader();
>         // Assign an initial split with some records so the source has work to do.
>         MockSourceSplit initialSplit = new MockSourceSplit(0, 0, 2);
>         initialSplit.addRecord(100);
>         initialSplit.addRecord(200);
>         AddSplitEvent<MockSourceSplit> initialAssignment =
>                 new AddSplitEvent<>(
>                         Collections.singletonList(initialSplit),
>                         new MockSourceSplitSerializer());
>         testHarness
>                 .getStreamTask()
>                 .dispatchOperatorEvent(OPERATOR_ID, new SerializedValue<>(initialAssignment));
>         // Process all records. The source will reach END_OF_INPUT, triggering endData() which
>         // calls finish() on the operator. The mailbox loop then suspends.
>         reader.markAvailable();
>         testHarness.processAll();
>         // At this point:
>         // - The source reader has returned END_OF_INPUT
>         // - SourceOperator.finish() has been called
>         // - The mailbox is suspended but still OPEN (not quiesced)
>         assertThat(reader.hasFinishedReading).isTrue();
>         assertThat(reader.splitsAddedAfterFinish).isEmpty();
>         // Now simulate a coordinator sending a late split assignment. This can happen in
>         // practice when the enumerator assigns splits concurrently with the reader finishing.
>         MockSourceSplit lateSplit = new MockSourceSplit(1, 0);
>         lateSplit.addRecord(999);
>         AddSplitEvent<MockSourceSplit> lateAssignment =
>                 new AddSplitEvent<>(
>                         Collections.singletonList(lateSplit),
>                         new MockSourceSplitSerializer());
>         testHarness
>                 .getStreamTask()
>                 .dispatchOperatorEvent(OPERATOR_ID, new SerializedValue<>(lateAssignment));
>         // Drive the task through afterInvoke(), which runs a second mailbox loop that will
>         // process the queued split assignment event.
>         testHarness.finishProcessing();
>         // The late split was accepted by the reader even though the source had already
>         // finished reading. This split will never be processed — it represents lost data.
>         assertThat(reader.splitsAddedAfterFinish)
>                 .as(
>                         "Split assignment was accepted after the source finished reading, "
>                                 + "leading to potential data loss")
>                 .hasSize(1);
>         assertThat(reader.splitsAddedAfterFinish.get(0).splitId()).isEqualTo("1");
>     }
> }
> private static class SplitAssignmentTrackingSource extends MockSource {
>     private static final long serialVersionUID = 1L;
>     SplitAssignmentTrackingSource(Boundedness boundedness) {
>         super(boundedness, 1);
>     }
>     @Override
>     public SourceReader<Integer, MockSourceSplit> createReader(
>             SourceReaderContext readerContext) {
>         return new SplitAssignmentTrackingReader();
>     }
> }
> private static class SplitAssignmentTrackingReader extends MockSourceReader {
>     volatile boolean hasFinishedReading = false;
>     volatile boolean hasClosed = false;
>     final List<MockSourceSplit> splitsAddedAfterFinish = new ArrayList<>();
>     final List<MockSourceSplit> splitsAddedAfterClose = new ArrayList<>();
>     @Override
>     public InputStatus pollNext(ReaderOutput<Integer> sourceOutput) throws Exception {
>         InputStatus status = super.pollNext(sourceOutput);
>         if (status == InputStatus.END_OF_INPUT) {
>             hasFinishedReading = true;
>         }
>         return status;
>     }
>     @Override
>     public void close() throws Exception {
>         hasClosed = true;
>         super.close();
>     }
>     @Override
>     public void addSplits(List<MockSourceSplit> splits) {
>         if (hasFinishedReading) {
>             splitsAddedAfterFinish.addAll(splits);
>         }
>         if (hasClosed) {
>             splitsAddedAfterClose.addAll(splits);
>         }
>         super.addSplits(splits);
>     }
> }{code}
> Full test added in {{{}SourceOperatorStreamTaskTest{}}}.
> h2. Possible Fixes
>  - {{SourceOperator.handleOperatorEvent()}} could check {{operatingMode}} and reject events when in {{DATA_FINISHED}} (or any post-reading state). The coordinator would then receive an error and
> could handle it (e.g., reassign the split to another subtask).
>  - {{StreamTask.dispatchOperatorEvent()}} could check whether operators have been finished before enqueueing the event.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)