Add apiCallTimeout and apiCallAttemptTimeout for synchronous operations

Author: zoewangg

.changes/next-release/feature-AWSSDKforJavav2-fade6e0.json

@@ -0,0 +1,5 @@
+{
+    "category": "AWS SDK for Java v2", 
+    "type": "feature", 
+    "description": "Add apiCallTimeout and apiCallAttemptTimeout feature for synchronous calls."
+}

core/sdk-core/src/it/java/software/amazon/awssdk/core/http/timers/client/AbortedExceptionClientExecutionTimerIntegrationTest.java

@@ -15,51 +15,27 @@
 
 package software.amazon.awssdk.core.http.timers.client;
 
-import static org.hamcrest.Matchers.instanceOf;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static software.amazon.awssdk.core.internal.http.timers.ClientExecutionAndRequestTimerTestUtils.createMockGetRequest;
 import static software.amazon.awssdk.core.internal.http.timers.ClientExecutionAndRequestTimerTestUtils.execute;
-import static software.amazon.awssdk.core.internal.http.timers.ClientExecutionAndRequestTimerTestUtils.interruptCurrentThreadAfterDelay;
 
-import java.io.InputStream;
-import java.util.Arrays;
+import java.time.Duration;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
-import software.amazon.awssdk.annotations.ReviewBeforeRelease;
 import software.amazon.awssdk.core.exception.AbortedException;
 import software.amazon.awssdk.core.exception.ApiCallTimeoutException;
-import software.amazon.awssdk.core.exception.SdkClientException;
-import software.amazon.awssdk.core.http.ExecutionContext;
-import software.amazon.awssdk.core.http.MockServerTestBase;
-import software.amazon.awssdk.core.http.NoopTestRequest;
-import software.amazon.awssdk.core.http.server.MockServer;
-import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
-import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
 import software.amazon.awssdk.core.internal.http.AmazonSyncHttpClient;
-import software.amazon.awssdk.core.internal.http.request.SlowExecutionInterceptor;
-import software.amazon.awssdk.core.internal.http.response.DummyResponseHandler;
-import software.amazon.awssdk.core.interceptor.ExecutionInterceptorChain;
-import software.amazon.awssdk.core.interceptor.InterceptorContext;
-import software.amazon.awssdk.core.signer.NoOpSigner;
 import software.amazon.awssdk.http.AbortableCallable;
-import software.amazon.awssdk.http.AbortableInputStream;
 import software.amazon.awssdk.http.SdkHttpClient;
 import software.amazon.awssdk.http.SdkHttpFullResponse;
 import utils.HttpTestUtils;
 
 @RunWith(MockitoJUnitRunner.class)
