Improve exception message for high burst traffic using netty client

Author: varunnvs92

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/NettyNioAsyncHttpClient.java

@@ -85,7 +85,7 @@ protected ChannelPool newPool(URI key) {
                                             FixedChannelPool.AcquireTimeoutAction.FAIL,
                                             configuration.connectionAcquisitionTimeout(),
                                             configuration.maxConnectionsPerEndpoint(),
-                                            10_000);
+                                            configuration.maxPendingAcquires());
             }
         };
     }

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/NettySdkHttpClientFactory.java

@@ -46,6 +46,7 @@ public final class NettySdkHttpClientFactory
     private final Duration readTimeout;
     private final Duration writeTimeout;
     private final Duration connectionAcquisitionTimeout;
+    private final Integer maxPendingAcquires;
 
     private NettySdkHttpClientFactory(DefaultBuilder builder) {
         this.standardOptions = builder.standardOptions.build();
@@ -54,6 +55,7 @@ private NettySdkHttpClientFactory(DefaultBuilder builder) {
         this.readTimeout = validateIsWholeSecond(builder.readTimeout, "readTimeout");
         this.writeTimeout = validateIsWholeSecond(builder.writeTimeout, "writeTimeout");
         this.connectionAcquisitionTimeout = builder.connectionAcquisitionTimeout;
+        this.maxPendingAcquires = builder.maxPendingAcquires;
     }
 
     /**
@@ -64,6 +66,14 @@ public Optional<Integer> maxConnectionsPerEndpoint() {
         return Optional.ofNullable(standardOptions.get(MAX_CONNECTIONS));
     }
 
+    /**
+     * @return Optional of the maxPendingAcquires setting.
+     * @see Builder#maxPendingAcquires(Integer)
+     */
+    public Optional<Integer> maxPendingAcquires() {
+        return Optional.ofNullable(maxPendingAcquires);
+    }
+
     /**
      * @return Optional of the writeTimeout setting.
      * @see Builder#writeTimeout(Duration)
@@ -177,6 +187,14 @@ public interface Builder extends CopyableBuilder<Builder, NettySdkHttpClientFact
          */
         Builder maxConnectionsPerEndpoint(Integer maxConnectionsPerEndpoint);
 
+        /**
+         * The maximum number of pending acquires allowed. Once this exceeds, acquire tries will be failed.
+         *
+         * @param maxPendingAcquires Max number of pending acquires
+         * @return This builder for method chaining.
+         */
+        Builder maxPendingAcquires(Integer maxPendingAcquires);
+
         /**
          * The amount of time to wait for a read on a socket before an exception is thrown.
          * <br/>
@@ -259,6 +277,7 @@ private static final class DefaultBuilder implements Builder {
         private Duration readTimeout;
         private Duration writeTimeout;
         private Duration connectionAcquisitionTimeout;
+        private Integer maxPendingAcquires;
 
         private DefaultBuilder(AttributeMap.Builder standardOptions) {
             this.standardOptions = standardOptions;
@@ -270,6 +289,12 @@ public Builder maxConnectionsPerEndpoint(Integer maxConnectionsPerEndpoint) {
             return this;
         }
 
+        @Override
+        public Builder maxPendingAcquires(Integer maxPendingAcquires) {
+            this.maxPendingAcquires = maxPendingAcquires;
+            return this;
+        }
+
         @Override
         public Builder readTimeout(Duration timeout) {
             this.readTimeout = timeout;

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

@@ -34,6 +34,7 @@
 public final class NettyConfiguration {
     private static final Duration DEFAULT_READ_TIMEOUT = Duration.ofSeconds(60);
     private static final Duration DEFAULT_WRITE_TIMEOUT = Duration.ofSeconds(60);
+    private static final Integer MAX_PENDING_ACQUIRE = 10_000;
     private final AttributeMap serviceDefaults;
     private final NettySdkHttpClientFactory factory;
 
@@ -57,6 +58,13 @@ public int maxConnectionsPerEndpoint() {
         return serviceDefaults.get(MAX_CONNECTIONS);
     }
 
+    /**
+     * @see NettySdkHttpClientFactory.Builder#maxPendingAcquires(Integer)
+     */
+    public int maxPendingAcquires() {
+        return factory.maxPendingAcquires().orElse(MAX_PENDING_ACQUIRE);
+    }
+
     @ReviewBeforeRelease("Support disabling strict hostname verification")
     public <T> Optional<T> getConfigurationValue(AttributeMap.Key<T> key) {
         return key == USE_STRICT_HOSTNAME_VERIFICATION ? Optional.empty() : Optional.ofNullable(serviceDefaults.get(key));

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

@@ -34,6 +34,7 @@
 import io.netty.util.concurrent.Future;
 import java.net.URI;
 import java.nio.ByteBuffer;
+import java.util.concurrent.TimeoutException;
 import java.util.function.Supplier;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
@@ -111,13 +112,81 @@ private URI endpoint() {
     private void handleFailure(Supplier<String> msg, Throwable cause) {
         log.error(msg.get(), cause);
         runAndLogError("Exception thrown from AsyncResponseHandler",
-            () -> context.handler().exceptionOccurred(cause));
+            () -> context.handler().exceptionOccurred(modifyHighBurstTrafficException(cause)));
         if (channel != null) {
-            runAndLogError("Unable to release channel back to the pool.",
-                () -> closeAndRelease(channel));
+            runAndLogError("Unable to release channel back to the pool.", () -> closeAndRelease(channel));
         }
     }
 
+    private Throwable modifyHighBurstTrafficException(Throwable originalCause) {
+        String originalMessage = originalCause.getMessage();
+        String newMessage = null;
+
+        if (originalCause instanceof TimeoutException &&
+            originalMessage.contains("Acquire operation took longer")) {
+            newMessage = getMessageForAcquireTimeoutException();
+
+        } else if (originalCause instanceof IllegalStateException &&
+                   originalMessage.contains("Too many outstanding acquire operations")) {
+            newMessage = getMessageForTooManyAcquireOperationsError();
+
+        } else {
+            return originalCause;
+        }
+
+        return new Throwable(newMessage, originalCause);
+    }
+
+
+    private String getMessageForAcquireTimeoutException() {
+        StringBuilder stringBuilder = new StringBuilder();
+
+        stringBuilder
+            .append("Acquire operation took longer than the configured maximum time. This indicates that a request cannot get a "
+                  + "connection from the pool within the specified maximum time. This can be due to high request rate.\n")
+
+            .append("Consider taking any of the following actions to mitigate the issue: increase max connections, "
+                  + "increase acquire timeout, or slowing the request rate.\n")
+
+            .append("Increasing the max connections can increase client throughput (unless the network interface is already "
+                    + "fully utilized), but can eventually start to hit operation system limitations on the number of file "
+                    + "descriptors used by the process. If you already are fully utilizing your network interface or cannot "
+                    + "further increase your connection count, increasing the acquire timeout gives extra time for requests to "
+                    + "acquire a connection before timing out. If the connections doesn't free up, the subsequent requests "
+                    + "will still timeout.\n")
+
+            .append("If the above mechanisms are not able to fix the issue, try smoothing out your requests so that large "
+                    + "traffic bursts cannot overload the client, being more efficient with the number of times you need to "
+                    + "call AWS, or by increasing the number of hosts sending requests.");
+
+        return stringBuilder.toString();
+    }
+
+    private String getMessageForTooManyAcquireOperationsError() {
+        StringBuilder  stringBuilder = new StringBuilder();
+
+        stringBuilder
+            .append("Maximum pending connection acquisitions exceeded. The request rate is too high for the client to keep up.\n")
+
+            .append("Consider taking any of the following actions to mitigate the issue: increase max connections, "
+                  + "increase max pending acquire count, decrease pool lease timeout, or slowing the request rate.\n")
+
+            .append("Increasing the max connections can increase client throughput (unless the network interface is already "
+                    + "fully utilized), but can eventually start to hit operation system limitations on the number of file "
+                    + "descriptors used by the process. If you already are fully utilizing your network interface or cannot "
+                    + "further increase your connection count, increasing the pending acquire count allows extra requests to be "
+                    + "buffered by the client, but can cause additional request latency and higher memory usage. If your request"
+                    + " latency or memory usage is already too high, decreasing the lease timeout will allow requests to fail "
+                    + "more quickly, reducing the number of pending connection acquisitions, but likely won't decrease the total "
+                    + "number of failed requests.\n")
+
+            .append("If the above mechanisms are not able to fix the issue, try smoothing out your requests so that large "
+                    + "traffic bursts cannot overload the client, being more efficient with the number of times you need to call "
+                    + "AWS, or by increasing the number of hosts sending requests.");
+
+        return stringBuilder.toString();
+    }
+
     private static void closeAndRelease(Channel channel) {
         RequestContext requestCtx = channel.attr(REQUEST_CONTEXT_KEY).get();
         channel.close().addListener(ignored -> requestCtx.channelPool().release(channel));

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

@@ -33,6 +33,7 @@
 import static org.apache.commons.lang3.StringUtils.isBlank;
 import static org.apache.commons.lang3.StringUtils.reverse;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
@@ -43,10 +44,13 @@
 import io.netty.channel.nio.NioEventLoopGroup;
 import java.net.URI;
 import java.nio.ByteBuffer;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Stream;
@@ -332,4 +336,60 @@ public Thread newThread(Runnable r) {
             return new Thread(r);
         }
     }
+
+    @Test
+    public void testExceptionMessageChanged_WhenPendingAcquireQueueIsFull() throws Exception {
+        String expectedErrorMsg = "Maximum pending connection acquisitions exceeded.";
+
+        SdkAsyncHttpClient customClient = NettySdkHttpClientFactory.builder()
+                                                                   .maxConnectionsPerEndpoint(1)
+                                                                   .maxPendingAcquires(1)
+                                                                   .build()
+                                                                   .createHttpClient();
+
+        List<CompletableFuture<Void>> futures = new ArrayList<>();
+        for (int i = 0; i < 10; i++) {
+            futures.add(makeSimpleRequestAndReturnResponseHandler(customClient).completeFuture);
+        }
+
+        assertThatThrownBy(() -> {
+            CompletableFuture.allOf(futures.stream().toArray(CompletableFuture[]::new)).join();
+        }).hasMessageContaining(expectedErrorMsg);
+
+        customClient.close();
+    }
+
+
+    @Test
+    public void testExceptionMessageChanged_WhenConnectionTimeoutErrorEncountered() throws Exception {
+        String expectedErrorMsg = "Acquire operation took longer than the configured maximum time. This indicates that a request "
+                                  + "cannot get a connection from the pool within the specified maximum time.";
+
+        SdkAsyncHttpClient customClient = NettySdkHttpClientFactory.builder()
+                                                                   .maxConnectionsPerEndpoint(1)
+                                                                   .connectionTimeout(Duration.ofNanos(1))
+                                                                   .build()
+                                                                   .createHttpClient();
+
+        List<CompletableFuture<Void>> futures = new ArrayList<>();
+        for (int i = 0; i < 2; i++) {
+            futures.add(makeSimpleRequestAndReturnResponseHandler(customClient).completeFuture);
+        }
+
+        assertThatThrownBy(() -> {
+            CompletableFuture.allOf(futures.stream().toArray(CompletableFuture[]::new)).join();
+        }).hasMessageContaining(expectedErrorMsg);
+
+        customClient.close();
+    }
+
+    private RecordingResponseHandler makeSimpleRequestAndReturnResponseHandler(SdkAsyncHttpClient client) throws Exception {
+        String body = randomAlphabetic(10);
+        URI uri = URI.create("http://localhost:" + mockServer.port());
+        stubFor(any(urlPathEqualTo("/")).willReturn(aResponse().withBody(body).withFixedDelay(1000)));
+        SdkHttpRequest request = createRequest(uri);
+        RecordingResponseHandler recorder = new RecordingResponseHandler();
+        client.prepareRequest(request, requestContext, createProvider(""), recorder).run();
+        return recorder;
+    }
 }