This is an automated email from the ASF dual-hosted git repository. 1996fanrui pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4aff72938d631b5478ac592631155cac92fe58bd Author: Roman Khachatryan <[email protected]> AuthorDate: Tue Jun 30 01:42:53 2026 +0200 [hotfix] Report END_OF_INPUT from multiple-input processor when all inputs finished StreamMultipleInputProcessor.processInput returned NOTHING_AVAILABLE when no input was selectable, even once all inputs were finished. Since getAvailableFuture() returns AVAILABLE as soon as all inputs are finished, the mailbox loop then never blocks and would spin on processInput at 100% CPU if it were ever invoked again after the operator finished. This is a latent inconsistency with single-input processors (which keep returning END_OF_INPUT once finished) and is independent of any particular feature; it is just not reached today because a task suspends on the first END_OF_INPUT and processInput is not called again. The checkpointing-during- recovery work (later in this series) does re-invoke processInput after an operator finishes during recovery, which would otherwise turn this into a hang. Report END_OF_INPUT when there is nothing to read and all inputs are finished, consistent with single-input processors and MultipleInputSelectionHandler#calculateOverallStatus. --- .../streaming/runtime/io/StreamMultipleInputProcessor.java | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessor.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessor.java index 565447b25fd..a5ef79bc3be 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessor.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessor.java @@ -79,7 +79,17 @@ public final class StreamMultipleInputProcessor implements StreamInputProcessor readingInputIndex = selectFirstReadingInputIndex(); } if (readingInputIndex == InputSelection.NONE_AVAILABLE) { - return DataInputStatus.NOTHING_AVAILABLE; + // If all inputs are already finished, there is nothing left to read: report + // END_OF_INPUT rather than NOTHING_AVAILABLE. Otherwise, because getAvailableFuture() + // returns AVAILABLE once all inputs are finished, the mailbox loop would never block + // and + // would spin on processInput. This matters when processInput is invoked again after the + // operator already finished -- e.g. a task that reached END_OF_INPUT during recovery + // and + // is resumed once recovery completes (see StreamTask#processInput). + return inputSelectionHandler.areAllInputsFinished() + ? DataInputStatus.END_OF_INPUT + : DataInputStatus.NOTHING_AVAILABLE; } lastReadInputIndex = readingInputIndex;
