add a simple reproducer

jrhee17 · Nov 1, 2021 · 225a684 · 225a684
1 parent 4682d53
commit 225a684
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 1 deletion.
diff --git a/core/src/main/java/com/linecorp/armeria/client/HttpClientPipelineConfigurator.java b/core/src/main/java/com/linecorp/armeria/client/HttpClientPipelineConfigurator.java
@@ -98,6 +98,7 @@
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslHandler;
 import io.netty.handler.ssl.SslHandshakeCompletionEvent;
+import io.netty.handler.traffic.ChannelTrafficShapingHandler;
 import io.netty.util.AsciiString;
 import io.netty.util.ReferenceCountUtil;
 
@@ -164,6 +165,9 @@ public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, Sock
         final ChannelPipeline p = ch.pipeline();
         p.addLast(new FlushConsolidationHandler());
         p.addLast(ReadSuppressingAndChannelDeactivatingHandler.INSTANCE);
+        // throttle bytes written s.t. http2 preface is sent (so channel becomes active),
+        // but subsequent writes fail
+        p.addLast(new ChannelTrafficShapingHandler(30, 0));
 
         try {
             if (sslCtx != null) {

diff --git a/core/src/test/java/com/linecorp/armeria/client/HttpClientMaxConcurrentStreamTest.java b/core/src/test/java/com/linecorp/armeria/client/HttpClientMaxConcurrentStreamTest.java
@@ -16,6 +16,7 @@
 package com.linecorp.armeria.client;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.assertj.core.api.Assertions.catchThrowable;
 import static org.awaitility.Awaitility.await;
 
@@ -36,9 +37,13 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
+import com.google.common.base.Strings;
+
 import com.linecorp.armeria.common.AggregatedHttpResponse;
 import com.linecorp.armeria.common.ClosedSessionException;
+import com.linecorp.armeria.common.HttpMethod;
 import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.common.RequestHeaders;
 import com.linecorp.armeria.common.SessionProtocol;
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.logging.ClientConnectionTimings;
@@ -69,6 +74,7 @@ protected void configure(ServerBuilder sb) throws Exception {
                 responses.add(f);
                 return HttpResponse.from(f);
             });
+
             sb.http2MaxStreamsPerConnection(MAX_CONCURRENT_STREAMS);
             sb.maxNumConnections(MAX_NUM_CONNECTIONS);
             sb.idleTimeoutMillis(3000);
@@ -137,7 +143,7 @@ void tearDown() throws Exception {
         }
 
         if (clientFactory != null) {
-            clientFactory.close();
+            clientFactory.closeAsync().get(10, TimeUnit.SECONDS);
         }
 
         await().until(() -> server.server().numConnections() == 0);
@@ -404,4 +410,27 @@ public void connectionClosed(SessionProtocol protocol, InetSocketAddress remoteA
             }
         };
     }
+
+    @Test
+    void testMaxConcurrentStreamsViolatedFromWrite() throws Exception {
+        connectionPoolListener = ConnectionPoolListener.noop();
+
+        final String requestBody = Strings.repeat("12345", 1000);
+        final WebClient webClient = WebClient.builder(server.uri(SessionProtocol.H2C))
+                                             .writeTimeoutMillis(1)
+                                             .factory(clientFactory)
+                                             .build();
+        final RequestHeaders requestHeaders = RequestHeaders.builder()
+                                                            .method(HttpMethod.GET).path(PATH).build();
+
+        // queue requests which occupies all available streams
+        for (int i = 0; i < MAX_CONCURRENT_STREAMS; i++) {
+            assertThatThrownBy(() -> webClient.execute(requestHeaders, requestBody).aggregate().join())
+                    .hasRootCauseInstanceOf(WriteTimeoutException.class);
+        }
+
+        // this request should not throw a http2 exception
+        assertThatThrownBy(() -> webClient.execute(requestHeaders, requestBody).aggregate().join())
+                .hasRootCauseInstanceOf(WriteTimeoutException.class);
+    }
 }