Mitigate HTTP/2 MYR test on CI

Author: vietj

vertx-core/src/test/java/io/vertx/tests/http/Http2MYRServerTest.java

@@ -13,10 +13,7 @@
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.ByteBuf;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInitializer;
+import io.netty.channel.*;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.http2.AbstractHttp2ConnectionHandlerBuilder;
@@ -44,6 +41,7 @@
 import io.vertx.test.core.Repeat;
 import org.junit.Test;
 
+import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -82,7 +80,7 @@ private void testMYR(boolean multiplexImplementation) throws Exception {
     AtomicInteger inflightRequests = new AtomicInteger();
     AtomicInteger maxInflightRequests = new AtomicInteger();
     AtomicInteger receivedRstFrames = new AtomicInteger();
-    CompletableFuture<Void> goAway = new CompletableFuture<>();
+    CompletableFuture<Boolean> goAway = new CompletableFuture<>();
 
     server.requestHandler(req -> {
       int val = inflightRequests.incrementAndGet();
@@ -154,7 +152,7 @@ public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) th
 
                   @Override
                   public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData) throws Http2Exception {
-                    goAway.complete(null);
+                    goAway.complete(true);
                   }
                 });
                 return super.build();
@@ -164,6 +162,16 @@ public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long error
             Builder clientHandlerBuilder = new Builder();
             Http2ConnectionHandler clientHandler = clientHandlerBuilder.build();
             ch.pipeline().addLast(clientHandler);
+            ch.pipeline().addLast(new ChannelDuplexHandler() {
+              @Override
+              public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+                if (cause instanceof IOException && cause.getMessage().startsWith("Connection reset")) {
+                  goAway.complete(false);
+                } else {
+                  goAway.completeExceptionally(cause);
+                }
+              }
+            });
           }
         };
       }
@@ -192,10 +200,14 @@ public ChannelFuture connect(int port, String host, BiConsumer<ChannelHandlerCon
       chctx.flush();
     }).sync();
 
-    goAway.get(10, TimeUnit.SECONDS);
-
     // Check the number of rst frame received before getting a go away
-    assertEquals(receivedRstFrames.get(), maxRstFramePerWindow + 1);
+    if (goAway.get(20, TimeUnit.SECONDS)) {
+      assertEquals(receivedRstFrames.get(), maxRstFramePerWindow + 1);
+    } else {
+      // Mitigate CI behavior
+      assertTrue(receivedRstFrames.get() < maxRstFramePerWindow + 1);
+    }
+
     assertTrue(maxInflightRequests.get() <= 2 * maxRstFramePerWindow );
   }
 }