Merge pull request #1905 from aws/millem/fix-release

Revert "Validate that the response content-length header matches what…

Author: millems

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

@@ -70,9 +70,6 @@ public final class ChannelAttributeKey {
     public static final AttributeKey<Http2FrameStream> HTTP2_FRAME_STREAM = NettyUtils.getOrCreateAttributeKey(
         "aws.http.nio.netty.async.http2FrameStream");
 
-    public static final AttributeKey<ChannelDiagnostics> CHANNEL_DIAGNOSTICS = NettyUtils.getOrCreateAttributeKey(
-        "aws.http.nio.netty.async.channelDiagnostics");
-
     /**
      * {@link AttributeKey} to keep track of whether we should close the connection after this request
      * has completed.
@@ -91,12 +88,6 @@ public final class ChannelAttributeKey {
     static final AttributeKey<Boolean> RESPONSE_COMPLETE_KEY = NettyUtils.getOrCreateAttributeKey(
         "aws.http.nio.netty.async.responseComplete");
 
-    static final AttributeKey<Long> RESPONSE_CONTENT_LENGTH = NettyUtils.getOrCreateAttributeKey(
-        "aws.http.nio.netty.async.responseContentLength");
-
-    static final AttributeKey<Long> RESPONSE_DATA_READ = NettyUtils.getOrCreateAttributeKey(
-        "aws.http.nio.netty.async.responseDataRead");
-
     /**
      * {@link AttributeKey} to keep track of whether we have received the {@link LastHttpContent}.
      */

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

@@ -15,7 +15,6 @@
 
 package software.amazon.awssdk.http.nio.netty.internal;
 
-import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.CHANNEL_DIAGNOSTICS;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.HTTP2_CONNECTION;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.HTTP2_INITIAL_WINDOW_SIZE;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.PROTOCOL_FUTURE;
@@ -89,7 +88,6 @@ public ChannelPipelineInitializer(Protocol protocol,
 
     @Override
     public void channelCreated(Channel ch) {
-        ch.attr(CHANNEL_DIAGNOSTICS).set(new ChannelDiagnostics(ch));
         ch.attr(PROTOCOL_FUTURE).set(new CompletableFuture<>());
         ChannelPipeline pipeline = ch.pipeline();
         if (sslCtx != null) {

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

@@ -16,17 +16,15 @@
 package software.amazon.awssdk.http.nio.netty.internal;
 
 import static software.amazon.awssdk.http.HttpMetric.CONCURRENCY_ACQUIRE_DURATION;
-import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.CHANNEL_DIAGNOSTICS;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.EXECUTE_FUTURE_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.EXECUTION_ID_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.IN_USE;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.KEEP_ALIVE;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.LAST_HTTP_CONTENT_RECEIVED_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.REQUEST_CONTEXT_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_COMPLETE_KEY;
-import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_CONTENT_LENGTH;
-import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_DATA_READ;
 import static software.amazon.awssdk.http.nio.netty.internal.NettyRequestMetrics.measureTimeTaken;
+import static software.amazon.awssdk.http.nio.netty.internal.utils.NettyUtils.CLOSED_CHANNEL_MESSAGE;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
@@ -40,18 +38,22 @@
 import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.timeout.ReadTimeoutException;
 import io.netty.handler.timeout.ReadTimeoutHandler;
+import io.netty.handler.timeout.WriteTimeoutException;
 import io.netty.handler.timeout.WriteTimeoutHandler;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 import io.netty.util.concurrent.Promise;
 import java.io.IOException;
 import java.net.URI;
 import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
 import java.time.Duration;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
 import org.reactivestreams.Publisher;
@@ -174,13 +176,9 @@ private void makeRequestListener(Future<Channel> channelFuture) {
         if (channelFuture.isSuccess()) {
             channel = channelFuture.getNow();
             NettyUtils.doInEventLoop(channel.eventLoop(), () -> {
-                try {
-                    configureChannel();
-                    configurePipeline();
+                configureChannel();
+                if (tryConfigurePipeline()) {
                     makeRequest();
-                } catch (Throwable t) {
-                    closeAndRelease(channel);
-                    handleFailure(channel, () -> "Failed to initiate request to " + endpoint(), t);
                 }
             });
         } else {
@@ -194,13 +192,10 @@ private void configureChannel() {
         channel.attr(REQUEST_CONTEXT_KEY).set(context);
         channel.attr(RESPONSE_COMPLETE_KEY).set(false);
         channel.attr(LAST_HTTP_CONTENT_RECEIVED_KEY).set(false);
-        channel.attr(RESPONSE_CONTENT_LENGTH).set(null);
-        channel.attr(RESPONSE_DATA_READ).set(null);
-        channel.attr(CHANNEL_DIAGNOSTICS).get().incrementRequestCount();
         channel.config().setOption(ChannelOption.AUTO_READ, false);
     }
 
-    private void configurePipeline() throws IOException {
+    private boolean tryConfigurePipeline() {
         Protocol protocol = ChannelAttributeKey.getProtocolNow(channel);
         ChannelPipeline pipeline = channel.pipeline();
 
@@ -215,7 +210,10 @@ private void configurePipeline() throws IOException {
                 requestAdapter = REQUEST_ADAPTER_HTTP1_1;
                 break;
             default:
-                throw new IOException("Unknown protocol: " + protocol);
+                String errorMsg = "Unknown protocol: " + protocol;
+                closeAndRelease(channel);
+                handleFailure(channel, () -> errorMsg, new RuntimeException(errorMsg));
+                return false;
         }
 
         pipeline.addLast(LastHttpContentHandler.create());
@@ -229,8 +227,13 @@ private void configurePipeline() throws IOException {
         // handler (which will monitor for it going inactive from now on).
         // Make sure it's active here, or the request will never complete: https://github.com/aws/aws-sdk-java-v2/issues/1207
         if (!channel.isActive()) {
-            throw new IOException(NettyUtils.closedChannelMessage(channel));
+            String errorMessage = "Channel was closed before it could be written to.";
+            closeAndRelease(channel);
+            handleFailure(channel, () -> errorMessage, new IOException(errorMessage));
+            return false;
         }
+
+        return true;
     }
 
     private void makeRequest() {
@@ -305,11 +308,80 @@ private URI endpoint() {
 
     private void handleFailure(Channel channel, Supplier<String> msgSupplier, Throwable cause) {
         log.debug(channel, msgSupplier, cause);
-        cause = NettyUtils.decorateException(channel, cause);
+        cause = decorateException(cause);
         context.handler().onError(cause);
         executeFuture.completeExceptionally(cause);
     }
 
+    private Throwable decorateException(Throwable originalCause) {
+        if (isAcquireTimeoutException(originalCause)) {
+            return new Throwable(getMessageForAcquireTimeoutException(), originalCause);
+        } else if (isTooManyPendingAcquiresException(originalCause)) {
+            return new Throwable(getMessageForTooManyAcquireOperationsError(), originalCause);
+        } else if (originalCause instanceof ReadTimeoutException) {
+            return new IOException("Read timed out", originalCause);
+        } else if (originalCause instanceof WriteTimeoutException) {
+            return new IOException("Write timed out", originalCause);
+        } else if (originalCause instanceof ClosedChannelException) {
+            return new IOException(CLOSED_CHANNEL_MESSAGE, originalCause);
+        }
+
+        return originalCause;
+    }
+
+    private boolean isAcquireTimeoutException(Throwable originalCause) {
+        String message = originalCause.getMessage();
+        return originalCause instanceof TimeoutException &&
+                message != null &&
+                message.contains("Acquire operation took longer");
+    }
+
+    private boolean isTooManyPendingAcquiresException(Throwable originalCause) {
+        String message = originalCause.getMessage();
+        return originalCause instanceof IllegalStateException &&
+               message != null &&
+               originalCause.getMessage().contains("Too many outstanding acquire operations");
+    }
+
+    private String getMessageForAcquireTimeoutException() {
+        return "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"
+
+                + "Consider taking any of the following actions to mitigate the issue: increase max connections, "
+                + "increase acquire timeout, or slowing the request rate.\n"
+
+                + "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"
+
+                + "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.";
+    }
+
+    private String getMessageForTooManyAcquireOperationsError() {
+        return "Maximum pending connection acquisitions exceeded. The request rate is too high for the client to keep up.\n"
+
+                + "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"
+
+                + "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"
+
+                + "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.";
+    }
+
     /**
      * Close and release the channel back to the pool.
      *