By default (when this is not set), the client shares a single, process-wide event-loop group sized to
+ * {@code Runtime.getRuntime().availableProcessors()}, shared with every other CRT client in the JVM. When this value is
+ * set, the client instead creates and owns a private event-loop group of the given size; that group is shut down when
+ * this client is closed and is not shared with any other client.
+ *
+ * This is an advanced tuning and isolation control, and each client configured with an explicit size consumes that
+ * many additional IO threads. Oversizing wastes threads and adds context-switching and memory overhead without improving
+ * throughput; undersizing can leave the client's IO as a bottleneck and underutilize available cores. The best value
+ * depends on your workload and hardware, so benchmark your own application before changing it from the default. An
+ * excessively high value relative to the number of available processors is logged as a warning.
+ *
+ * @param numEventLoopThreads the number of event-loop threads; must be greater than 1, or {@code null} to use the shared
+ * default.
+ * @return The builder for method chaining.
+ */
+ AwsCrtAsyncHttpClient.Builder numEventLoopThreads(Integer numEventLoopThreads);
+
/**
* Sets the http proxy configuration to use for this client.
* @param proxyConfiguration The http proxy configuration to use
diff --git a/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtHttpClient.java b/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtHttpClient.java
index 999875889764..6eeab8a20242 100644
--- a/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtHttpClient.java
+++ b/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtHttpClient.java
@@ -56,12 +56,16 @@
public final class AwsCrtHttpClient extends AwsCrtHttpClientBase implements SdkHttpClient {
private AwsCrtHttpClient(DefaultBuilder builder, AttributeMap config) {
- super(builder, config);
- if (this.protocol == Protocol.HTTP2) {
+ super(builder, validateProtocol(config));
+ }
+
+ private static AttributeMap validateProtocol(AttributeMap config) {
+ if (config.get(SdkHttpConfigurationOption.PROTOCOL) == Protocol.HTTP2) {
throw new UnsupportedOperationException(
"HTTP/2 is not supported for sync HTTP clients. Either use HTTP/1.1 (the default) or use an async "
+ "HTTP client (e.g., AwsCrtAsyncHttpClient).");
}
+ return config;
}
public static AwsCrtHttpClient.Builder builder() {
@@ -181,6 +185,26 @@ public interface Builder extends SdkHttpClient.Builder
*/
AwsCrtHttpClient.Builder readBufferSizeInBytes(Long readBufferSize);
+ /**
+ * Configure the number of event-loop (IO) threads in this client's event-loop group.
+ *
+ * By default (when this is not set), the client shares a single, process-wide event-loop group sized to
+ * {@code Runtime.getRuntime().availableProcessors()}, shared with every other CRT client in the JVM. When this value is
+ * set, the client instead creates and owns a private event-loop group of the given size; that group is shut down when
+ * this client is closed and is not shared with any other client.
+ *
+ *
This is an advanced tuning and isolation control, and each client configured with an explicit size consumes that
+ * many additional IO threads. Oversizing wastes threads and adds context-switching and memory overhead without improving
+ * throughput; undersizing can leave the client's IO as a bottleneck and underutilize available cores. The best value
+ * depends on your workload and hardware, so benchmark your own application before changing it from the default. An
+ * excessively high value relative to the number of available processors is logged as a warning.
+ *
+ * @param numEventLoopThreads the number of event-loop threads; must be greater than 1, or {@code null} to use the shared
+ * default.
+ * @return The builder for method chaining.
+ */
+ AwsCrtHttpClient.Builder numEventLoopThreads(Integer numEventLoopThreads);
+
/**
* Sets the http proxy configuration to use for this client.
* @param proxyConfiguration The http proxy configuration to use
diff --git a/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientBase.java b/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientBase.java
index 3c5ec7ced8f8..4d8dc88ba6f6 100644
--- a/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientBase.java
+++ b/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientBase.java
@@ -40,6 +40,7 @@
import software.amazon.awssdk.crt.http.HttpStreamManagerOptions;
import software.amazon.awssdk.crt.http.HttpVersion;
import software.amazon.awssdk.crt.io.ClientBootstrap;
+import software.amazon.awssdk.crt.io.EventLoopGroup;
import software.amazon.awssdk.crt.io.SocketOptions;
import software.amazon.awssdk.crt.io.TlsConnectionOptions;
import software.amazon.awssdk.crt.io.TlsContext;
@@ -72,6 +73,10 @@ abstract class AwsCrtHttpClientBase implements SdkAutoCloseable {
private static final String AWS_COMMON_RUNTIME = "AwsCommonRuntime";
private static final long DEFAULT_STREAM_WINDOW_SIZE = 16L * 1024L * 1024L; // 16 MB
+ // Heuristic threshold (not an API contract) for warning about likely-accidental oversizing of the per-client
+ // event-loop group. A value at or above this multiple of the available processors is almost certainly unintended.
+ private static final int NUM_EVENT_LOOP_THREADS_WARN_MULTIPLIER = 4;
+
protected final long readBufferSize;
protected final Protocol protocol;
private final Map connectionPools = new ConcurrentHashMap<>();
@@ -89,35 +94,77 @@ abstract class AwsCrtHttpClientBase implements SdkAutoCloseable {
private boolean isClosed = false;
AwsCrtHttpClientBase(AwsCrtClientBuilderBase builder, AttributeMap config) {
- ClientBootstrap clientBootstrap = new ClientBootstrap(null, null);
- SocketOptions clientSocketOptions = buildSocketOptions(builder.getTcpKeepAliveConfiguration(),
- config.get(SdkHttpConfigurationOption.CONNECTION_TIMEOUT));
- TlsContextOptions clientTlsContextOptions =
- TlsContextOptions.createDefaultClient()
- .withCipherPreference(resolveCipherPreference(builder.getPostQuantumTlsEnabled()))
- .withMinimumTlsVersion(resolveMinTlsVersion(builder.getMinTlsVersion()))
- .withVerifyPeer(!config.get(SdkHttpConfigurationOption.TRUST_ALL_CERTIFICATES));
- this.protocol = config.get(PROTOCOL);
- if (protocol == Protocol.HTTP2) {
- clientTlsContextOptions = clientTlsContextOptions.withAlpnList("h2");
+ // These native resources are created before being registered as owned, and each creation can throw
+ // (e.g. CrtRuntimeException). Because close() is never invoked on a client that failed to construct, any
+ // already-created resource must be released here if a later step throws - otherwise the private EventLoopGroup
+ // (and its OS threads) and the other native handles would leak.
+ EventLoopGroup eventLoopGroup = null;
+ ClientBootstrap clientBootstrap = null;
+ SocketOptions clientSocketOptions = null;
+ TlsContextOptions clientTlsContextOptions = null;
+ TlsContext clientTlsContext = null;
+ try {
+ Integer numEventLoopThreads = builder.getNumEventLoopThreads();
+ if (numEventLoopThreads != null) {
+ warnIfNumEventLoopThreadsIsExcessive(numEventLoopThreads);
+ eventLoopGroup = new EventLoopGroup(numEventLoopThreads);
+ }
+ clientBootstrap = new ClientBootstrap(eventLoopGroup, null);
+ clientSocketOptions = buildSocketOptions(builder.getTcpKeepAliveConfiguration(),
+ config.get(SdkHttpConfigurationOption.CONNECTION_TIMEOUT));
+ clientTlsContextOptions =
+ TlsContextOptions.createDefaultClient()
+ .withCipherPreference(resolveCipherPreference(builder.getPostQuantumTlsEnabled()))
+ .withMinimumTlsVersion(resolveMinTlsVersion(builder.getMinTlsVersion()))
+ .withVerifyPeer(!config.get(SdkHttpConfigurationOption.TRUST_ALL_CERTIFICATES));
+ this.protocol = config.get(PROTOCOL);
+ if (protocol == Protocol.HTTP2) {
+ clientTlsContextOptions = clientTlsContextOptions.withAlpnList("h2");
+ }
+
+ this.tlsContextOptions = registerOwnedResource(clientTlsContextOptions);
+ clientTlsContext = new TlsContext(clientTlsContextOptions);
+
+ // The bootstrap holds a native reference to its event-loop group, so the group is registered before (and thus
+ // closed after) the bootstrap to keep CRT teardown ordering correct. A null group leaves the shared static
+ // default group untouched.
+ registerOwnedResource(eventLoopGroup);
+ this.bootstrap = registerOwnedResource(clientBootstrap);
+ this.socketOptions = registerOwnedResource(clientSocketOptions);
+ this.tlsContext = registerOwnedResource(clientTlsContext);
+ this.tlsNegotiationTimeout = config.get(SdkHttpConfigurationOption.TLS_NEGOTIATION_TIMEOUT);
+ this.readBufferSize = builder.getReadBufferSizeInBytes() == null ?
+ DEFAULT_STREAM_WINDOW_SIZE : builder.getReadBufferSizeInBytes();
+ this.maxStreamsPerEndpoint = config.get(SdkHttpConfigurationOption.MAX_CONNECTIONS);
+ this.monitoringOptions =
+ resolveHttpMonitoringOptions(builder.getConnectionHealthConfiguration())
+ .orElse(null);
+ 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);
+ } catch (RuntimeException e) {
+ // Release in reverse dependency order: the TlsContext wraps its options, and the bootstrap holds a
+ // reference to the event-loop group, so those wrappers are closed before what they depend on.
+ IoUtils.closeQuietly(clientTlsContext, log.logger());
+ IoUtils.closeQuietly(clientTlsContextOptions, log.logger());
+ IoUtils.closeQuietly(clientSocketOptions, log.logger());
+ IoUtils.closeQuietly(clientBootstrap, log.logger());
+ IoUtils.closeQuietly(eventLoopGroup, log.logger());
+ throw e;
}
+ }
- this.tlsContextOptions = registerOwnedResource(clientTlsContextOptions);
- TlsContext clientTlsContext = new TlsContext(clientTlsContextOptions);
-
- this.bootstrap = registerOwnedResource(clientBootstrap);
- this.socketOptions = registerOwnedResource(clientSocketOptions);
- this.tlsContext = registerOwnedResource(clientTlsContext);
- this.tlsNegotiationTimeout = config.get(SdkHttpConfigurationOption.TLS_NEGOTIATION_TIMEOUT);
- this.readBufferSize = builder.getReadBufferSizeInBytes() == null ?
- DEFAULT_STREAM_WINDOW_SIZE : builder.getReadBufferSizeInBytes();
- this.maxStreamsPerEndpoint = config.get(SdkHttpConfigurationOption.MAX_CONNECTIONS);
- this.monitoringOptions =
- resolveHttpMonitoringOptions(builder.getConnectionHealthConfiguration())
- .orElse(null);
- 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);
+ private static void warnIfNumEventLoopThreadsIsExcessive(int numEventLoopThreads) {
+ int availableProcessors = Math.max(1, Runtime.getRuntime().availableProcessors());
+ if (numEventLoopThreads >= NUM_EVENT_LOOP_THREADS_WARN_MULTIPLIER * availableProcessors) {
+ log.warn(() -> String.format(
+ "numEventLoopThreads is set to %d, which is unusually high relative to the %d available processor(s). "
+ + "Each client configured with numEventLoopThreads gets its own private event-loop group, so a high count "
+ + "multiplied across multiple clients can lead to thread explosion, excessive context-switching, and increased "
+ + "memory use without improving throughput. Consider benchmarking your workload to confirm this value is "
+ + "necessary.",
+ numEventLoopThreads, availableProcessors));
+ }
}
/**
diff --git a/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/AwsCrtClientBuilderBase.java b/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/AwsCrtClientBuilderBase.java
index 8114bc53c208..eff0e8a6d874 100644
--- a/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/AwsCrtClientBuilderBase.java
+++ b/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/AwsCrtClientBuilderBase.java
@@ -35,6 +35,7 @@ public class AwsCrtClientBuilderBase {
private TcpKeepAliveConfiguration tcpKeepAliveConfiguration;
private Boolean postQuantumTlsEnabled;
private TlsVersion minTlsVersion;
+ private Integer numEventLoopThreads;
protected AwsCrtClientBuilderBase() {
}
@@ -63,6 +64,17 @@ public Long getReadBufferSizeInBytes() {
return this.readBufferSize;
}
+ public BuilderT numEventLoopThreads(Integer numEventLoopThreads) {
+ Validate.isTrue(numEventLoopThreads == null || numEventLoopThreads > 1,
+ "numEventLoopThreads must be greater than 1");
+ this.numEventLoopThreads = numEventLoopThreads;
+ return thisBuilder();
+ }
+
+ public Integer getNumEventLoopThreads() {
+ return this.numEventLoopThreads;
+ }
+
public BuilderT proxyConfiguration(ProxyConfiguration proxyConfiguration) {
this.proxyConfiguration = proxyConfiguration;
diff --git a/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtAsyncHttpClientWireMockTest.java b/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtAsyncHttpClientWireMockTest.java
index 4990a9fab879..fae454dc05a8 100644
--- a/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtAsyncHttpClientWireMockTest.java
+++ b/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtAsyncHttpClientWireMockTest.java
@@ -17,61 +17,68 @@
import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
import static com.github.tomakehurst.wiremock.client.WireMock.any;
-import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo;
import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static software.amazon.awssdk.http.HttpTestUtils.createProvider;
-import static software.amazon.awssdk.http.SdkHttpConfigurationOption.PROTOCOL;
import static software.amazon.awssdk.http.crt.CrtHttpClientTestUtils.createRequest;
+import static software.amazon.awssdk.http.crt.CrtHttpClientTestUtils.liveEventLoopGroups;
+import static software.amazon.awssdk.http.crt.CrtHttpClientTestUtils.newEventLoopGroups;
+import static software.amazon.awssdk.http.crt.CrtHttpClientTestUtils.waitForEventLoopGroupsReleased;
-import com.github.tomakehurst.wiremock.junit.WireMockRule;
+import com.github.tomakehurst.wiremock.junit5.WireMockExtension;
+import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo;
import java.net.URI;
import java.time.Duration;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Rule;
-import org.junit.Test;
+import org.apache.logging.log4j.Level;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
import software.amazon.awssdk.crt.CrtResource;
import software.amazon.awssdk.crt.Log;
import software.amazon.awssdk.http.HttpMetric;
-import software.amazon.awssdk.http.Protocol;
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.metrics.MetricCollection;
+import software.amazon.awssdk.testutils.LogCaptor;
import software.amazon.awssdk.utils.AttributeMap;
public class AwsCrtAsyncHttpClientWireMockTest {
- @Rule
- public WireMockRule mockServer = new WireMockRule(wireMockConfig()
- .dynamicPort()
- .dynamicHttpsPort());
+ @RegisterExtension
+ static WireMockExtension mockServer = WireMockExtension.newInstance()
+ .options(wireMockConfig().dynamicPort().dynamicHttpsPort())
+ .build();
- @BeforeClass
+ @BeforeAll
public static void setup() {
System.setProperty("aws.crt.debugnative", "true");
Log.initLoggingToStdout(Log.LogLevel.Warn);
}
@Test
- public void closeClient_reuse_throwException() {
+ public void closeClient_reuse_throwException(WireMockRuntimeInfo wm) {
SdkAsyncHttpClient client = AwsCrtAsyncHttpClient.create();
client.close();
- assertThatThrownBy(() -> makeSimpleRequest(client)).hasMessageContaining("is closed");
+ assertThatThrownBy(() -> makeSimpleRequest(client, wm)).hasMessageContaining("is closed");
}
@Test
- public void sendRequest_withCollector_shouldCollectMetrics() throws Exception {
+ public void sendRequest_withCollector_shouldCollectMetrics(WireMockRuntimeInfo wm) throws Exception {
try (SdkAsyncHttpClient client = AwsCrtAsyncHttpClient.builder().maxConcurrency(10).build()) {
- RecordingResponseHandler recorder = makeSimpleRequest(client);
+ RecordingResponseHandler recorder = makeSimpleRequest(client, wm);
MetricCollection metrics = recorder.collector().collect();
assertThat(metrics.metricValues(HttpMetric.HTTP_CLIENT_NAME)).containsExactly("AwsCommonRuntime");
@@ -83,30 +90,116 @@ public void sendRequest_withCollector_shouldCollectMetrics() throws Exception {
}
@Test
- public void sharedEventLoopGroup_closeOneClient_shouldNotAffectOtherClients() throws Exception {
+ public void sharedEventLoopGroup_closeOneClient_shouldNotAffectOtherClients(WireMockRuntimeInfo wm) throws Exception {
try (SdkAsyncHttpClient client = AwsCrtAsyncHttpClient.create()) {
- makeSimpleRequest(client);
+ makeSimpleRequest(client, wm);
}
try (SdkAsyncHttpClient anotherClient = AwsCrtAsyncHttpClient.create()) {
- makeSimpleRequest(anotherClient);
+ makeSimpleRequest(anotherClient, wm);
}
}
@Test
- public void tlsNegotiationTimeout_customValue_clientStartsSuccessfully() throws Exception {
+ public void tlsNegotiationTimeout_customValue_clientStartsSuccessfully(WireMockRuntimeInfo wm) throws Exception {
AttributeMap defaults = AttributeMap.builder().put(SdkHttpConfigurationOption.TRUST_ALL_CERTIFICATES, true).build();
try (SdkAsyncHttpClient client = AwsCrtAsyncHttpClient.builder()
.tlsNegotiationTimeout(Duration.ofSeconds(3))
.buildWithDefaults(defaults)) {
- makeSimpleHttpsRequest(client);
+ makeSimpleHttpsRequest(client, wm);
}
}
- private RecordingResponseHandler makeSimpleHttpsRequest(SdkAsyncHttpClient client) throws Exception {
+ @ParameterizedTest
+ @ValueSource(ints = {0, -1, 1})
+ public void numEventLoopThreads_notGreaterThanOne_shouldThrowException(int value) {
+ assertThatThrownBy(() -> AwsCrtAsyncHttpClient.builder().numEventLoopThreads(value))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("numEventLoopThreads must be greater than 1");
+ }
+
+ @Test
+ public void numEventLoopThreads_null_shouldBeAccepted() {
+ assertThatCode(() -> AwsCrtAsyncHttpClient.builder().numEventLoopThreads(null))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ public void defaultBuilder_sharesStaticDefaultEventLoopGroup() {
+ warmUpStaticDefaultEventLoopGroup();
+ Set before = liveEventLoopGroups();
+
+ try (SdkAsyncHttpClient client = AwsCrtAsyncHttpClient.create();
+ SdkAsyncHttpClient anotherClient = AwsCrtAsyncHttpClient.create()) {
+ assertThat(newEventLoopGroups(before)).isEmpty();
+ }
+ }
+
+ @Test
+ public void numEventLoopThreads_createsPrivateGroupsNotShared() {
+ warmUpStaticDefaultEventLoopGroup();
+ Set before = liveEventLoopGroups();
+
+ try (SdkAsyncHttpClient client = AwsCrtAsyncHttpClient.builder().numEventLoopThreads(2).build();
+ SdkAsyncHttpClient anotherClient = AwsCrtAsyncHttpClient.builder().numEventLoopThreads(2).build()) {
+ assertThat(newEventLoopGroups(before)).hasSize(2);
+ }
+ }
+
+ @Test
+ public void numEventLoopThreads_executesRequest(WireMockRuntimeInfo wm) throws Exception {
+ try (SdkAsyncHttpClient client = AwsCrtAsyncHttpClient.builder().numEventLoopThreads(2).build()) {
+ RecordingResponseHandler recorder = makeSimpleRequest(client, wm);
+ assertThat(recorder.responses().get(0).statusCode()).isEqualTo(200);
+ }
+ }
+
+ @Test
+ public void numEventLoopThreads_closeReleasesPrivateGroup() {
+ warmUpStaticDefaultEventLoopGroup();
+ Set before = liveEventLoopGroups();
+ SdkAsyncHttpClient client = AwsCrtAsyncHttpClient.builder().numEventLoopThreads(2).build();
+ Set privateGroup = newEventLoopGroups(before);
+ assertThat(privateGroup).hasSize(1);
+
+ client.close();
+
+ assertThat(waitForEventLoopGroupsReleased(privateGroup, Duration.ofSeconds(30)))
+ .as("private event-loop group should be released on close")
+ .isTrue();
+ }
+
+ @ParameterizedTest
+ @CsvSource({"4, true", "1, false"})
+ public void numEventLoopThreads_warnsOnlyWhenExcessive(int multipleOfProcessors, boolean expectWarning) {
+ int processors = Math.max(1, Runtime.getRuntime().availableProcessors());
+ int size = Math.max(2, multipleOfProcessors * processors);
+ try (LogCaptor logCaptor = LogCaptor.create(Level.WARN);
+ SdkAsyncHttpClient client = AwsCrtAsyncHttpClient.builder().numEventLoopThreads(size).build()) {
+ if (expectWarning) {
+ assertThat(logCaptor.loggedEvents()).anySatisfy(event -> {
+ assertThat(event.getLevel()).isEqualTo(Level.WARN);
+ assertThat(event.getMessage().getFormattedMessage())
+ .contains("numEventLoopThreads")
+ .contains("private event-loop group");
+ });
+ } else {
+ assertThat(logCaptor.loggedEvents()).noneSatisfy(event ->
+ assertThat(event.getMessage().getFormattedMessage()).contains("numEventLoopThreads"));
+ }
+ }
+ }
+
+ private void warmUpStaticDefaultEventLoopGroup() {
+ // A default client and a private-group client both lazily create the shared static default group (via the host
+ // resolver), so create it up front to keep the before/after group diff stable.
+ AwsCrtAsyncHttpClient.create().close();
+ }
+
+ private RecordingResponseHandler makeSimpleHttpsRequest(SdkAsyncHttpClient client, WireMockRuntimeInfo wm) throws Exception {
String body = randomAlphabetic(10);
- URI uri = URI.create("https://localhost:" + mockServer.httpsPort());
- stubFor(any(urlPathEqualTo("/")).willReturn(aResponse().withBody(body)));
+ URI uri = URI.create("https://localhost:" + wm.getHttpsPort());
+ mockServer.stubFor(any(urlPathEqualTo("/")).willReturn(aResponse().withBody(body)));
SdkHttpRequest request = createRequest(uri);
RecordingResponseHandler recorder = new RecordingResponseHandler();
client.execute(AsyncExecuteRequest.builder()
@@ -123,10 +216,10 @@ private RecordingResponseHandler makeSimpleHttpsRequest(SdkAsyncHttpClient clien
*
* @param client Client to make request with.
*/
- private RecordingResponseHandler makeSimpleRequest(SdkAsyncHttpClient client) throws Exception {
+ private RecordingResponseHandler makeSimpleRequest(SdkAsyncHttpClient client, WireMockRuntimeInfo wm) throws Exception {
String body = randomAlphabetic(10);
- URI uri = URI.create("http://localhost:" + mockServer.port());
- stubFor(any(urlPathEqualTo("/")).willReturn(aResponse().withBody(body)));
+ URI uri = URI.create("http://localhost:" + wm.getHttpPort());
+ mockServer.stubFor(any(urlPathEqualTo("/")).willReturn(aResponse().withBody(body)));
SdkHttpRequest request = createRequest(uri);
RecordingResponseHandler recorder = new RecordingResponseHandler();
client.execute(AsyncExecuteRequest.builder()
diff --git a/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientWireMockTest.java b/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientWireMockTest.java
index 0678d2b9039a..87d2726d6204 100644
--- a/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientWireMockTest.java
+++ b/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientWireMockTest.java
@@ -22,22 +22,29 @@
import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static software.amazon.awssdk.http.SdkHttpConfigurationOption.PROTOCOL;
import static software.amazon.awssdk.http.SdkHttpConfigurationOption.TRUST_ALL_CERTIFICATES;
import static software.amazon.awssdk.http.crt.CrtHttpClientTestUtils.createRequest;
+import static software.amazon.awssdk.http.crt.CrtHttpClientTestUtils.liveEventLoopGroups;
+import static software.amazon.awssdk.http.crt.CrtHttpClientTestUtils.newEventLoopGroups;
+import static software.amazon.awssdk.http.crt.CrtHttpClientTestUtils.waitForEventLoopGroupsReleased;
import com.github.tomakehurst.wiremock.junit.WireMockRule;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.net.URI;
import java.time.Duration;
+import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import org.apache.logging.log4j.Level;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
+import software.amazon.awssdk.crt.CrtResource;
import software.amazon.awssdk.crt.Log;
import software.amazon.awssdk.http.ExecutableHttpRequest;
import software.amazon.awssdk.http.HttpExecuteRequest;
@@ -50,6 +57,7 @@
import software.amazon.awssdk.http.SdkHttpRequest;
import software.amazon.awssdk.metrics.MetricCollection;
import software.amazon.awssdk.metrics.MetricCollector;
+import software.amazon.awssdk.testutils.LogCaptor;
import software.amazon.awssdk.utils.AttributeMap;
public class AwsCrtHttpClientWireMockTest extends SdkHttpClientTestSuite {
@@ -85,6 +93,123 @@ public void invalidProtocol_shouldThrowException() {
.isInstanceOf(UnsupportedOperationException.class);
}
+ @Test
+ public void numEventLoopThreads_zero_shouldThrowException() {
+ assertThatThrownBy(() -> AwsCrtHttpClient.builder().numEventLoopThreads(0))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("numEventLoopThreads must be greater than 1");
+ }
+
+ @Test
+ public void numEventLoopThreads_negative_shouldThrowException() {
+ assertThatThrownBy(() -> AwsCrtHttpClient.builder().numEventLoopThreads(-1))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("numEventLoopThreads must be greater than 1");
+ }
+
+ @Test
+ public void numEventLoopThreads_one_shouldThrowException() {
+ assertThatThrownBy(() -> AwsCrtHttpClient.builder().numEventLoopThreads(1))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("numEventLoopThreads must be greater than 1");
+ }
+
+ @Test
+ public void numEventLoopThreads_null_shouldBeAccepted() {
+ assertThatCode(() -> AwsCrtHttpClient.builder().numEventLoopThreads(null))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ public void defaultBuilder_sharesStaticDefaultEventLoopGroup() {
+ warmUpStaticDefaultEventLoopGroup();
+ Set before = liveEventLoopGroups();
+
+ try (SdkHttpClient client = AwsCrtHttpClient.create();
+ SdkHttpClient anotherClient = AwsCrtHttpClient.create()) {
+ assertThat(newEventLoopGroups(before)).isEmpty();
+ }
+ }
+
+ @Test
+ public void numEventLoopThreads_createsPrivateGroupsNotShared() {
+ warmUpStaticDefaultEventLoopGroup();
+ Set before = liveEventLoopGroups();
+
+ try (SdkHttpClient client = AwsCrtHttpClient.builder().numEventLoopThreads(2).build();
+ SdkHttpClient anotherClient = AwsCrtHttpClient.builder().numEventLoopThreads(2).build()) {
+ assertThat(newEventLoopGroups(before)).hasSize(2);
+ }
+ }
+
+ @Test
+ public void numEventLoopThreads_executesRequest() throws Exception {
+ try (SdkHttpClient client = AwsCrtHttpClient.builder().numEventLoopThreads(2).build()) {
+ HttpExecuteResponse response = makeSimpleRequest(client, null);
+ assertThat(response.httpResponse().statusCode()).isEqualTo(200);
+ }
+ }
+
+ @Test
+ public void numEventLoopThreads_closeReleasesPrivateGroup() {
+ warmUpStaticDefaultEventLoopGroup();
+ Set before = liveEventLoopGroups();
+ SdkHttpClient client = AwsCrtHttpClient.builder().numEventLoopThreads(2).build();
+ Set privateGroup = newEventLoopGroups(before);
+ assertThat(privateGroup).hasSize(1);
+
+ client.close();
+
+ assertThat(waitForEventLoopGroupsReleased(privateGroup, Duration.ofSeconds(30)))
+ .as("private event-loop group should be released on close")
+ .isTrue();
+ }
+
+ @Test
+ public void numEventLoopThreads_excessivelyHigh_logsWarning() {
+ int excessive = 4 * Math.max(1, Runtime.getRuntime().availableProcessors());
+ try (LogCaptor logCaptor = LogCaptor.create(Level.WARN);
+ SdkHttpClient client = AwsCrtHttpClient.builder().numEventLoopThreads(excessive).build()) {
+ assertThat(logCaptor.loggedEvents()).anySatisfy(event -> {
+ assertThat(event.getLevel()).isEqualTo(Level.WARN);
+ assertThat(event.getMessage().getFormattedMessage())
+ .contains("numEventLoopThreads")
+ .contains("private event-loop group");
+ });
+ }
+ }
+
+ @Test
+ public void numEventLoopThreads_normalValue_doesNotLogWarning() {
+ try (LogCaptor logCaptor = LogCaptor.create(Level.WARN);
+ SdkHttpClient client = AwsCrtHttpClient.builder().numEventLoopThreads(2).build()) {
+ assertThat(logCaptor.loggedEvents()).noneSatisfy(event ->
+ assertThat(event.getMessage().getFormattedMessage()).contains("numEventLoopThreads"));
+ }
+ }
+
+ @Test
+ public void http2WithNumEventLoopThreads_throwsAndDoesNotLeakPrivateGroup() {
+ warmUpStaticDefaultEventLoopGroup();
+ Set before = liveEventLoopGroups();
+ AttributeMap attributeMap = AttributeMap.builder()
+ .put(PROTOCOL, Protocol.HTTP2)
+ .build();
+
+ assertThatThrownBy(() -> AwsCrtHttpClient.builder().numEventLoopThreads(2).buildWithDefaults(attributeMap))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThat(newEventLoopGroups(before))
+ .as("HTTP/2 rejection must not leave a private event-loop group behind")
+ .isEmpty();
+ }
+
+ private void warmUpStaticDefaultEventLoopGroup() {
+ // A default client and a private-group client both lazily create the shared static default group (via the host
+ // resolver), so create it up front to keep the before/after group diff stable.
+ AwsCrtHttpClient.create().close();
+ }
+
@Test
public void sendRequest_withCollector_shouldCollectMetrics() throws Exception {
diff --git a/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/CrtHttpClientTestUtils.java b/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/CrtHttpClientTestUtils.java
index d564afd596b8..f09510f20475 100644
--- a/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/CrtHttpClientTestUtils.java
+++ b/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/CrtHttpClientTestUtils.java
@@ -1,8 +1,25 @@
+/*
+ * 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 org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
+import software.amazon.awssdk.crt.CrtResource;
+import software.amazon.awssdk.crt.io.EventLoopGroup;
import software.amazon.awssdk.http.SdkHttpFullRequest;
import software.amazon.awssdk.http.SdkHttpMethod;
import software.amazon.awssdk.http.SdkHttpResponse;
@@ -10,7 +27,11 @@
import java.net.URI;
import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.IdentityHashMap;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicReference;
@@ -18,6 +39,59 @@
public class CrtHttpClientTestUtils {
+ private static final String EVENT_LOOP_GROUP = EventLoopGroup.class.getCanonicalName();
+
+ /**
+ * The {@link EventLoopGroup} native resources created since {@code before} was captured, by identity. Groups are only ever
+ * created synchronously while a client is constructed, so the difference is the exact set of groups the client under test
+ * created, immune to groups from other tests still draining asynchronously in the reused fork. Relies on
+ * {@code aws.crt.debugnative} being set.
+ */
+ static Set newEventLoopGroups(Set before) {
+ Set created = liveEventLoopGroups();
+ created.removeAll(before);
+ return created;
+ }
+
+ static Set liveEventLoopGroups() {
+ Set groups = Collections.newSetFromMap(new IdentityHashMap<>());
+ CrtResource.collectNativeResource(resource -> {
+ if (EVENT_LOOP_GROUP.equals(resource.canonicalName)) {
+ groups.add(resource.getWrapper());
+ }
+ });
+ return groups;
+ }
+
+ /**
+ * Blocks until none of {@code groups} are in the live event-loop-group set, or the timeout elapses. Event-loop groups are
+ * released asynchronously (the bootstrap holds a native reference that drops on its own shutdown callback), so a released
+ * group leaves the live set shortly after {@code close()}. Unlike {@code CrtResource.waitForNoResources()}, this only waits
+ * on the specific groups passed in, so it does not tear down shared static defaults or depend on other tests' resources
+ * having drained.
+ *
+ * @return {@code true} if all groups were released before the timeout.
+ */
+ static boolean waitForEventLoopGroupsReleased(Set groups, Duration timeout) {
+ long deadline = System.nanoTime() + timeout.toNanos();
+ while (System.nanoTime() < deadline) {
+ Set stillLive = liveEventLoopGroups();
+ stillLive.retainAll(groups);
+ if (stillLive.isEmpty()) {
+ return true;
+ }
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+ Set stillLive = liveEventLoopGroups();
+ stillLive.retainAll(groups);
+ return stillLive.isEmpty();
+ }
+
static Subscriber createDummySubscriber() {
return new Subscriber() {
@Override
diff --git a/test/architecture-tests/pom.xml b/test/architecture-tests/pom.xml
index 9568f9f6452b..b94e581370f0 100644
--- a/test/architecture-tests/pom.xml
+++ b/test/architecture-tests/pom.xml
@@ -166,6 +166,11 @@
apache5-client
${awsjavasdk.version}
+
+ software.amazon.awssdk
+ aws-crt-client
+ ${awsjavasdk.version}
+
org.junit.jupiter
junit-jupiter
diff --git a/test/architecture-tests/src/test/java/software/amazon/awssdk/archtests/CodingConventionWithSuppressionTest.java b/test/architecture-tests/src/test/java/software/amazon/awssdk/archtests/CodingConventionWithSuppressionTest.java
index a7867216a65f..26d56dd1fa45 100644
--- a/test/architecture-tests/src/test/java/software/amazon/awssdk/archtests/CodingConventionWithSuppressionTest.java
+++ b/test/architecture-tests/src/test/java/software/amazon/awssdk/archtests/CodingConventionWithSuppressionTest.java
@@ -57,6 +57,10 @@ public class CodingConventionWithSuppressionTest {
ArchUtils.classNameToPattern("software.amazon.awssdk.services.s3.internal.crt.S3CrtResponseHandlerAdapter"),
ArchUtils.classNameToPattern(
"software.amazon.awssdk.services.s3.internal.crt.CrtResponseFileResponseTransformer"),
+ ArchUtils.classNameToPattern("software.amazon.awssdk.http.crt.AwsCrtHttpClientBase"),
+ ArchUtils.classNameToPattern("software.amazon.awssdk.http.crt.internal.AwsCrtConfigurationUtils"),
+ ArchUtils.classNameToPattern(
+ "software.amazon.awssdk.http.crt.internal.response.CrtResponseAdapter"),
ArchUtils.classNameToPattern(RetryableSubAsyncRequestBody.class),
ArchUtils.classNameToPattern(KnownContentLengthAsyncRequestBodySubscriber.class),
ArchUtils.classNameToPattern(UnknownContentLengthAsyncRequestBodySubscriber.class),
@@ -65,6 +69,7 @@ public class CodingConventionWithSuppressionTest {
private static final Set ALLOWED_ERROR_LOG_SUPPRESSION = new HashSet<>(
Arrays.asList(
ArchUtils.classNameToPattern(EmfMetricLoggingPublisher.class),
+ ArchUtils.classNameToPattern("software.amazon.awssdk.http.crt.internal.CrtAsyncRequestExecutor"),
ArchUtils.classWithInnerClassesToPattern(ResponseTransformer.class)));
@Test