Mark a connection as unreusable if there was a 5xx error so that new request will execute on new connection

Author: zoewangg

.changes/next-release/bugfix-NettyNIOHTTPClient-20fc78a.json

@@ -0,0 +1,5 @@
+{
+    "category": "Netty NIO HTTP Client", 
+    "type": "bugfix", 
+    "description": "Mark a connection as unreusable if there was a 5xx server error so that a new request will establish a new connection."
+}

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

@@ -52,6 +52,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.http.HttpStatusFamily;
 import software.amazon.awssdk.http.Protocol;
 import software.amazon.awssdk.http.SdkCancellationException;
 import software.amazon.awssdk.http.SdkHttpFullResponse;
@@ -85,7 +86,7 @@ protected void channelRead0(ChannelHandlerContext channelContext, HttpObject msg
                                                              .statusCode(response.status().code())
                                                              .statusText(response.status().reasonPhrase())
                                                              .build();
-            channelContext.channel().attr(KEEP_ALIVE).set(HttpUtil.isKeepAlive(response));
+            channelContext.channel().attr(KEEP_ALIVE).set(shouldKeepAlive(response));
             requestContext.handler().onHeaders(sdkResponse);
         }
 
@@ -128,6 +129,13 @@ private static void finalizeResponse(RequestContext requestContext, ChannelHandl
         }
     }
 
+    private boolean shouldKeepAlive(HttpResponse response) {
+        if (HttpStatusFamily.of(response.status().code()) == HttpStatusFamily.SERVER_ERROR) {
+            return false;
+        }
+        return HttpUtil.isKeepAlive(response);
+    }
+
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
         RequestContext requestContext = ctx.channel().attr(REQUEST_CONTEXT_KEY).get();

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

@@ -0,0 +1,29 @@
+/*
+ * 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.nio.netty.internal.http2;
+
+import software.amazon.awssdk.annotations.SdkInternalApi;
+
+/**
+ * Exception indicating a connection is terminating
+ */
+@SdkInternalApi
+final class Http2ConnectionTerminatingException extends RuntimeException {
+
+    Http2ConnectionTerminatingException(String message) {
+        super(message);
+    }
+}

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

