[flink] 03/04: [FLINK-20145][task] Ignore double notification in StreamTwoInputProcessor.

[ar...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit a76caeaaf3238d2710f94e82c3a1cd7c7a6cb013
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Thu Nov 19 12:57:24 2020 +0100

    [FLINK-20145][task] Ignore double notification in StreamTwoInputProcessor.
    
    Currently, when a priority event is enqueued as the first event, it triggers both a priority and availability notification through respective futures.
    If the priority event is processed through CheckpointedInputGate#processPriorityEvent, the InputProcessor pulls on an empty gate.
    While we could try to avoid the double notification, there are a few race condition that may lead to no notifications and blocked application.
    It's much easier to simply allow double notification and ignore it when they are not useful.
    Note that StreamOneInputProcessor and StreamMultipleInputProcessor are already doing it.
---
 .../flink/streaming/runtime/io/StreamTwoInputProcessor.java       | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java
index c35a24b..ebb9a2c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java
@@ -76,14 +76,14 @@ public final class StreamTwoInputProcessor<IN1, IN2> implements StreamInputProce
 		int readingInputIndex;
 		if (isPrepared) {
 			readingInputIndex = selectNextReadingInputIndex();
-			assert readingInputIndex != InputSelection.NONE_AVAILABLE;
 		} else {
 			// the preparations here are not placed in the constructor because all work in it
 			// must be executed after all operators are opened.
 			readingInputIndex = selectFirstReadingInputIndex();
-			if (readingInputIndex == InputSelection.NONE_AVAILABLE) {
-				return InputStatus.NOTHING_AVAILABLE;
-			}
+		}
+		// In case of double notification (especially with priority notification), there may not be an input after all.
+		if (readingInputIndex == InputSelection.NONE_AVAILABLE) {
+			return InputStatus.NOTHING_AVAILABLE;
 		}
 
 		lastReadInputIndex = readingInputIndex;