Switch CRT HTTP sync client from pull-based body to push-based writeData

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

@@ -0,0 +1,6 @@
+{
+    "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."
+}

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

@@ -18,13 +18,16 @@
 import static software.amazon.awssdk.http.HttpMetric.HTTP_CLIENT_NAME;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.function.Consumer;
 import software.amazon.awssdk.annotations.SdkPublicApi;
 import software.amazon.awssdk.crt.http.HttpException;
 import software.amazon.awssdk.crt.http.HttpStreamManager;
+import software.amazon.awssdk.http.ContentStreamProvider;
 import software.amazon.awssdk.http.ExecutableHttpRequest;
 import software.amazon.awssdk.http.HttpExecuteRequest;
 import software.amazon.awssdk.http.HttpExecuteResponse;
@@ -35,6 +38,8 @@
 import software.amazon.awssdk.http.crt.internal.AwsCrtClientBuilderBase;
 import software.amazon.awssdk.http.crt.internal.CrtRequestContext;
 import software.amazon.awssdk.http.crt.internal.CrtRequestExecutor;
+import software.amazon.awssdk.http.crt.internal.CrtStreamHandler;
+import software.amazon.awssdk.http.crt.internal.CrtUtils;
 import software.amazon.awssdk.utils.AttributeMap;
 import software.amazon.awssdk.utils.CompletableFutureUtils;
 
@@ -64,7 +69,7 @@ private AwsCrtHttpClient(DefaultBuilder builder, AttributeMap config) {
         }
     }
 
-    public static AwsCrtHttpClient.Builder builder() {
+    public static Builder builder() {
         return new DefaultBuilder();
     }
 
@@ -107,6 +112,8 @@ public ExecutableHttpRequest prepareRequest(HttpExecuteRequest request) {
     }
 
     private static final class CrtHttpRequest implements ExecutableHttpRequest {
+        private static final int WRITE_BUFFER_SIZE = 16 * 1024;
+
         private final CrtRequestContext context;
         private volatile CompletableFuture<SdkHttpFullResponse> responseFuture;
 
@@ -117,37 +124,70 @@ 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 {
-                responseFuture = new CrtRequestExecutor().execute(context);
+                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);
-                }
+            } 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 IOException) {
-                    throw (IOException) 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 (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);
+        }
 
-                if (cause instanceof HttpException) {
-                    throw (HttpException) cause;
-                }
+        private void writeRequestBody(CrtStreamHandler streamHandler) throws IOException {
+            ContentStreamProvider provider = context.sdkRequest().contentStreamProvider().orElse(null);
+            if (provider == null) {
+                return;
+            }
 
-                if (cause instanceof InterruptedException) {
-                    Thread.currentThread().interrupt();
-                    throw new IOException("Request was cancelled", cause);
+            streamHandler.waitForStream();
+            try (InputStream inputStream = provider.newStream()) {
+                byte[] buf = new byte[WRITE_BUFFER_SIZE];
+                int read;
+                while ((read = inputStream.read(buf, 0, buf.length)) >= 0) {
+                    byte[] chunk = read == buf.length ? buf : Arrays.copyOf(buf, read);
+                    CompletableFutureUtils.joinInterruptibly(streamHandler.writeData(chunk, false));
                 }
-                throw new RuntimeException(e.getCause());
+                CompletableFutureUtils.joinInterruptibly(streamHandler.writeData(null, true));
             }
         }
 
@@ -162,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
@@ -179,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.
@@ -214,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
@@ -224,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
@@ -260,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
@@ -274,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);
 
         /**
@@ -293,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

@@ -67,24 +67,28 @@ private void doExecute(CrtAsyncRequestContext executionContext,
 
         HttpRequestBase crtRequest = toAsyncCrtRequest(executionContext);
 
-        HttpStreamBaseResponseHandler crtResponseHandler =
-            CrtResponseAdapter.toCrtResponseHandler(requestFuture, asyncRequest.responseHandler());
+        CompletableFuture<HttpStreamBase> streamFuture = new CompletableFuture<>();
+        CrtStreamHandler streamHandler = new CrtStreamHandler(streamFuture);
 
-        CompletableFuture<HttpStreamBase> streamFuture =
-            executionContext.streamManager().acquireStream(crtRequest, crtResponseHandler);
+        HttpStreamBaseResponseHandler crtResponseHandler =
+            CrtResponseAdapter.toCrtResponseHandler(requestFuture, asyncRequest.responseHandler(), streamHandler);
 
         long finalAcquireStartTime = acquireStartTime;
 
-        streamFuture.whenComplete((stream, throwable) -> {
-            if (shouldPublishMetrics) {
-                reportMetrics(executionContext.streamManager(), metricCollector, finalAcquireStartTime);
-            }
-
-            if (throwable != null) {
-                Throwable toThrow = wrapCrtException(throwable);
-                reportAsyncFailure(toThrow, requestFuture, asyncRequest.responseHandler());
-            }
-        });
+        executionContext.streamManager().acquireStream(crtRequest, crtResponseHandler)
+                        .whenComplete((stream, throwable) -> {
+                            if (shouldPublishMetrics) {
+                                reportMetrics(executionContext.streamManager(), metricCollector, finalAcquireStartTime);
+                            }
+
+                            if (throwable != null) {
+                                Throwable toThrow = wrapCrtException(throwable);
+                                streamFuture.completeExceptionally(toThrow);
+                                reportAsyncFailure(toThrow, requestFuture, asyncRequest.responseHandler());
+                            } else {
+                                streamFuture.complete(stream);
+                            }
+                        });
     }
 
     /**