-@Ignore
-@ReviewBeforeRelease("add it back once execution time out is added back")
-public class AbortedExceptionClientExecutionTimerIntegrationTest extends MockServerTestBase {
+public class AbortedExceptionClientExecutionTimerIntegrationTest  {
 
     private AmazonSyncHttpClient httpClient;
 
@@ -72,17 +48,14 @@ public class AbortedExceptionClientExecutionTimerIntegrationTest extends MockSer
     @Before
     public void setup() throws Exception {
         when(sdkHttpClient.prepareRequest(any())).thenReturn(abortableCallable);
-        httpClient = HttpTestUtils.testClientBuilder().httpClient(sdkHttpClient).build();
+        httpClient = HttpTestUtils.testClientBuilder().httpClient(sdkHttpClient)
+                                  .apiCallTimeout(Duration.ofMillis(1000))
+                                  .build();
         when(abortableCallable.call()).thenReturn(SdkHttpFullResponse.builder()
                                                                      .statusCode(200)
                                                                      .build());
     }
 
-    @Override
-    protected MockServer buildMockServer() {
-        return new MockServer(MockServer.DummyResponseServerBehavior.build(200, "Hi", "Dummy response"));
-    }
-
     @Test(expected = AbortedException.class)
     public void clientExecutionTimeoutEnabled_aborted_exception_occurs_timeout_not_expired() throws Exception {
         when(abortableCallable.call()).thenThrow(AbortedException.builder().build());
@@ -100,44 +73,4 @@ public void clientExecutionTimeoutEnabled_aborted_exception_occurs_timeout_expir
 
         execute(httpClient, createMockGetRequest());
     }
-
-    /**
-     * Tests that a streaming operation has it's request properly cleaned up if the client is interrupted after the
-     * response is received.
-     *
-     * see TT0070103230
-     */
-    @Test
-    public void clientInterruptedDuringResponseHandlers_DoesNotLeakConnection() throws Exception {
-        InputStream mockContent = mock(InputStream.class);
-        when(abortableCallable.call()).thenReturn(SdkHttpFullResponse.builder()
-                                                                     .statusCode(200)
-                                                                     .content(AbortableInputStream.create(mockContent))
-                                                                     .build());
-        interruptCurrentThreadAfterDelay(1000);
-        try {
-            requestBuilder()
-                    .originalRequest(NoopTestRequest.builder().build())
-                    .executionContext(withInterceptors(new SlowExecutionInterceptor().afterTransmissionWaitInSeconds(10)))
-                    .execute(new DummyResponseHandler().leaveConnectionOpen());
-            fail("Expected exception");
-        } catch (SdkClientException e) {
-            assertThat(e.getCause(), instanceOf(InterruptedException.class));
-        }
-
-        verify(mockContent).close();
-    }
-
-    private AmazonSyncHttpClient.RequestExecutionBuilder requestBuilder() {
-        return httpClient.requestExecutionBuilder().request(newGetRequest());
-    }
-
-    private ExecutionContext withInterceptors(ExecutionInterceptor... requestHandlers) {
-        return ExecutionContext.builder()
-                               .signer(new NoOpSigner())
-                               .executionAttributes(new ExecutionAttributes())
-                               .interceptorContext(InterceptorContext.builder().request(NoopTestRequest.builder().build()).build())
-                               .interceptorChain(new ExecutionInterceptorChain(Arrays.asList(requestHandlers)))
-                               .build();
-    }
-}
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/client/config/ClientOverrideConfiguration.java

@@ -28,6 +28,7 @@
 import software.amazon.awssdk.annotations.SdkPublicApi;
 import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
 import software.amazon.awssdk.core.retry.RetryPolicy;
+import software.amazon.awssdk.core.sync.ResponseTransformer;
 import software.amazon.awssdk.utils.AttributeMap;
 import software.amazon.awssdk.utils.CollectionUtils;
 import software.amazon.awssdk.utils.ToString;
@@ -299,9 +300,14 @@ default Builder retryPolicy(Consumer<RetryPolicy.Builder> retryPolicy) {
          * requests that don't get aborted until several seconds after the timer has been breached. Because of this, the client
          * execution timeout feature should not be used when absolute precision is needed.
          *
+         * <p>
+         * For synchronous streaming operations, implementations of {@link ResponseTransformer} must handle interrupt
+         * properly to allow the the SDK to timeout the request in a timely manner.
+         *
          * <p>This may be used together with {@link #apiCallAttemptTimeout()} to enforce both a timeout on each individual HTTP
          * request (i.e. each retry) and the total time spent on all requests across retries (i.e. the 'api call' time).
          *
+         *
          * @see ClientOverrideConfiguration#apiCallTimeout()
          */
         Builder apiCallTimeout(Duration apiCallTimeout);
@@ -314,8 +320,11 @@ default Builder retryPolicy(Consumer<RetryPolicy.Builder> retryPolicy) {
          *
          * <p>The request timeout feature doesn't have strict guarantees on how quickly a request is aborted when the timeout is
          * breached. The typical case aborts the request within a few milliseconds but there may occasionally be requests that
-         * don't get aborted until several seconds after the timer has been breached. Because of this, the request timeout
-         * feature should not be used when absolute precision is needed.
+         * don't get aborted until several seconds after the timer has been breached. Because of this, the api call attempt
+         * timeout feature should not be used when absolute precision is needed.
+         *
+         * <p>For synchronous streaming operations, the process in {@link ResponseTransformer} is not timed and will not
+         * be aborted.
          *
          * <p>This may be used together with {@link #apiCallTimeout()} to enforce both a timeout on each individual HTTP
          * request (i.e. each retry) and the total time spent on all requests across retries (i.e. the 'api call' time).
@@ -461,4 +470,4 @@ public ClientOverrideConfiguration build() {
             return new ClientOverrideConfiguration(this);
         }
     }
-}
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/exception/AbortedException.java

@@ -26,10 +26,14 @@
 @SdkPublicApi
 public final class AbortedException extends SdkClientException {
 
-    protected AbortedException(Builder b) {
+    private AbortedException(Builder b) {
         super(b);
     }
 
+    public static AbortedException create(String message) {
+        return builder().message(message).build();
+    }
+
     public static AbortedException create(String message, Throwable cause) {
         return builder().message(message).cause(cause).build();
     }

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/AmazonSyncHttpClient.java

@@ -30,6 +30,8 @@
 import software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.AfterExecutionInterceptorsStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.AfterTransmissionExecutionInterceptorsStage;
+import software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallAttemptTimeoutTrackingStage;
+import software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallTimeoutTrackingStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.ApplyTransactionIdStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.ApplyUserAgentStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.BeforeTransmissionExecutionInterceptorsStage;
@@ -45,6 +47,7 @@
 import software.amazon.awssdk.core.internal.http.pipeline.stages.MoveParametersToBodyStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.SigningStage;
+import software.amazon.awssdk.core.internal.http.pipeline.stages.TimeoutExceptionHandlingStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.UnwrapResponseContainer;
 import software.amazon.awssdk.core.internal.retry.SdkDefaultRetrySetting;
 import software.amazon.awssdk.core.internal.util.CapacityManager;
@@ -245,26 +248,29 @@ public <OutputT> OutputT execute(HttpResponseHandler<OutputT> responseHandler) {
                 return RequestPipelineBuilder
                     // Start of mutating request
                     .first(RequestPipelineBuilder
-                                   .first(MakeRequestMutableStage::new)
-                                   .then(ApplyTransactionIdStage::new)
-                                   .then(ApplyUserAgentStage::new)
-                                   .then(MergeCustomHeadersStage::new)
-                                   .then(MergeCustomQueryParamsStage::new)
-                                   .then(MoveParametersToBodyStage::new)
-                                   .then(MakeRequestImmutableStage::new)
-                                   // End of mutating request
-                                   .then(RequestPipelineBuilder
-                                             .first(SigningStage::new)
-                                             .then(BeforeTransmissionExecutionInterceptorsStage::new)
-                                             .then(MakeHttpRequestStage::new)
-                                             .then(AfterTransmissionExecutionInterceptorsStage::new)
-                                             .then(Crc32ValidationStage::new)
-                                             .then(BeforeUnmarshallingExecutionInterceptorsStage::new)
-                                             .then(() -> new HandleResponseStage<>(
-                                                 getNonNullResponseHandler(responseHandler),
-                                                 getNonNullResponseHandler(errorResponseHandler)))
-                                             .wrappedWith(RetryableStage::new)::build)
-                                   .wrappedWith(StreamManagingStage::new)::build)
+                               .first(MakeRequestMutableStage::new)
+                               .then(ApplyTransactionIdStage::new)
+                               .then(ApplyUserAgentStage::new)
+                               .then(MergeCustomHeadersStage::new)
+                               .then(MergeCustomQueryParamsStage::new)
+                               .then(MoveParametersToBodyStage::new)
+                               .then(MakeRequestImmutableStage::new)
+                               // End of mutating request
+                               .then(RequestPipelineBuilder
+                                         .first(SigningStage::new)
+                                         .then(BeforeTransmissionExecutionInterceptorsStage::new)
+                                         .then(MakeHttpRequestStage::new)
+                                         .then(AfterTransmissionExecutionInterceptorsStage::new)
+                                         .then(Crc32ValidationStage::new)
+                                         .then(BeforeUnmarshallingExecutionInterceptorsStage::new)
+                                         .then(() -> new HandleResponseStage<>(
+                                             getNonNullResponseHandler(responseHandler),
+                                             getNonNullResponseHandler(errorResponseHandler)))
+                                         .wrappedWith(ApiCallAttemptTimeoutTrackingStage::new)
+                                         .wrappedWith(TimeoutExceptionHandlingStage::new)
+                                         .wrappedWith(RetryableStage::new)::build)
+                               .wrappedWith(StreamManagingStage::new)
+                               .wrappedWith(ApiCallTimeoutTrackingStage::new)::build)
                     .then(() -> new UnwrapResponseContainer<>())
                     .then(() -> new AfterExecutionInterceptorsStage<>())
                     .wrappedWith(ExecutionFailureExceptionReportingStage::new)
@@ -291,4 +297,4 @@ private RequestExecutionContext createRequestExecutionDependencies() {
 
     }
 
-}
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/RequestExecutionContext.java

@@ -41,6 +41,7 @@ public final class RequestExecutionContext {
     private final SdkRequest originalRequest;
     private final ExecutionContext executionContext;
     private TimeoutTracker apiCallTimeoutTracker;
+    private TimeoutTracker apiCallAttemptTimeoutTracker;
 
     private RequestExecutionContext(Builder builder) {
         this.requestProvider = builder.requestProvider;
@@ -109,6 +110,13 @@ public void apiCallTimeoutTracker(TimeoutTracker timeoutTracker) {
         this.apiCallTimeoutTracker = timeoutTracker;
     }
 
+    public TimeoutTracker apiCallAttemptTimeoutTracker() {
+        return apiCallAttemptTimeoutTracker;
+    }
+
+    public void apiCallAttemptTimeoutTracker(TimeoutTracker timeoutTracker) {
+        this.apiCallAttemptTimeoutTracker = timeoutTracker;
+    }
 
     /**
      * An SDK-internal implementation of {@link Builder}.

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/StreamManagingStage.java

@@ -53,6 +53,7 @@ public StreamManagingStage(RequestPipeline<SdkHttpFullRequest, Response<OutputT>
     public Response<OutputT> execute(SdkHttpFullRequest request, RequestExecutionContext context) throws Exception {
         Optional<InputStream> toBeClosed = createManagedStream(request);
         try {
+            InterruptMonitor.checkInterrupted();
             return wrapped.execute(request.toBuilder().content(nonCloseableInputStream(toBeClosed).orElse(null)).build(),
                                    context);
         } finally {

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

@@ -30,6 +30,7 @@ public class AfterTransmissionExecutionInterceptorsStage
     @Override
     public Pair<SdkHttpFullRequest, SdkHttpFullResponse> execute(Pair<SdkHttpFullRequest, SdkHttpFullResponse> input,
                                                                  RequestExecutionContext context) throws Exception {
+        InterruptMonitor.checkInterrupted();
         // Update interceptor context
         InterceptorContext interceptorContext =
                 context.executionContext().interceptorContext().copy(b -> b.httpResponse(input.right()));
@@ -43,8 +44,6 @@ public Pair<SdkHttpFullRequest, SdkHttpFullResponse> execute(Pair<SdkHttpFullReq
         // Store updated context
         context.executionContext().interceptorContext(interceptorContext);
 
-        // TODO: Why do we do this for sync, but not async? Are there other places it should be? Not having this fails the
-        // AbortedExceptionClientExecutionTimerIntegrationTest
         InterruptMonitor.checkInterrupted(interceptorContext.httpResponse());
 
         return Pair.of(input.left(), interceptorContext.httpResponse());