Revert "Attempt to fix the flakiness of the codegen-generated-classes-test tests. (#2769)"

This caused intermittent test failures.

This reverts commit dc41c0d.

Author: millems

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/MakeAsyncHttpRequestStage.java

@@ -193,15 +193,18 @@ private CompletableFuture<Response<OutputT>> executeHttpRequest(SdkHttpFullReque
         TimeoutTracker timeoutTracker = setupAttemptTimer(responseFuture, context);
         context.apiCallAttemptTimeoutTracker(timeoutTracker);
 
-        // Forward potential cancellations to the upstream futures that our result future depends on.
-        CompletableFutureUtils.forwardExceptionTo(responseFuture, httpClientFuture);
-        CompletableFutureUtils.forwardExceptionTo(responseFuture, responseHandlerFuture);
+        // Forward the cancellation
+        responseFuture.whenComplete((r, t) -> {
+            if (t != null) {
+                httpClientFuture.completeExceptionally(t);
+            }
+        });
 
-        // When the response handler and HTTP client are done processing the request, use the future completion executor
-        // to complete the future returned by this function.
-        CompletableFuture.allOf(responseHandlerFuture, httpClientFuture).whenCompleteAsync((r, t) -> {
+        // Offload the completion of the future returned from this stage onto
+        // the future completion executor
+        responseHandlerFuture.whenCompleteAsync((r, t) -> {
             if (t == null) {
-                responseFuture.complete(responseHandlerFuture.join());
+                responseFuture.complete(r);
             } else {
                 responseFuture.completeExceptionally(t);
             }
@@ -215,6 +218,7 @@ private CompletableFuture<Void> doExecuteHttpRequest(RequestExecutionContext con
         long callStart = System.nanoTime();
         CompletableFuture<Void> httpClientFuture = sdkAsyncHttpClient.execute(executeRequest);
 
+        // Offload the metrics reporting from this stage onto the future completion executor
         CompletableFuture<Void> result = httpClientFuture.whenComplete((r, t) -> {
             long duration = System.nanoTime() - callStart;
             metricCollector.reportMetric(CoreMetric.SERVICE_CALL_DURATION, Duration.ofNanos(duration));

core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/MakeAsyncHttpRequestStageTest.java

@@ -40,24 +40,20 @@
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
-import software.amazon.awssdk.core.Response;
-import software.amazon.awssdk.core.async.EmptyPublisher;
 import software.amazon.awssdk.core.client.config.SdkAdvancedAsyncClientOption;
 import software.amazon.awssdk.core.client.config.SdkClientConfiguration;
 import software.amazon.awssdk.core.http.ExecutionContext;
 import software.amazon.awssdk.core.http.NoopTestRequest;
 import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
 import software.amazon.awssdk.core.internal.http.HttpClientDependencies;
 import software.amazon.awssdk.core.internal.http.RequestExecutionContext;
-import software.amazon.awssdk.core.internal.http.TransformingAsyncResponseHandler;
 import software.amazon.awssdk.core.internal.http.timers.ClientExecutionAndRequestTimerTestUtils;
 import software.amazon.awssdk.core.internal.util.AsyncResponseHandlerTestUtils;
 import software.amazon.awssdk.http.SdkHttpFullRequest;
 import software.amazon.awssdk.http.SdkHttpMethod;
 import software.amazon.awssdk.http.async.AsyncExecuteRequest;
 import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
 import software.amazon.awssdk.metrics.MetricCollector;
-import software.amazon.awssdk.metrics.NoOpMetricCollector;
 import utils.ValidSdkObjects;
 
 @RunWith(MockitoJUnitRunner.class)
@@ -69,7 +65,7 @@ public class MakeAsyncHttpRequestStageTest {
     @Mock
     private ScheduledExecutorService timeoutExecutor;
 
-    private CompletableFuture<Void> clientExecuteFuture = new CompletableFuture<>();
+    private CompletableFuture<Void> clientExecuteFuture = CompletableFuture.completedFuture(null);
 
     @Mock
     private ScheduledFuture future;
@@ -111,86 +107,8 @@ public void apiCallAttemptTimeoutNotEnabled_shouldNotInvokeExecutor() throws Exc
         verify(timeoutExecutor, never()).schedule(any(Runnable.class), anyLong(), any(TimeUnit.class));
     }
 
-    @Test
-    public void success_stageShouldNotCompleteBeforeHttpClientFutureIsCompleted() throws Exception {
-        TransformingAsyncResponseHandler<Response<Object>> handler =
-            AsyncResponseHandlerTestUtils.noOpResponseHandler();
-
-        stage = new MakeAsyncHttpRequestStage<>(handler, clientDependencies(null));
-        CompletableFuture<SdkHttpFullRequest> requestFuture =
-            CompletableFuture.completedFuture(ValidSdkObjects.sdkHttpFullRequest().build());
-
-        CompletableFuture<?> result = stage.execute(requestFuture, requestContext());
-
-        assertThat(result.isDone()).isFalse();
-        handler.onStream(new EmptyPublisher<>());
-        assertThat(result.isDone()).isFalse();
-        clientExecuteFuture.complete(null);
-        assertThat(result.isDone()).isTrue();
-        assertThat(result.isCompletedExceptionally()).isFalse();
-    }
-
-    @Test
-    public void success_stageShouldNotCompleteBeforeResponseHandlerFutureIsCompleted() throws Exception {
-        TransformingAsyncResponseHandler<Response<Object>> handler =
-            AsyncResponseHandlerTestUtils.noOpResponseHandler();
-
-        stage = new MakeAsyncHttpRequestStage<>(handler, clientDependencies(null));
-        CompletableFuture<SdkHttpFullRequest> requestFuture =
-            CompletableFuture.completedFuture(ValidSdkObjects.sdkHttpFullRequest().build());
-
-        CompletableFuture<?> result = stage.execute(requestFuture, requestContext());
-
-        assertThat(result.isDone()).isFalse();
-        clientExecuteFuture.complete(null);
-        assertThat(result.isDone()).isFalse();
-        handler.onStream(new EmptyPublisher<>());
-        assertThat(result.isDone()).isTrue();
-        assertThat(result.isCompletedExceptionally()).isFalse();
-    }
-
-    @Test
-    public void failure_stageShouldNotCompleteBeforeHttpClientFutureIsCompleted() throws Exception {
-        TransformingAsyncResponseHandler<Response<Object>> handler =
-            AsyncResponseHandlerTestUtils.noOpResponseHandler();
-
-        stage = new MakeAsyncHttpRequestStage<>(handler, clientDependencies(null));
-        CompletableFuture<SdkHttpFullRequest> requestFuture =
-            CompletableFuture.completedFuture(ValidSdkObjects.sdkHttpFullRequest().build());
-
-        CompletableFuture<?> result = stage.execute(requestFuture, requestContext());
-
-        assertThat(result.isDone()).isFalse();
-        handler.onError(new Throwable());
-        assertThat(result.isDone()).isFalse();
-        clientExecuteFuture.complete(null);
-        assertThat(result.isDone()).isTrue();
-        assertThat(result.isCompletedExceptionally()).isTrue();
-    }
-
-    @Test
-    public void failure_stageShouldNotCompleteBeforeResponseHandlerFutureIsCompleted() throws Exception {
-        TransformingAsyncResponseHandler<Response<Object>> handler =
-            AsyncResponseHandlerTestUtils.noOpResponseHandler();
-
-        stage = new MakeAsyncHttpRequestStage<>(handler, clientDependencies(null));
-        CompletableFuture<SdkHttpFullRequest> requestFuture =
-            CompletableFuture.completedFuture(ValidSdkObjects.sdkHttpFullRequest().build());
-
-        CompletableFuture<?> result = stage.execute(requestFuture, requestContext());
-
-        assertThat(result.isDone()).isFalse();
-        clientExecuteFuture.completeExceptionally(new Throwable());
-        assertThat(result.isDone()).isFalse();
-        handler.onStream(new EmptyPublisher<>());
-        assertThat(result.isDone()).isTrue();
-        assertThat(result.isCompletedExceptionally()).isTrue();
-    }
-
     @Test
     public void testExecute_contextContainsMetricCollector_addsChildToExecuteRequest() {
-        clientExecuteFuture.complete(null);
-
         stage = new MakeAsyncHttpRequestStage<>(
                 combinedAsyncResponseHandler(AsyncResponseHandlerTestUtils.noOpResponseHandler(),
                         AsyncResponseHandlerTestUtils.noOpResponseHandler()),
@@ -250,11 +168,9 @@ private HttpClientDependencies clientDependencies(Duration timeout) {
 
     private RequestExecutionContext requestContext() {
         ExecutionContext executionContext = ClientExecutionAndRequestTimerTestUtils.executionContext(ValidSdkObjects.sdkHttpFullRequest().build());
-        RequestExecutionContext requestContext = RequestExecutionContext.builder()
-                                                                        .executionContext(executionContext)
-                                                                        .originalRequest(NoopTestRequest.builder().build())
-                                                                        .build();
-        requestContext.attemptMetricCollector(NoOpMetricCollector.create());
-        return requestContext;
+        return RequestExecutionContext.builder()
+                                      .executionContext(executionContext)
+                                      .originalRequest(NoopTestRequest.builder().build())
+                                      .build();
     }
 }

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

@@ -56,6 +56,7 @@
 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;
 import software.amazon.awssdk.http.SdkHttpResponse;
 import software.amazon.awssdk.http.async.SdkAsyncHttpResponseHandler;
@@ -238,8 +239,10 @@ private void onCancel() {
                         return;
                     }
                     try {
-                        log.warn("Subscriber cancelled before all events were published.");
-                        executeFuture.complete(null);
+                        SdkCancellationException e = new SdkCancellationException(
+                                "Subscriber cancelled before all events were published");
+                        log.warn("Subscriber cancelled before all events were published");
+                        executeFuture.completeExceptionally(e);
                     } finally {
                         runAndLogError("Could not release channel back to the pool",
                             () -> closeAndRelease(channelContext));