refactor code and address feedback

Author: zoewangg

.changes/next-release/bugfix-AWSCRTHTTPClient-aee08c2.json

@@ -2,5 +2,5 @@
     "type": "bugfix",
     "category": "AWS CRT HTTP Client",
     "contributor": "",
-    "description": "Fixed a potential deadlock in `AwsCrtHttpClient` that could occur when the request body `InputStream` blocked waiting for data on the CRT event loop thread. This could happen when a blocking stream (e.g., a `BufferedInputStream` wrapping a `ResponseInputStream`) was used as a request body and the read depended on the same event loop thread to deliver data. Request body writing now happens on the caller thread."
+    "description": "Fixed a potential deadlock in both `AwsCrtHttpClient` and `AwsCrtAsyncHttpClient` that could occur when the request body source delivered data on the CRT event loop thread. For sync, this happened when a blocking `InputStream` (e.g., a `BufferedInputStream` wrapping a `ResponseInputStream`) was used as a request body. For async, this happened when the user-supplied `Publisher<ByteBuffer>` scheduled `onNext` back onto the same event loop. Request body data is now pushed via the CRT push-based `writeData` API: sync writes from the caller thread, async writes from a Reactive Streams subscriber outside the event loop."
 }

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtHttpClient.java

@@ -69,7 +69,7 @@ private AwsCrtHttpClient(DefaultBuilder builder, AttributeMap config) {
         }
     }
 