@@ -426,7 +426,7 @@ public void channelInactive(ChannelHandlerContext ctx) {
 
         @Override
         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
-            if (cause instanceof Http2StreamExceptionHandler.Http2StreamIoException) {
+            if (cause instanceof Http2ConnectionTerminatingException) {
                 closeConnectionToNewRequests(ctx, cause);
             } else {
                 closeAndReleaseParent(ctx, cause);

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

@@ -19,7 +19,6 @@
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
-import io.netty.handler.codec.http2.Http2Stream;
 import io.netty.handler.timeout.TimeoutException;
 import java.io.IOException;
 import software.amazon.awssdk.annotations.SdkInternalApi;
@@ -31,7 +30,7 @@
 @ChannelHandler.Sharable
 @SdkInternalApi
 public final class Http2StreamExceptionHandler extends ChannelInboundHandlerAdapter {
-    private static final Logger log = Logger.loggerFor(Http2Stream.class);
+    private static final Logger log = Logger.loggerFor(Http2StreamExceptionHandler.class);
     private static final Http2StreamExceptionHandler INSTANCE = new Http2StreamExceptionHandler();
 
     private Http2StreamExceptionHandler() {
@@ -46,8 +45,9 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         if (isIoError(cause) && ctx.channel().parent() != null) {
             Channel parent = ctx.channel().parent();
             log.debug(() -> "An I/O error occurred on an Http2 stream, notifying the connection channel " + parent);
-            parent.pipeline().fireExceptionCaught(new Http2StreamIoException("An I/O error occurred on an associated Http2 "
-                                                                             + "stream"));
+            parent.pipeline().fireExceptionCaught(new Http2ConnectionTerminatingException("An I/O error occurred on an "
+                                                                                          + "associated Http2 "
+                                                                                          + "stream " + ctx.channel()));
         }
 
         ctx.fireExceptionCaught(cause);
@@ -56,10 +56,4 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
     private boolean isIoError(Throwable cause) {
         return cause instanceof TimeoutException || cause instanceof IOException;
     }
-
-    static final class Http2StreamIoException extends IOException {
-        Http2StreamIoException(String message) {
-            super(message);
-        }
-    }
 }

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

@@ -16,11 +16,13 @@
 package software.amazon.awssdk.http.nio.netty.internal.http2;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.handler.codec.http.DefaultHttpContent;
 import io.netty.handler.codec.http.DefaultLastHttpContent;
 import io.netty.handler.codec.http.HttpObject;
+import io.netty.handler.codec.http.HttpResponse;
 import io.netty.handler.codec.http2.Http2DataFrame;
 import io.netty.handler.codec.http2.Http2Error;
 import io.netty.handler.codec.http2.Http2Exception;
@@ -30,13 +32,16 @@
 import io.netty.handler.codec.http2.HttpConversionUtil;
 import java.io.IOException;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.http.HttpStatusFamily;
+import software.amazon.awssdk.utils.Logger;
 
 /**
  * Converts {@link Http2Frame}s to {@link HttpObject}s. Ignores the majority of {@link Http2Frame}s like PING
  * or SETTINGS.
  */
 @SdkInternalApi
 public class Http2ToHttpInboundAdapter extends SimpleChannelInboundHandler<Http2Frame> {
+    private static final Logger log = Logger.loggerFor(Http2ToHttpInboundAdapter.class);
 
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, Http2Frame frame) throws Exception {
@@ -54,7 +59,22 @@ protected void channelRead0(ChannelHandlerContext ctx, Http2Frame frame) throws
     }
 
     private void onHeadersRead(Http2HeadersFrame headersFrame, ChannelHandlerContext ctx) throws Http2Exception {
-        ctx.fireChannelRead(HttpConversionUtil.toHttpResponse(headersFrame.stream().id(), headersFrame.headers(), true));
+
+        HttpResponse httpResponse = HttpConversionUtil.toHttpResponse(headersFrame.stream().id(), headersFrame.headers(), true);
+        ctx.fireChannelRead(httpResponse);
+
+        if (HttpStatusFamily.of(httpResponse.status().code()) == HttpStatusFamily.SERVER_ERROR) {
+            fireConnectionExceptionForServerError(ctx);
+        }
+    }
+
+    private void fireConnectionExceptionForServerError(ChannelHandlerContext ctx) {
+        if (ctx.channel().parent() != null) {
+            Channel parent = ctx.channel().parent();
+            log.debug(() -> "A 5xx server error occurred on an Http2 stream, notifying the connection channel " + ctx.channel());
+            parent.pipeline().fireExceptionCaught(new Http2ConnectionTerminatingException("A 5xx server error occurred on an "
+                                                                                          + "Http2 stream " + ctx.channel()));
+        }
     }
 
     private void onDataRead(Http2DataFrame dataFrame, ChannelHandlerContext ctx) throws Http2Exception {

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

@@ -0,0 +1,63 @@
+/*
+ * 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.nio.netty;
+
+
+import io.reactivex.Flowable;
+import java.nio.ByteBuffer;
+import java.util.concurrent.CompletableFuture;
+import org.reactivestreams.Publisher;
+import software.amazon.awssdk.http.SdkHttpFullRequest;
+import software.amazon.awssdk.http.SdkHttpMethod;
+import software.amazon.awssdk.http.SdkHttpResponse;
+import software.amazon.awssdk.http.async.AsyncExecuteRequest;
+import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
+import software.amazon.awssdk.http.async.SdkAsyncHttpResponseHandler;
+
+public class TestUtils {
+
+    public static CompletableFuture<Void> sendGetRequest(int serverPort, SdkAsyncHttpClient client) {
+        AsyncExecuteRequest req = AsyncExecuteRequest.builder()
+                                                     .responseHandler(new SdkAsyncHttpResponseHandler() {
+                                                         private SdkHttpResponse headers;
+
+                                                         @Override
+                                                         public void onHeaders(SdkHttpResponse headers) {
+                                                             this.headers = headers;
+                                                         }
+
+                                                         @Override
+                                                         public void onStream(Publisher<ByteBuffer> stream) {
+                                                             Flowable.fromPublisher(stream).forEach(b -> {
+                                                             });
+                                                         }
+
+                                                         @Override
+                                                         public void onError(Throwable error) {
+                                                         }
+                                                     })
+                                                     .request(SdkHttpFullRequest.builder()
+                                                                                .method(SdkHttpMethod.GET)
+                                                                                .protocol("https")
+                                                                                .host("localhost")
+                                                                                .port(serverPort)
+                                                                                .build())
+                                                     .requestContentPublisher(new EmptyPublisher())
+                                                     .build();
+
+        return client.execute(req);
+    }
+}