fix(netty-nio-client): fix streaming timeout exception decoration so read timeouts are retried (#7048)

Author: joviegas

.changes/next-release/bugfix-NettyNIOHTTPClient-0bec94c.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Netty NIO HTTP Client",
+    "description": "Decorate streaming response publisher failures so S3AsyncClient getObject can retry Netty read timeouts.",
+    "contributor": "goutamadwant"
+}

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

@@ -342,11 +342,12 @@ public void onError(Throwable t) {
                     if (!isDone.compareAndSet(false, true)) {
                         return;
                     }
+                    Throwable throwable = NettyUtils.decorateException(channelContext.channel(), t);
                     try {
                         runAndLogError(channelContext.channel(),
                                        () -> String.format("Subscriber %s threw an exception in onError.", subscriber),
-                                       () -> subscriber.onError(t));
-                        notifyError(t);
+                                       () -> subscriber.onError(throwable));
+                        notifyError(throwable);
                     } finally {
                         runAndLogError(channelContext.channel(), () -> "Could not release channel back to the pool",
                             () -> closeAndRelease(channelContext));

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

@@ -37,13 +37,16 @@
 import io.netty.handler.codec.http.HttpContent;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.timeout.ReadTimeoutException;
 import io.reactivex.Flowable;
+import java.io.IOException;
 import java.net.URI;
 import java.nio.ByteBuffer;
 import java.util.concurrent.CompletableFuture;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.reactivestreams.Publisher;
@@ -167,6 +170,33 @@ public void errorOccurred_shouldInvokeResponseHandler() {
         verify(responseHandler).onError(exception);
     }
 
+    @Test
+    public void onError_whenReadTimeoutException_decoratesToRetryableIOException() {
+        Flowable<HttpContent> testPublisher = Flowable.error(ReadTimeoutException.INSTANCE);
+
+        StreamedHttpResponse streamedHttpResponse = new DefaultStreamedHttpResponse(HttpVersion.HTTP_1_1,
+                                                                                    HttpResponseStatus.ACCEPTED,
+                                                                                    testPublisher);
+
+
+
+        ResponseHandler.PublisherAdapter publisherAdapter = new ResponseHandler.PublisherAdapter(streamedHttpResponse,
+                                                                                                 ctx,
+                                                                                                 requestContext,
+                                                                                                 executeFuture
+        );
+        TestSubscriber subscriber = new TestSubscriber();
+
+        publisherAdapter.subscribe(subscriber);
+
+        ArgumentCaptor<Throwable> errorCaptor = ArgumentCaptor.forClass(Throwable.class);
+        verify(responseHandler).onError(errorCaptor.capture());
+        assertThat(errorCaptor.getValue()).isInstanceOf(IOException.class)
+                                          .hasCauseInstanceOf(ReadTimeoutException.class);
+        assertThat(subscriber.error).isSameAs(errorCaptor.getValue());
+        assertThat(executeFuture).isCompletedExceptionally();
+    }
+
     @Test
     public void subscriptionCancelled_upstreamPublisherCallsOnNext_httpContentReleased() {
         HttpContent firstContent = mock(HttpContent.class);
@@ -290,6 +320,7 @@ static final class TestSubscriber implements Subscriber<ByteBuffer> {
         private Subscription subscription;
         private boolean isCompleted = false;
         private boolean errorOccurred = false;
+        private Throwable error;
 
         @Override
         public void onSubscribe(Subscription s) {
@@ -305,6 +336,7 @@ public void onNext(ByteBuffer byteBuffer) {
         @Override
         public void onError(Throwable t) {
             errorOccurred = true;
+            error = t;
         }
 
         @Override