-    public static AwsCrtHttpClient.Builder builder() {
+    public static Builder builder() {
         return new DefaultBuilder();
     }
 
@@ -124,45 +124,53 @@ private CrtHttpRequest(CrtRequestContext context) {
         @Override
         public HttpExecuteResponse call() throws IOException {
             HttpExecuteResponse.Builder builder = HttpExecuteResponse.builder();
+            CrtRequestExecutor.Result result = new CrtRequestExecutor().execute(context);
+            responseFuture = result.responseFuture();
 
             try {
-                CrtRequestExecutor.Result result = new CrtRequestExecutor().execute(context);
-                responseFuture = result.responseFuture();
                 writeRequestBody(result.streamHandler());
 
                 SdkHttpFullResponse response = CompletableFutureUtils.joinInterruptibly(responseFuture);
                 builder.response(response);
                 builder.responseBody(response.content().orElse(null));
                 return builder.build();
-            } catch (CompletionException e) {
-                Throwable cause = e.getCause();
-
-                // Complete the future exceptionally to trigger connection cleanup in the response handler.
-                // Handles thread-interrupt case where joinInterruptibly throws due to
-                // InterruptedException. Without this, the
-                // Ensures that closeConnection() is invoked to prevent leaking the connection from the pool.
-                if (responseFuture != null) {
-                    responseFuture.completeExceptionally(cause != null ? cause : e);
-                }
-
-                if (cause instanceof IOException) {
-                    throw (IOException) cause;
-                }
-
-                if (cause instanceof HttpException) {
-                    Throwable wrapped = CrtUtils.wrapCrtException(cause);
-                    if (wrapped instanceof IOException) {
-                        throw (IOException) wrapped;
-                    }
-                    throw (HttpException) cause;
-                }
+            } catch (Throwable t) {
+                // CompletionException is the wrapper from joinInterruptibly; direct throws
+                // (e.g., IOException from inputStream.read in writeRequestBody) arrive unwrapped.
+                Throwable cause = (t instanceof CompletionException && t.getCause() != null) ? t.getCause() : t;
+
+                // Tear down the stream so the connection is not leaked back to the pool.
+                // closeConnection() is idempotent and a no-op if the stream is not yet acquired
+                // or is already closed.
+                result.streamHandler().closeConnection();
+                responseFuture.completeExceptionally(cause);
+
+                throw mapToIoExceptionOrRethrow(cause);
+            }
+        }
 
-                if (cause instanceof InterruptedException) {
-                    Thread.currentThread().interrupt();
-                    throw new IOException("Request was cancelled", cause);
+        private static IOException mapToIoExceptionOrRethrow(Throwable cause) {
+            if (cause instanceof IOException) {
+                return (IOException) cause;
+            }
+            if (cause instanceof HttpException) {
+                Throwable wrapped = CrtUtils.wrapCrtException(cause);
+                if (wrapped instanceof IOException) {
+                    return (IOException) wrapped;
                 }
-                throw new RuntimeException(e.getCause());
+                throw (HttpException) cause;
+            }
+            if (cause instanceof InterruptedException) {
+                Thread.currentThread().interrupt();
+                return new IOException("Request was cancelled", cause);
+            }
+            if (cause instanceof RuntimeException) {
+                throw (RuntimeException) cause;
+            }
+            if (cause instanceof Error) {
+                throw (Error) cause;
             }
+            return new IOException(cause);
         }
 
         private void writeRequestBody(CrtStreamHandler streamHandler) throws IOException {
@@ -194,14 +202,14 @@ public void abort() {
     /**
      * Builder that allows configuration of the AWS CRT HTTP implementation.
      */
-    public interface Builder extends SdkHttpClient.Builder<AwsCrtHttpClient.Builder> {
+    public interface Builder extends SdkHttpClient.Builder<Builder> {
 
         /**
          * The Maximum number of allowed concurrent requests. For HTTP/1.1 this is the same as max connections.
          * @param maxConcurrency maximum concurrency per endpoint
          * @return The builder of the method chaining.
          */
-        AwsCrtHttpClient.Builder maxConcurrency(Integer maxConcurrency);
+        Builder maxConcurrency(Integer maxConcurrency);
 
         /**
          * Configures the number of unread bytes that can be buffered in the
@@ -211,22 +219,22 @@ public interface Builder extends SdkHttpClient.Builder<AwsCrtHttpClient.Builder>
          * @param readBufferSize The number of bytes that can be buffered.
          * @return The builder of the method chaining.
          */
-        AwsCrtHttpClient.Builder readBufferSizeInBytes(Long readBufferSize);
+        Builder readBufferSizeInBytes(Long readBufferSize);
 
         /**
          * Sets the http proxy configuration to use for this client.
          * @param proxyConfiguration The http proxy configuration to use
          * @return The builder of the method chaining.
          */
-        AwsCrtHttpClient.Builder proxyConfiguration(ProxyConfiguration proxyConfiguration);
+        Builder proxyConfiguration(ProxyConfiguration proxyConfiguration);
 
         /**
          * Sets the http proxy configuration to use for this client.
          *
          * @param proxyConfigurationBuilderConsumer The consumer of the proxy configuration builder object.
          * @return the builder for method chaining.
          */
-        AwsCrtHttpClient.Builder proxyConfiguration(Consumer<ProxyConfiguration.Builder> proxyConfigurationBuilderConsumer);
+        Builder proxyConfiguration(Consumer<ProxyConfiguration.Builder> proxyConfigurationBuilderConsumer);
 
         /**
          * Configure the health checks for all connections established by this client.
@@ -246,7 +254,7 @@ public interface Builder extends SdkHttpClient.Builder<AwsCrtHttpClient.Builder>
          * @param healthChecksConfiguration The health checks config to use
          * @return The builder of the method chaining.
          */
-        AwsCrtHttpClient.Builder connectionHealthConfiguration(ConnectionHealthConfiguration healthChecksConfiguration);
+        Builder connectionHealthConfiguration(ConnectionHealthConfiguration healthChecksConfiguration);
 
         /**
          * A convenience method that creates an instance of the {@link ConnectionHealthConfiguration} builder, avoiding the
@@ -256,29 +264,29 @@ public interface Builder extends SdkHttpClient.Builder<AwsCrtHttpClient.Builder>
          * @return The builder of the method chaining.
          * @see #connectionHealthConfiguration(ConnectionHealthConfiguration)
          */
-        AwsCrtHttpClient.Builder connectionHealthConfiguration(Consumer<ConnectionHealthConfiguration.Builder>
+        Builder connectionHealthConfiguration(Consumer<ConnectionHealthConfiguration.Builder>
                                                                         healthChecksConfigurationBuilder);
 
         /**
          * Configure the maximum amount of time that a connection should be allowed to remain open while idle.
          * @param connectionMaxIdleTime the maximum amount of connection idle time
          * @return The builder of the method chaining.
          */
-        AwsCrtHttpClient.Builder connectionMaxIdleTime(Duration connectionMaxIdleTime);
+        Builder connectionMaxIdleTime(Duration connectionMaxIdleTime);
 
         /**
          * The amount of time to wait when initially establishing a connection before giving up and timing out.
          * @param connectionTimeout timeout
          * @return The builder of the method chaining.
          */
-        AwsCrtHttpClient.Builder connectionTimeout(Duration connectionTimeout);
+        Builder connectionTimeout(Duration connectionTimeout);
 
         /**
          * The amount of time to wait when acquiring a connection from the pool before giving up and timing out.
          * @param connectionAcquisitionTimeout the timeout duration
          * @return this builder for method chaining.
          */
-        AwsCrtHttpClient.Builder connectionAcquisitionTimeout(Duration connectionAcquisitionTimeout);
+        Builder connectionAcquisitionTimeout(Duration connectionAcquisitionTimeout);
 
         /**
          * Configure whether to enable {@code tcpKeepAlive} and relevant configuration for all connections established by this
@@ -292,7 +300,7 @@ AwsCrtHttpClient.Builder connectionHealthConfiguration(Consumer<ConnectionHealth
          * @param tcpKeepAliveConfiguration The TCP keep-alive configuration to use
          * @return The builder of the method chaining.
          */
-        AwsCrtHttpClient.Builder tcpKeepAliveConfiguration(TcpKeepAliveConfiguration tcpKeepAliveConfiguration);
+        Builder tcpKeepAliveConfiguration(TcpKeepAliveConfiguration tcpKeepAliveConfiguration);
 
         /**
          * Configure whether to enable {@code tcpKeepAlive} and relevant configuration for all connections established by this
@@ -306,7 +314,7 @@ AwsCrtHttpClient.Builder connectionHealthConfiguration(Consumer<ConnectionHealth
          * @return The builder of the method chaining.
          * @see #tcpKeepAliveConfiguration(TcpKeepAliveConfiguration)
          */
-        AwsCrtHttpClient.Builder tcpKeepAliveConfiguration(Consumer<TcpKeepAliveConfiguration.Builder>
+        Builder tcpKeepAliveConfiguration(Consumer<TcpKeepAliveConfiguration.Builder>
                                                                     tcpKeepAliveConfigurationBuilder);
 
         /**
@@ -325,15 +333,15 @@ AwsCrtHttpClient.Builder tcpKeepAliveConfiguration(Consumer<TcpKeepAliveConfigur
          * @param postQuantumTlsEnabled whether to prefer Post Quantum TLS
          * @return The builder of the method chaining.
          */
-        AwsCrtHttpClient.Builder postQuantumTlsEnabled(Boolean postQuantumTlsEnabled);
+        Builder postQuantumTlsEnabled(Boolean postQuantumTlsEnabled);
     }
 
     /**
      * Factory that allows more advanced configuration of the AWS CRT HTTP implementation.
      * Use {@link #builder()} to configure and construct an immutable instance of the factory.
      */
     private static final class DefaultBuilder
-        extends AwsCrtClientBuilderBase<AwsCrtHttpClient.Builder> implements AwsCrtHttpClient.Builder {
+        extends AwsCrtClientBuilderBase<Builder> implements Builder {
 
 
         @Override

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/CrtAsyncRequestExecutor.java

@@ -89,13 +89,22 @@ private void doExecute(CrtAsyncRequestContext executionContext,
                             }
                             try {
                                 stream.activate();
-                                streamFuture.complete(stream);
-                                asyncRequest.requestContentPublisher().subscribe(bodySubscriber);
                             } catch (Throwable t) {
+                                // Stream is acquired but not activated and not yet published via
+                                // streamFuture. No other path can reach it, so clean it up directly.
+                                stream.cancel();
+                                stream.close();
                                 handleAcquireFailure(t, streamFuture, requestFuture, errorNotifier);
+                                return null;
                             }
+                            streamFuture.complete(stream);
+                            asyncRequest.requestContentPublisher().subscribe(bodySubscriber);
                             return null;
                         }).exceptionally(t -> {
+                            // Reached when the handle lambda throws (e.g., publisher.subscribe).
+                            // closeConnection is a no-op if the stream isn't in streamFuture yet;
+                            // otherwise it tears down the published stream.
+                            streamHandler.closeConnection();
                             handleAcquireFailure(t, streamFuture, requestFuture, errorNotifier);
                             return null;
                         });

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/CrtRequestExecutor.java

@@ -81,12 +81,18 @@ private void doExecute(CrtRequestContext executionContext,
                             }
                             try {
                                 streamBase.activate();
-                                streamFuture.complete(streamBase);
                             } catch (Throwable t) {
+                                // Stream is acquired but not activated and not yet published via
+                                // streamFuture. No other path can reach it, so clean it up directly.
+                                streamBase.cancel();
+                                streamBase.close();
                                 handleAcquireFailure(t, streamFuture, responseFuture);
+                                return null;
                             }
+                            streamFuture.complete(streamBase);
                             return null;
                         }).exceptionally(t -> {
+                            // Defensive: only reached if the handle lambda itself throws.
                             handleAcquireFailure(t, streamFuture, responseFuture);
                             return null;
                         });

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/CrtStreamHandler.java

@@ -106,7 +106,7 @@ private static Throwable unwrap(Throwable t) {
 
     public void incrementWindow(int windowSize) {
         synchronized (streamLock) {
-            HttpStreamBase s = streamFuture.getNow(null);
+            HttpStreamBase s = streamIfAvailable();
             if (!streamClosed && s != null) {
                 s.incrementWindow(windowSize);
             }
@@ -119,7 +119,7 @@ public void incrementWindow(int windowSize) {
      */
     public void releaseConnection() {
         synchronized (streamLock) {
-            HttpStreamBase s = streamFuture.getNow(null);
+            HttpStreamBase s = streamIfAvailable();
             if (!streamClosed && s != null) {
                 streamClosed = true;
                 s.close();
@@ -134,12 +134,24 @@ public void releaseConnection() {
      */
     public void closeConnection() {
         synchronized (streamLock) {
-            HttpStreamBase s = streamFuture.getNow(null);
+            HttpStreamBase s = streamIfAvailable();
             if (!streamClosed && s != null) {
                 streamClosed = true;
                 s.cancel();
                 s.close();
             }
         }
     }
+
+    /**
+     * Returns the acquired stream if {@link #streamFuture} completed normally, otherwise {@code null}.
+     * Tolerates exceptional or pending completion (in contrast to {@link CompletableFuture#getNow}, which
+     * throws {@link CompletionException} when the future is exceptional).
+     */
+    private HttpStreamBase streamIfAvailable() {
+        if (!streamFuture.isDone() || streamFuture.isCompletedExceptionally()) {
+            return null;
+        }
+        return streamFuture.getNow(null);
+    }
 }

http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientWireMockTest.java

@@ -113,6 +113,36 @@ public void sharedEventLoopGroup_closeOneClient_shouldNotAffectOtherClients() th
         }
     }
 
+    @Test
+    public void contentStreamReadThrows_propagatesIoExceptionAndDoesNotLeakConnection() throws Exception {
+        try (SdkHttpClient client = AwsCrtHttpClient.builder().maxConcurrency(1).build()) {
+            URI uri = URI.create("http://localhost:" + mockServer.port());
+            stubFor(any(urlPathEqualTo("/")).willReturn(aResponse().withBody(randomAlphabetic(10))));
+            SdkHttpRequest request = createRequest(uri);
+
+            IOException readError = new IOException("simulated read failure");
+            ExecutableHttpRequest failing = client.prepareRequest(
+                HttpExecuteRequest.builder()
+                                  .request(request)
+                                  .contentStreamProvider(() -> new java.io.InputStream() {
+                                      @Override
+                                      public int read() throws IOException {
+                                          throw readError;
+                                      }
+                                  })
+                                  .build());
+
+            assertThatThrownBy(failing::call)
+                .isInstanceOf(IOException.class)
+                .isSameAs(readError);
+
+            // If the connection leaked, this second request would hang since maxConcurrency=1.
+            // Use a short overall test timeout to fail fast if the leak regresses.
+            HttpExecuteResponse second = makeSimpleRequest(client, null);
+            assertThat(second.httpResponse().statusCode()).isEqualTo(200);
+        }
+    }
+
     @Test
     public void abortRequest_shouldFailTheExceptionWithIOException() throws Exception {
         try (SdkHttpClient client = AwsCrtHttpClient.create()) {