[flink] 02/04: [FLINK-20145][network] Double-check if gate still has priority buffer when enqueuing in UnionInputGate.

[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 0d6eb6d3dc4f57be6de1310f7648fc3716cac190
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Thu Oct 8 14:25:17 2020 +0200

    [FLINK-20145][network] Double-check if gate still has priority buffer when enqueuing in UnionInputGate.
    
    Since notification is not atomic in respect to gate enqueuing, priority event already polled by task thread when netty enqueues the gate.
---
 .../flink/runtime/io/network/partition/consumer/UnionInputGate.java | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java
index ee08164..2a7f940 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java
@@ -314,6 +314,12 @@ public class UnionInputGate extends InputGate {
 					return;
 				}
 
+				if (priority && !inputGate.getPriorityEventAvailableFuture().isDone()) {
+					// Since notification is not atomic in respect to gate enqueuing, priority event may already be polled by
+					// task thread when netty enqueues the gate, so just ignore the notification.
+					return;
+				}
+
 				inputGatesWithData.add(inputGate, priority, alreadyEnqueued);
 
 				if (priority && inputGatesWithData.getNumPriorityElements() == 1) {