This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 063b46ef668 [FLINK-40011][table-planner] Fix NoMoreSplitsEvent race in
CommonExecSinkITCase
063b46ef668 is described below
commit 063b46ef66843868bde7de9f9366ffff3e53055e
Author: Martijn Visser <[email protected]>
AuthorDate: Mon Jun 29 11:19:48 2026 +0200
[FLINK-40011][table-planner] Fix NoMoreSplitsEvent race in
CommonExecSinkITCase
The TestSource reader in CommonExecSinkITCase returned END_OF_INPUT as soon
as
it had emitted its rows. The SingleSplitEnumerator assigns the split and
then,
through the coordinator thread, delivers a NoMoreSplitsEvent as a separate
event. If the split-holding subtask finished before that event was
delivered,
the coordinator sent it to an already-FINISHED task. The framework treats
the
undeliverable event as a lost OperatorEvent and triggers a failover, which
the
job's NoRestartBackoffTimeStrategy turns into a hard failure (observed
once, in
build 76431, on subtask (1/4), the split holder).
Defer END_OF_INPUT until notifyNoMoreSplits() has been received so the
reader
cannot finish before the event lands, mirroring how
SourceReaderBase.finishedOrAvailableLater() defers completion until
no-more-splits is set. This is the chosen approach for this test, not a
general
source-reader requirement. No emitted data or assertion changes.
Generated-by: Claude Opus 4.8 (1M context)
---
.../planner/plan/nodes/exec/common/CommonExecSinkITCase.java | 11 ++++++++---
1 file changed, 8 insertions(+), 3 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSinkITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSinkITCase.java
index a909761b51a..e7f3c51ea5f 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSinkITCase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSinkITCase.java
@@ -626,7 +626,8 @@ class CommonExecSinkITCase {
@Override
public void notifyNoMoreSplits() {
- // Called for readers that never get a
split → they can finish
+ // Gates completion for every reader,
including the split
+ // holder, so pollNext only finishes once
this signal arrives.
noMoreSplits = true;
}
@@ -638,9 +639,13 @@ class CommonExecSinkITCase {
out.collect(
(RowData)
converter.toInternal(r)));
emitted = true;
- return InputStatus.END_OF_INPUT;
}
- if (!hasSplit && noMoreSplits) {
+ // Finish only after the NoMoreSplitsEvent
has arrived.
+ // Returning END_OF_INPUT earlier lets the
task reach FINISHED
+ // before the coordinator delivers that
event, which surfaces as
+ // a lost OperatorEvent and fails the job
under
+ // NoRestartBackoffTimeStrategy.
+ if (noMoreSplits) {
return InputStatus.END_OF_INPUT;
}
return InputStatus.NOTHING_AVAILABLE;
