Enable default connection health monitoring for CRT HTTP clients

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

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS CRT HTTP Client",
+    "contributor": "",
+    "description": "Enabled default connection health monitoring for the AWS CRT HTTP client. Connections that remain stalled below 1 byte per second for the duration the read/write timeout (default 30 seconds) are now automatically terminated. This behavior can be overridden via ConnectionHealthConfiguration."
+}

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

@@ -154,8 +154,10 @@ public interface Builder extends SdkAsyncHttpClient.Builder<AwsCrtAsyncHttpClien
          * then the connection is considered unhealthy and will be shut down.
          *
          * <p>
-         * By default, monitoring options are disabled. You can enable {@code healthChecks} by providing this configuration
-         * and specifying the options for monitoring for the connection manager.
+         * If not explicitly configured, a default health configuration is applied with a minimum throughput of 1 byte per
+         * second and a throughput failure interval of 30 seconds. The failure interval is derived from the read/write timeout
+         * settings and will change if those are overridden by service specific defaults.
+         *
          * @param healthChecksConfiguration The health checks config to use
          * @return The builder of the method chaining.
          */

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

@@ -193,8 +193,10 @@ public interface Builder extends SdkHttpClient.Builder<AwsCrtHttpClient.Builder>
          * then the connection is considered unhealthy and will be shut down.
          *
          * <p>
-         * By default, monitoring options are disabled. You can enable {@code healthChecks} by providing this configuration
-         * and specifying the options for monitoring for the connection manager.
+         * If not explicitly configured, a default health configuration is applied with a minimum throughput of 1 byte per
+         * second and a throughput failure interval of 30 seconds. The failure interval is derived from the read/write timeout
+         * settings and will change if those are overridden by service specific defaults.
+         *
          * @param healthChecksConfiguration The health checks config to use
          * @return The builder of the method chaining.
          */

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

@@ -19,6 +19,7 @@
 import static software.amazon.awssdk.crtcore.CrtConfigurationUtils.resolveProxy;
 import static software.amazon.awssdk.http.SdkHttpConfigurationOption.PROTOCOL;
 import static software.amazon.awssdk.http.crt.internal.AwsCrtConfigurationUtils.buildSocketOptions;
+import static software.amazon.awssdk.http.crt.internal.AwsCrtConfigurationUtils.defaultConnectionHealthConfiguration;
 import static software.amazon.awssdk.http.crt.internal.AwsCrtConfigurationUtils.resolveCipherPreference;
 import static software.amazon.awssdk.utils.FunctionalUtils.invokeSafely;
 
@@ -90,7 +91,9 @@ abstract class AwsCrtHttpClientBase implements SdkAutoCloseable {
             this.readBufferSize = builder.getReadBufferSizeInBytes() == null ?
                                   DEFAULT_STREAM_WINDOW_SIZE : builder.getReadBufferSizeInBytes();
             this.maxConnectionsPerEndpoint = config.get(SdkHttpConfigurationOption.MAX_CONNECTIONS);
-            this.monitoringOptions = resolveHttpMonitoringOptions(builder.getConnectionHealthConfiguration()).orElse(null);
+            this.monitoringOptions =
+                resolveHttpMonitoringOptions(builder.getConnectionHealthConfiguration())
+                    .orElseGet(() -> defaultConnectionHealthConfiguration(config));
             this.maxConnectionIdleInMilliseconds = config.get(SdkHttpConfigurationOption.CONNECTION_MAX_IDLE_TIMEOUT).toMillis();
             this.connectionAcquisitionTimeout = config.get(SdkHttpConfigurationOption.CONNECTION_ACQUIRE_TIMEOUT).toMillis();
             this.proxyOptions = resolveProxy(builder.getProxyConfiguration(), tlsContext).orElse(null);

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

@@ -18,10 +18,13 @@
 
 import java.time.Duration;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.crt.http.HttpMonitoringOptions;
 import software.amazon.awssdk.crt.io.SocketOptions;
 import software.amazon.awssdk.crt.io.TlsCipherPreference;
+import software.amazon.awssdk.http.SdkHttpConfigurationOption;
 import software.amazon.awssdk.http.crt.AwsCrtAsyncHttpClient;
 import software.amazon.awssdk.http.crt.TcpKeepAliveConfiguration;
+import software.amazon.awssdk.utils.AttributeMap;
 import software.amazon.awssdk.utils.Logger;
 import software.amazon.awssdk.utils.NumericUtils;
 
@@ -70,4 +73,14 @@ public static TlsCipherPreference resolveCipherPreference(Boolean postQuantumTls
         return pqTls;
     }
 
+    public static HttpMonitoringOptions defaultConnectionHealthConfiguration(AttributeMap config) {
+        HttpMonitoringOptions httpMonitoringOptions = new HttpMonitoringOptions();
+        httpMonitoringOptions.setMinThroughputBytesPerSecond(1);
+        long readTimeout = config.get(SdkHttpConfigurationOption.READ_TIMEOUT).getSeconds();
+        long writeTimeout = config.get(SdkHttpConfigurationOption.WRITE_TIMEOUT).getSeconds();
+        int maxTimeout = NumericUtils.saturatedCast(Math.max(readTimeout, writeTimeout));
+        httpMonitoringOptions.setAllowableThroughputFailureIntervalSeconds(maxTimeout);
+        return httpMonitoringOptions;
+    }
+
 }

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

