[flink] branch release-1.13 updated: [FLINK-22085][tests] Update NetworkFailureHandler to prevent closeOnFlush() from being called recursively

[dw...@apache.org]

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

dwysakowicz pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.13 by this push:
     new 766f717  [FLINK-22085][tests] Update NetworkFailureHandler to prevent closeOnFlush() from being called recursively
766f717 is described below

commit 766f7175d62f5ef572027331e3fceb31b1b4cc4b
Author: Dong Lin <li...@gmail.com>
AuthorDate: Thu Apr 15 22:14:27 2021 +0800

    [FLINK-22085][tests] Update NetworkFailureHandler to prevent closeOnFlush() from being called recursively
---
 .../flink/networking/NetworkFailureHandler.java     | 21 ++++++++++++++++++---
 1 file changed, 18 insertions(+), 3 deletions(-)

diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/networking/NetworkFailureHandler.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/networking/NetworkFailureHandler.java
index b764440..f2de546 100644
--- a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/networking/NetworkFailureHandler.java
+++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/networking/NetworkFailureHandler.java
@@ -35,6 +35,7 @@ import org.slf4j.LoggerFactory;
 
 import java.net.InetSocketAddress;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
@@ -55,8 +56,21 @@ class NetworkFailureHandler extends SimpleChannelUpstreamHandler {
     private final ClientSocketChannelFactory channelFactory;
     private final String remoteHost;
     private final int remotePort;
-
     private final AtomicBoolean blocked;
+    // The set of channels that are being closed in closeOnFlush(). This is needed to avoid
+    // closing a channel recursively. See FLINK-22085 for more detail.
+    private static final Set<Channel> channelsBeingClosed = ConcurrentHashMap.newKeySet();
+
+    private static final ChannelFutureListener CLOSE_WITH_BOOKKEEPING =
+            new ChannelFutureListener() {
+                public void operationComplete(ChannelFuture future) {
+                    future.getChannel()
+                            .close()
+                            .addListener(
+                                    channelFuture ->
+                                            channelsBeingClosed.remove(channelFuture.getChannel()));
+                }
+            };
 
     public NetworkFailureHandler(
             AtomicBoolean blocked,
@@ -73,8 +87,9 @@ class NetworkFailureHandler extends SimpleChannelUpstreamHandler {
 
     /** Closes the specified channel after all queued write requests are flushed. */
     static void closeOnFlush(Channel channel) {
-        if (channel.isConnected()) {
-            channel.write(ChannelBuffers.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
+        if (channel.isConnected() && !channelsBeingClosed.contains(channel)) {
+            channelsBeingClosed.add(channel);
+            channel.write(ChannelBuffers.EMPTY_BUFFER).addListener(CLOSE_WITH_BOOKKEEPING);
         }
     }