Fixed an issue in AWS CRT-based S3 client where a GetObject request may hang if streaming failed mid request

Author: zoewangg

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java

@@ -111,6 +111,7 @@ private S3MetaRequestWrapper s3MetaRequest() {
 
     @Override
     public void onResponseHeaders(int statusCode, HttpHeader[] headers) {
+        log.debug(() -> "Received response header with status code " + statusCode);
         // Note, we cannot call responseHandler.onHeaders() here because the response status code and headers may not represent
         // whether the request has succeeded or not (e.g. if this is for a HeadObject call that CRT calls under the hood). We
         // need to rely on onResponseBody/onFinished being called to determine this.
@@ -150,6 +151,7 @@ public int onResponseBody(ByteBuffer bodyBytesIn, long objectRangeStart, long ob
     @Override
     public void onFinished(S3FinishedResponseContext context) {
         int crtCode = context.getErrorCode();
+        log.debug(() -> "Request finished with code: " + crtCode);
         if (crtCode != CRT.AWS_CRT_SUCCESS) {
             handleError(context);
         } else {
@@ -192,6 +194,19 @@ private void handleIoError(S3FinishedResponseContext context, int crtCode) {
             SdkClientException.create("Failed to send the request: " +
                                       CRT.awsErrorString(crtCode), cause);
         failResponseHandlerAndFuture(sdkClientException);
+        notifyResponsePublisherErrorIfNeeded(sdkClientException);
+    }
+
+    private void notifyResponsePublisherErrorIfNeeded(Throwable error) {
+        if (responseHandlingInitiated) {
+            responsePublisher.error(error).handle((ignore, throwable) -> {
+                if (throwable != null) {
+                    log.error(() -> "Exception thrown in responsePublisher#error", throwable);
+                    return null;
+                }
+                return null;
+            });
+        }
     }
 
     private void handleServiceError(int responseStatus, HttpHeader[] headers, byte[] errorPayload) {
@@ -204,6 +219,7 @@ private void handleServiceError(int responseStatus, HttpHeader[] headers, byte[]
                 SdkClientException.create("Request failed during the transfer due to an error returned from S3");
             s3Exception.addSuppressed(sdkClientException);
             failResponseHandlerAndFuture(s3Exception);
+            notifyResponsePublisherErrorIfNeeded(s3Exception);
         } else {
             initiateResponseHandling(errorResponse.build());
             onErrorResponseComplete(errorPayload);