@@ -0,0 +1,120 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.http.crt;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static software.amazon.awssdk.http.crt.CrtHttpClientTestUtils.createRequest;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.URI;
+import java.time.Duration;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.crt.Log;
+import software.amazon.awssdk.http.ExecutableHttpRequest;
+import software.amazon.awssdk.http.HttpExecuteRequest;
+import software.amazon.awssdk.http.RecordingResponseHandler;
+import software.amazon.awssdk.http.SdkHttpConfigurationOption;
+import software.amazon.awssdk.http.SdkHttpRequest;
+import software.amazon.awssdk.http.async.AsyncExecuteRequest;
+import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
+import software.amazon.awssdk.http.SdkHttpClient;
+import software.amazon.awssdk.utils.AttributeMap;
+
+/**
+ * Functional tests verifying that the default connection health configuration
+ * (applied when no explicit {@link ConnectionHealthConfiguration} is set)
+ * correctly terminates connections to non-responding servers.
+ */
+class NonResponsiveServerTest {
+
+    private static final Duration SHORT_TIMEOUT = Duration.ofSeconds(2);
+    private static final AttributeMap SHORT_TIMEOUTS = AttributeMap.builder()
+                                                                   .put(SdkHttpConfigurationOption.READ_TIMEOUT, SHORT_TIMEOUT)
+                                                                   .put(SdkHttpConfigurationOption.WRITE_TIMEOUT, SHORT_TIMEOUT)
+                                                                   .build();
+
+    private ServerSocket serverSocket;
+
+    @BeforeEach
+    void setUp() throws IOException {
+        Log.initLoggingToStdout(Log.LogLevel.Warn);
+        serverSocket = new ServerSocket(0);
+        // Accept connections in a daemon thread but never respond
+        Thread acceptThread = new Thread(() -> {
+            while (!serverSocket.isClosed()) {
+                try {
+                    Socket socket = serverSocket.accept();
+                    // Hold the connection open, never send a response
+                    Thread.sleep(Long.MAX_VALUE);
+                } catch (Exception e) {
+                    // Server shutting down
+                }
+            }
+        });
+        acceptThread.setDaemon(true);
+        acceptThread.start();
+    }
+
+    @AfterEach
+    void tearDown() throws IOException {
+        if (serverSocket != null && !serverSocket.isClosed()) {
+            serverSocket.close();
+        }
+    }
+
+    @Test
+    void syncClient_noExplicitHealthConfig_serverNeverResponds_shouldThrow() {
+        try (SdkHttpClient client = AwsCrtHttpClient.builder().buildWithDefaults(SHORT_TIMEOUTS)) {
+            URI uri = URI.create("http://localhost:" + serverSocket.getLocalPort());
+            SdkHttpRequest request = createRequest(uri);
+            ExecutableHttpRequest executableRequest = client.prepareRequest(
+                HttpExecuteRequest.builder().request(request)
+                                  .contentStreamProvider(() -> new ByteArrayInputStream(new byte[0]))
+                                  .build());
+            assertThatThrownBy(executableRequest::call).isInstanceOf(IOException.class)
+                                                       .hasMessageContaining("failure to meet throughput minimum");
+        }
+    }
+
+    @Test
+    void asyncClient_noExplicitHealthConfig_serverNeverResponds_shouldCompleteExceptionally()
+        throws InterruptedException, TimeoutException {
+        try (SdkAsyncHttpClient client = AwsCrtAsyncHttpClient.builder().buildWithDefaults(SHORT_TIMEOUTS)) {
+            URI uri = URI.create("http://localhost:" + serverSocket.getLocalPort());
+            SdkHttpRequest request = createRequest(uri);
+            RecordingResponseHandler recorder = new RecordingResponseHandler();
+
+            client.execute(AsyncExecuteRequest.builder()
+                                              .request(request)
+                                              .requestContentPublisher(new EmptyPublisher())
+                                              .responseHandler(recorder)
+                                              .build());
+
+            assertThatThrownBy(() -> recorder.completeFuture().get(10, TimeUnit.SECONDS))
+                .hasCauseInstanceOf(IOException.class)
+                .hasMessageContaining("failure to meet throughput minimum");
+        }
+    }
+}

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

