Fixed a race condition in AWS CRT-based S3 client that could cause e… (#5441)

* Fixed a race condition in AWS CRT-based S3 client that could cause  error to be thrown.

* Fix checkstyle error

Author: zoewangg

.changes/next-release/bugfix-AWSCRTbasedS3Client-856bc6f.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS CRT-based S3 Client",
+    "contributor": "",
+    "description": "Fixed a race condition in AWS CRT-based S3 client that could cause `s3metaRequest is not initialized yet` error to be thrown."
+}

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

@@ -18,8 +18,13 @@
 import static software.amazon.awssdk.utils.FunctionalUtils.runAndLogError;
 
 import java.nio.ByteBuffer;
+import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.annotations.SdkTestInternalApi;
 import software.amazon.awssdk.core.async.listener.PublisherListener;
 import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.crt.CRT;
@@ -38,6 +43,7 @@
 @SdkInternalApi
 public final class S3CrtResponseHandlerAdapter implements S3MetaRequestResponseHandler {
     private static final Logger log = Logger.loggerFor(S3CrtResponseHandlerAdapter.class);
+    private static final Duration META_REQUEST_TIMEOUT = Duration.ofSeconds(10);
     private final CompletableFuture<Void> resultFuture;
     private final SdkAsyncHttpResponseHandler responseHandler;
 
@@ -47,13 +53,23 @@ public final class S3CrtResponseHandlerAdapter implements S3MetaRequestResponseH
     private final CompletableFuture<S3MetaRequestWrapper> metaRequestFuture;
 
     private final PublisherListener<S3MetaRequestProgress> progressListener;
+    private final Duration s3MetaRequestTimeout;
 
     private volatile boolean responseHandlingInitiated;
 
     public S3CrtResponseHandlerAdapter(CompletableFuture<Void> executeFuture,
                                        SdkAsyncHttpResponseHandler responseHandler,
                                        PublisherListener<S3MetaRequestProgress> progressListener,
                                        CompletableFuture<S3MetaRequestWrapper> metaRequestFuture) {
+        this(executeFuture, responseHandler, progressListener, metaRequestFuture, META_REQUEST_TIMEOUT);
+    }
+
+    @SdkTestInternalApi
+    public S3CrtResponseHandlerAdapter(CompletableFuture<Void> executeFuture,
+                                       SdkAsyncHttpResponseHandler responseHandler,
+                                       PublisherListener<S3MetaRequestProgress> progressListener,
+                                       CompletableFuture<S3MetaRequestWrapper> metaRequestFuture,
+                                       Duration s3MetaRequestTimeout) {
         this.resultFuture = executeFuture;
         this.metaRequestFuture = metaRequestFuture;
 
@@ -71,14 +87,20 @@ public S3CrtResponseHandlerAdapter(CompletableFuture<Void> executeFuture,
 
         this.responseHandler = responseHandler;
         this.progressListener = progressListener == null ? new NoOpPublisherListener() : progressListener;
+        this.s3MetaRequestTimeout = s3MetaRequestTimeout;
     }
 
     private S3MetaRequestWrapper s3MetaRequest() {
-        if (!metaRequestFuture.isDone()) {
-            return null;
+        try {
+            return metaRequestFuture.get(s3MetaRequestTimeout.toMillis(), TimeUnit.MILLISECONDS);
+        } catch (ExecutionException | TimeoutException e) {
+            failResponseHandlerAndFuture(
+                new RuntimeException("Timeout waiting for metaRequest to be ready", e));
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            failResponseHandlerAndFuture(new RuntimeException(e));
         }
-
-        return metaRequestFuture.join();
+        return null;
     }
 
     @Override
@@ -110,9 +132,6 @@ public int onResponseBody(ByteBuffer bodyBytesIn, long objectRangeStart, long ob
 
             S3MetaRequestWrapper metaRequest = s3MetaRequest();
             if (metaRequest == null) {
-                // should not happen
-                failResponseHandlerAndFuture(SdkClientException.create("Unexpected exception occurred: s3metaRequest is not "
-                                                                       + "initialized yet"));
                 return;
             }
             metaRequest.incrementReadWindow(bytesReceived);

services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterTest.java

@@ -26,6 +26,7 @@
 
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import org.junit.Before;
@@ -91,6 +92,21 @@ public void successfulResponse_shouldCompleteFutureSuccessfully() throws Excepti
         verify(sdkResponseHandler).onHeaders(any(SdkHttpResponse.class));
     }
 
+    @Test
+    public void s3MetaRequestNotFinish_shouldFailFuture() throws Exception {
+        S3CrtResponseHandlerAdapter responseHandlerAdapter = new S3CrtResponseHandlerAdapter(future,
+                                                                                             sdkResponseHandler,
+                                                                                             null,
+                                                                                             new CompletableFuture<>(),
+                                                                                             Duration.ofMillis(10));
+        int statusCode = 200;
+        responseHandlerAdapter.onResponseHeaders(statusCode, new HttpHeader[0]);
+        responseHandlerAdapter.onResponseBody(ByteBuffer.wrap("helloworld1".getBytes()), 1, 2);
+
+        assertThatThrownBy(() -> future.get(5, TimeUnit.SECONDS))
+            .hasMessageContaining("Timeout waiting for metaRequest");
+    }
+
     @Test
     public void nullByteBuffer_shouldCompleteFutureExceptionally() {
         HttpHeader[] httpHeaders = new HttpHeader[2];