Propagate exceptions properly when netty nio http client protocol configuration fails

Author: zoewangg

.changes/next-release/bugfix-NettyNIOHttpClient-699dfbc.json

@@ -0,0 +1,5 @@
+{
+    "category": "Netty NIO Http Client", 
+    "type": "bugfix", 
+    "description": "Propagate exception properly when an exception is thrown from protocol initialization."
+}

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/http2/HttpOrHttp2ChannelPool.java

@@ -127,18 +127,26 @@ private void failProtocolImplPromise(Throwable e) {
         });
     }
 
-    void completeProtocolConfiguration(Channel newChannel, Protocol protocol) {
+    private void completeProtocolConfiguration(Channel newChannel, Protocol protocol) {
         doInEventLoop(eventLoop, () -> {
             if (closed) {
-                newChannel.close();
-                delegatePool.release(newChannel);
-                protocolImplPromise.setFailure(new IllegalStateException("Pool closed"));
+                closeAndRelease(newChannel, new IllegalStateException("Pool closed"));
             } else {
-                protocolImplPromise.setSuccess(configureProtocol(newChannel, protocol));
+                try {
+                    protocolImplPromise.setSuccess(configureProtocol(newChannel, protocol));
+                } catch (Throwable e) {
+                    closeAndRelease(newChannel, e);
+                }
             }
         });
     }
 
+    private void closeAndRelease(Channel newChannel, Throwable e) {
+        newChannel.close();
+        delegatePool.release(newChannel);
+        protocolImplPromise.setFailure(e);
+    }
+
     private ChannelPool configureProtocol(Channel newChannel, Protocol protocol) {
         if (Protocol.HTTP1_1 == protocol) {
             // For HTTP/1.1 we use a traditional channel pool without multiplexing

http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/NettyNioAsyncHttpClientWireMockTest.java

@@ -129,6 +129,16 @@ public void overrideConnectionIdleTimeout_shouldHonor() {
         }
     }
 
+    @Test
+    public void invalidMaxPendingConnectionAcquireConfig_shouldPropagateException() {
+        try (SdkAsyncHttpClient customClient = NettyNioAsyncHttpClient.builder()
+                                                                 .maxConcurrency(1)
+                                                                 .maxPendingConnectionAcquires(0)
+                                                                 .build()) {
+            assertThatThrownBy(() -> makeSimpleRequest(customClient)).hasMessageContaining("java.lang.IllegalArgumentException: maxPendingAcquires: 0 (expected: >= 1)");
+        }
+    }
+
     @Test
     public void customFactoryIsUsed() throws Exception {
         ThreadFactory threadFactory = spy(new CustomThreadFactory());

http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/http2/HttpOrHttp2ChannelPoolTest.java

@@ -15,13 +15,23 @@
 
 package software.amazon.awssdk.http.nio.netty.internal.http2;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static software.amazon.awssdk.http.SdkHttpConfigurationOption.CONNECTION_ACQUIRE_TIMEOUT;
+import static software.amazon.awssdk.http.SdkHttpConfigurationOption.MAX_PENDING_CONNECTION_ACQUIRES;
+import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.PROTOCOL_FUTURE;
+
 import io.netty.channel.Channel;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.pool.ChannelPool;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.Promise;
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -30,20 +40,10 @@
 import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
 import software.amazon.awssdk.http.Protocol;
+import software.amazon.awssdk.http.nio.netty.internal.MockChannel;
 import software.amazon.awssdk.http.nio.netty.internal.NettyConfiguration;
 import software.amazon.awssdk.utils.AttributeMap;
 
-import java.time.Duration;
-import java.util.concurrent.CompletableFuture;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-import static software.amazon.awssdk.http.SdkHttpConfigurationOption.CONNECTION_ACQUIRE_TIMEOUT;
-import static software.amazon.awssdk.http.SdkHttpConfigurationOption.MAX_PENDING_CONNECTION_ACQUIRES;
-import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.PROTOCOL_FUTURE;
-
 /**
  * Tests for {@link HttpOrHttp2ChannelPool}.
  */
@@ -82,6 +82,34 @@ public void protocolConfigNotStarted_closeSucceeds() {
         httpOrHttp2ChannelPool.close();
     }
 
+    @Test(timeout = 5_000)
+    public void invalidProtocolConfig_shouldFailPromise() throws Exception {
+        HttpOrHttp2ChannelPool invalidChannelPool = new HttpOrHttp2ChannelPool(mockDelegatePool,
+                                                            eventLoopGroup,
+                                                            4,
+                                                            new NettyConfiguration(AttributeMap.builder()
+                                                                                               .put(CONNECTION_ACQUIRE_TIMEOUT, Duration.ofSeconds(1))
+                                                                                               .put(MAX_PENDING_CONNECTION_ACQUIRES, 0)
+                                                                                               .build()));
+
+        Promise<Channel> acquirePromise = eventLoopGroup.next().newPromise();
+        when(mockDelegatePool.acquire()).thenReturn(acquirePromise);
+
+        Thread.sleep(500);
+
+        Channel channel = new MockChannel();
+        eventLoopGroup.register(channel);
+
+        channel.attr(PROTOCOL_FUTURE).set(CompletableFuture.completedFuture(Protocol.HTTP1_1));
+
+        acquirePromise.setSuccess(channel);
+
+        Future<Channel> p = invalidChannelPool.acquire();
+        assertThat(p.await().cause().getMessage()).contains("maxPendingAcquires: 0 (expected: >= 1)");
+        verify(mockDelegatePool).release(channel);
+        assertThat(channel.isOpen()).isFalse();
+    }
+
     @Test
     public void protocolConfigNotStarted_closeClosesDelegatePool() throws InterruptedException {
         httpOrHttp2ChannelPool.close();