@@ -21,16 +21,17 @@
 
 import java.time.Duration;
 import java.util.stream.Stream;
-import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Assumptions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
-import software.amazon.awssdk.crt.CrtResource;
+import software.amazon.awssdk.crt.http.HttpMonitoringOptions;
 import software.amazon.awssdk.crt.io.SocketOptions;
 import software.amazon.awssdk.crt.io.TlsCipherPreference;
+import software.amazon.awssdk.http.SdkHttpConfigurationOption;
 import software.amazon.awssdk.http.crt.TcpKeepAliveConfiguration;
+import software.amazon.awssdk.utils.AttributeMap;
 
 class AwsCrtConfigurationUtilsTest {
     @ParameterizedTest
@@ -103,4 +104,31 @@ private static Stream<Arguments> tcpKeepAliveConfiguration() {
         );
     }
 
+    @ParameterizedTest
+    @MethodSource("defaultConnectionHealthConfigurationCases")
+    void defaultConnectionHealthConfiguration_shouldUseMaxOfReadWriteTimeout(Duration readTimeout,
+                                                                             Duration writeTimeout,
+                                                                             int expectedInterval) {
+        AttributeMap config = AttributeMap.builder()
+                                          .put(SdkHttpConfigurationOption.READ_TIMEOUT, readTimeout)
+                                          .put(SdkHttpConfigurationOption.WRITE_TIMEOUT, writeTimeout)
+                                          .build();
+
+        HttpMonitoringOptions result = AwsCrtConfigurationUtils.defaultConnectionHealthConfiguration(config);
+
+        assertThat(result.getMinThroughputBytesPerSecond()).isEqualTo(1);
+        assertThat(result.getAllowableThroughputFailureIntervalSeconds()).isEqualTo(expectedInterval);
+    }
+
+    private static Stream<Arguments> defaultConnectionHealthConfigurationCases() {
+        return Stream.of(
+            Arguments.of(Duration.ofSeconds(30), Duration.ofSeconds(30), 30),
+            Arguments.of(Duration.ofSeconds(60), Duration.ofSeconds(10), 60),
+            Arguments.of(Duration.ofSeconds(10), Duration.ofSeconds(45), 45),
+            // overflow: value exceeding Integer.MAX_VALUE should saturate
+            Arguments.of(Duration.ofSeconds((long) Integer.MAX_VALUE + 1), Duration.ofSeconds(1), Integer.MAX_VALUE),
+            Arguments.of(Duration.ofSeconds(1), Duration.ofSeconds((long) Integer.MAX_VALUE + 1), Integer.MAX_VALUE)
+        );
+    }
+
 }