From f4ab9859f3ed77d1c26edeec91e0d6a770c77df8 Mon Sep 17 00:00:00 2001 From: Dongie Agnir Date: Tue, 21 Jul 2026 15:06:50 -0700 Subject: [PATCH 1/2] Update AsyncRetryableStage + tests - Update AsyncRetryableStage to use the new *Async methods from RetryStrategy - Update AdaptiveRetryRateLimitingTest to test both sync and async clients - Update other tests to wait on future to avoid timing issues --- .../pipeline/stages/AsyncRetryableStage.java | 70 ++++++------ .../stages/utils/RetryableStageHelper.java | 106 ++++++++++++++---- ...AsyncClientRetryStrategyExceptionTest.java | 4 + ...syncClientHandlerSignerValidationTest.java | 13 ++- .../handler/AsyncClientHandlerTest.java | 10 ++ .../stages/AsyncRetryableStageTest.java | 23 ++-- .../services/AsyncRequestCompressionTest.java | 6 +- .../BusinessMetricsUserAgentTest.java | 2 +- .../services/HttpChecksumInHeaderTest.java | 8 +- .../services/NoneAuthTypeRequestTest.java | 8 +- .../ProfileFileConfigurationTest.java | 3 +- .../AsyncOperationCancelTest.java | 15 +++ .../MoveQueryParamsToBodyTest.java | 6 +- .../AsyncOperationCancelTest.java | 75 ++++++++----- .../AsyncOperationCancelTest.java | 15 +++ .../retry/AdaptiveRetryRateLimitingTest.java | 68 ++++++++--- 16 files changed, 315 insertions(+), 117 deletions(-) diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStage.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStage.java index 15b81d30d75f..19ba5036ecc7 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStage.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStage.java @@ -36,7 +36,6 @@ import software.amazon.awssdk.http.SdkHttpFullRequest; import software.amazon.awssdk.http.SdkHttpResponse; import software.amazon.awssdk.utils.CompletableFutureUtils; -import software.amazon.awssdk.utils.Either; import software.amazon.awssdk.utils.Logger; /** @@ -90,14 +89,14 @@ public CompletableFuture> execute() { } public void attemptFirstExecute(CompletableFuture> future) { - Duration backoffDelay = retryableStageHelper.acquireInitialToken(); - if (backoffDelay.isZero()) { - attemptExecute(future); - } else { - retryableStageHelper.logBackingOff(backoffDelay); - long totalDelayMillis = backoffDelay.toMillis(); - scheduledExecutor.schedule(() -> attemptExecute(future), totalDelayMillis, MILLISECONDS); - } + retryableStageHelper.acquireInitialTokenAsync().whenComplete((r, t) -> { + if (t != null) { + future.completeExceptionally(t); + return; + } + scheduledExecutor.schedule(() -> attemptExecute(future), r.toMillis(), MILLISECONDS); + }); + } private void attemptExecute(CompletableFuture> future) { @@ -140,33 +139,38 @@ private void attemptExecute(CompletableFuture> future) { } public void maybeAttemptExecute(CompletableFuture> future) { - Either backoffDelay = retryableStageHelper.tryRefreshToken(suggestedDelay()); - - Optional acquireFailureDelay = backoffDelay.right(); - if (acquireFailureDelay.isPresent()) { - Duration delay = acquireFailureDelay.get(); - retryableStageHelper.logAcquireFailureBackingOff(delay); - SdkException disallowedException = retryableStageHelper.retryPolicyDisallowedRetryException(); - // Avoid needless scheduling if we won't wait - if (delay.isZero()) { - future.completeExceptionally(disallowedException); - } else { - scheduledExecutor.schedule(() -> future.completeExceptionally(disallowedException), - delay.toMillis(), MILLISECONDS); + retryableStageHelper.tryRefreshTokenAsync(suggestedDelay()).whenComplete((backoffDelay, t) -> { + if (t != null) { + future.completeExceptionally(t); + return; } - return; - } - // We failed the last attempt, but will retry. The response handler wants to know when that happens. - responseHandler.onError(retryableStageHelper.getLastException()); - // Reset the request provider to the original one before retries, in case it was modified downstream. - context.requestProvider(originalRequestBody); + Optional acquireFailureDelay = backoffDelay.right(); + if (acquireFailureDelay.isPresent()) { + Duration delay = acquireFailureDelay.get(); + retryableStageHelper.logAcquireFailureBackingOff(delay); + SdkException disallowedException = retryableStageHelper.retryPolicyDisallowedRetryException(); + // Avoid needless scheduling if we won't wait + if (delay.isZero()) { + future.completeExceptionally(disallowedException); + } else { + scheduledExecutor.schedule(() -> future.completeExceptionally(disallowedException), + delay.toMillis(), MILLISECONDS); + } + return; + } + // We failed the last attempt, but will retry. The response handler wants to know when that happens. + responseHandler.onError(retryableStageHelper.getLastException()); + + // Reset the request provider to the original one before retries, in case it was modified downstream. + context.requestProvider(originalRequestBody); - // get() is safe, Either requires left OR right to be present - Duration successDelay = backoffDelay.left().get(); - retryableStageHelper.logBackingOff(successDelay); - long totalDelayMillis = successDelay.toMillis(); - scheduledExecutor.schedule(() -> attemptExecute(future), totalDelayMillis, MILLISECONDS); + // get() is safe, Either requires left OR right to be present + Duration successDelay = backoffDelay.left().get(); + retryableStageHelper.logBackingOff(successDelay); + long totalDelayMillis = successDelay.toMillis(); + scheduledExecutor.schedule(() -> attemptExecute(future), totalDelayMillis, MILLISECONDS); + }); } private void maybeRetryExecute(CompletableFuture> future, Exception exception) { diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/RetryableStageHelper.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/RetryableStageHelper.java index 840df78367fd..39780334c526 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/RetryableStageHelper.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/RetryableStageHelper.java @@ -23,6 +23,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Optional; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import software.amazon.awssdk.annotations.SdkInternalApi; import software.amazon.awssdk.core.Response; @@ -60,6 +61,7 @@ public final class RetryableStageHelper { public static final String SDK_RETRY_INFO_HEADER = "amz-sdk-request"; public static final ExecutionAttribute LAST_BACKOFF_DELAY_DURATION = new ExecutionAttribute<>("LastBackoffDuration"); + private static final String RETRY_TOKEN_SCOPE = "GLOBAL"; private final SdkHttpFullRequest request; private final boolean isLongPollingOperation; @@ -72,6 +74,7 @@ public final class RetryableStageHelper { private SdkHttpResponse lastResponse; private SdkException lastException; + public RetryableStageHelper(SdkHttpFullRequest request, RequestExecutionContext context, HttpClientDependencies dependencies) { @@ -106,14 +109,31 @@ public void startingAttempt() { * value is {@link AdaptiveRetryStrategy}. */ public Duration acquireInitialToken() { - String scope = "GLOBAL"; - AcquireInitialTokenRequest acquireRequest = AcquireInitialTokenRequest.create(scope); - AcquireInitialTokenResponse acquireResponse = retryStrategy().acquireInitialToken(acquireRequest); - RetryToken retryToken = acquireResponse.token(); - Duration delay = acquireResponse.delay(); - context.executionAttributes().putAttribute(RETRY_TOKEN, retryToken); - context.executionAttributes().putAttribute(LAST_BACKOFF_DELAY_DURATION, delay); - return delay; + AcquireInitialTokenResponse result = doBlockAcquireInitialToken(); + context.executionAttributes().putAttribute(RETRY_TOKEN, result.token()); + context.executionAttributes().putAttribute(LAST_BACKOFF_DELAY_DURATION, result.delay()); + return result.delay(); + } + + private AcquireInitialTokenResponse doBlockAcquireInitialToken() { + AcquireInitialTokenRequest acquireRequest = AcquireInitialTokenRequest.create(RETRY_TOKEN_SCOPE); + + return retryStrategy().acquireInitialToken(acquireRequest); + } + + public CompletableFuture acquireInitialTokenAsync() { + AcquireInitialTokenRequest acquireRequest = AcquireInitialTokenRequest.create(RETRY_TOKEN_SCOPE); + + return retryStrategy().acquireInitialTokenAsync(acquireRequest).whenComplete( + (result, t) -> { + if (t != null) { + return; + } + + context.executionAttributes().putAttribute(RETRY_TOKEN, result.token()); + context.executionAttributes().putAttribute(LAST_BACKOFF_DELAY_DURATION, result.delay()); + }) + .thenApply(AcquireInitialTokenResponse::delay); } /** @@ -138,16 +158,12 @@ public void recordAttemptSucceeded() { * code should not retry. */ public Either tryRefreshToken(Duration suggestedDelay) { - RetryToken retryToken = context.executionAttributes().getAttribute(RETRY_TOKEN); - RefreshRetryTokenResponse refreshResponse; + RetryToken retryToken; + Duration attemptDelay; try { - RefreshRetryTokenRequest refreshRequest = RefreshRetryTokenRequest.builder() - .failure(this.lastException) - .token(retryToken) - .isLongPolling(isLongPollingOperation) - .suggestedDelay(suggestedDelay) - .build(); - refreshResponse = retryStrategy().refreshRetryToken(refreshRequest); + RefreshRetryTokenResponse tryRefreshResult = doBlockingRefreshRetryToken(suggestedDelay); + attemptDelay = tryRefreshResult.delay(); + retryToken = tryRefreshResult.token(); } catch (TokenAcquisitionFailedException e) { context.executionAttributes().putAttribute(RETRY_TOKEN, e.token()); Optional acquireFailureDelay = e.delay(); @@ -158,12 +174,64 @@ public Either tryRefreshToken(Duration suggestedDelay) { } return Either.right(Duration.ZERO); } - Duration acquireSuccessDelay = refreshResponse.delay(); - context.executionAttributes().putAttribute(RETRY_TOKEN, refreshResponse.token()); + Duration acquireSuccessDelay = attemptDelay; + context.executionAttributes().putAttribute(RETRY_TOKEN, retryToken); context.executionAttributes().putAttribute(LAST_BACKOFF_DELAY_DURATION, acquireSuccessDelay); return Either.left(acquireSuccessDelay); } + private RefreshRetryTokenResponse doBlockingRefreshRetryToken(Duration suggestedDelay) { + RetryToken retryToken = context.executionAttributes().getAttribute(RETRY_TOKEN); + + RefreshRetryTokenRequest refreshRequest = RefreshRetryTokenRequest.builder() + .failure(this.lastException) + .token(retryToken) + .isLongPolling(isLongPollingOperation) + .suggestedDelay(suggestedDelay) + .build(); + + return retryStrategy().refreshRetryToken(refreshRequest); + } + + public CompletableFuture> tryRefreshTokenAsync(Duration suggestedDelay) { + CompletableFuture> cf = new CompletableFuture<>(); + + RetryToken retryToken = context.executionAttributes().getAttribute(RETRY_TOKEN); + + RefreshRetryTokenRequest refreshRequest = RefreshRetryTokenRequest.builder() + .failure(this.lastException) + .token(retryToken) + .isLongPolling(isLongPollingOperation) + .suggestedDelay(suggestedDelay) + .build(); + + retryStrategy().refreshRetryTokenAsync(refreshRequest).whenComplete((r, t) -> { + if (t != null) { + if (t instanceof TokenAcquisitionFailedException) { + TokenAcquisitionFailedException e = (TokenAcquisitionFailedException) t; + context.executionAttributes().putAttribute(RETRY_TOKEN, e.token()); + Optional acquireFailureDelay = e.delay(); + if (acquireFailureDelay.isPresent()) { + Duration acquireDelay = acquireFailureDelay.get(); + context.executionAttributes().putAttribute(LAST_BACKOFF_DELAY_DURATION, acquireDelay); + cf.complete(Either.right(acquireDelay)); + } + cf.complete(Either.right(Duration.ZERO)); + } else { + cf.completeExceptionally(t); + } + return; + } + + Duration acquireSuccessDelay = r.delay(); + context.executionAttributes().putAttribute(RETRY_TOKEN, r.token()); + context.executionAttributes().putAttribute(LAST_BACKOFF_DELAY_DURATION, acquireSuccessDelay); + cf.complete(Either.left(acquireSuccessDelay)); + }); + + return cf; + } + /** * Return the exception that should be thrown, because the retry strategy did not allow the request to be retried. */ diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/AsyncClientRetryStrategyExceptionTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/AsyncClientRetryStrategyExceptionTest.java index 983be14c8451..011df8916320 100644 --- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/AsyncClientRetryStrategyExceptionTest.java +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/AsyncClientRetryStrategyExceptionTest.java @@ -61,6 +61,7 @@ public class AsyncClientRetryStrategyExceptionTest { public void exceptionInInitialTokenReportedInFuture() { Exception exception = new RuntimeException(MESSAGE); when(retryStrategy.acquireInitialToken(any())).thenThrow(exception); + when(retryStrategy.acquireInitialTokenAsync(any())).thenCallRealMethod(); CompletableFuture responseFuture = makeRequest(); @@ -73,8 +74,11 @@ public void exceptionInRefreshTokenReportedInFuture() { AcquireInitialTokenResponse.create(new RetryToken() { }, Duration.ZERO) ); + when(retryStrategy.acquireInitialTokenAsync(any())).thenCallRealMethod(); + Exception exception = new RuntimeException(MESSAGE); when(retryStrategy.refreshRetryToken(any())).thenThrow(exception); + when(retryStrategy.refreshRetryTokenAsync(any())).thenCallRealMethod(); CompletableFuture responseFuture = makeRequest(); diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/handler/AsyncClientHandlerSignerValidationTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/handler/AsyncClientHandlerSignerValidationTest.java index b0c19a521895..4bceeb8a2d0b 100644 --- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/handler/AsyncClientHandlerSignerValidationTest.java +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/handler/AsyncClientHandlerSignerValidationTest.java @@ -18,6 +18,8 @@ import static org.assertj.core.api.Assertions.assertThatNoException; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import java.util.Arrays; @@ -26,6 +28,8 @@ import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -179,7 +183,6 @@ public void execute_requestHasHttpsEndpoint_usesBearerAuth_succeeds() throws Exc SdkAsyncClientHandler sdkAsyncClientHandler = new SdkAsyncClientHandler(testClientConfiguration()); CompletableFuture execute = sdkAsyncClientHandler.execute(executionParams()); - SdkAsyncHttpResponseHandler capturedHandler = executeRequestCaptor.getValue().responseHandler(); Map> headers = new HashMap<>(); headers.put("foo", Arrays.asList("bar")); @@ -227,10 +230,18 @@ private ClientExecutionParams executionParams() { } private SdkClientConfiguration testClientConfiguration() { + ScheduledExecutorService exec = mock(ScheduledExecutorService.class); + when(exec.schedule(any(Runnable.class), anyLong(), any(TimeUnit.class))).thenAnswer(i -> { + Runnable r = i.getArgument(0); + r.run(); + return null; + }); + return HttpTestUtils.testClientConfiguration() .toBuilder() .option(SdkClientOption.ASYNC_HTTP_CLIENT, httpClient) .option(SdkAdvancedClientOption.SIGNER, signer) + .option(SdkClientOption.SCHEDULED_EXECUTOR_SERVICE, exec) .build(); } } diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/handler/AsyncClientHandlerTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/handler/AsyncClientHandlerTest.java index 6482ab936fc2..fef4caf7f256 100644 --- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/handler/AsyncClientHandlerTest.java +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/handler/AsyncClientHandlerTest.java @@ -18,6 +18,8 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; @@ -27,6 +29,7 @@ import java.util.Map; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import org.junit.Before; import org.junit.Test; @@ -143,10 +146,17 @@ private ClientExecutionParams clientExecutionParams() { } public SdkClientConfiguration clientConfiguration() { + ScheduledExecutorService exec = mock(ScheduledExecutorService.class); + when(exec.schedule(any(Runnable.class), anyLong(), any(TimeUnit.class))).thenAnswer(i -> { + Runnable r = i.getArgument(0); + r.run(); + return null; + }); return HttpTestUtils.testClientConfiguration().toBuilder() .option(SdkClientOption.ASYNC_HTTP_CLIENT, httpClient) .option(SdkClientOption.RETRY_POLICY, RetryPolicy.none()) .option(SdkClientOption.RETRY_STRATEGY, DefaultRetryStrategy.doNotRetry()) + .option(SdkClientOption.SCHEDULED_EXECUTOR_SERVICE, exec) .build(); } } diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStageTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStageTest.java index 827e1a7d1cef..f4b3d830cd78 100644 --- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStageTest.java +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStageTest.java @@ -60,6 +60,7 @@ import software.amazon.awssdk.retries.api.RetryStrategy; import software.amazon.awssdk.retries.api.RetryToken; import software.amazon.awssdk.retries.api.TokenAcquisitionFailedException; +import software.amazon.awssdk.utils.CompletableFutureUtils; public class AsyncRetryableStageTest extends BaseRetryableStageTest { private RetryStrategy mockRetryStrategy; @@ -89,7 +90,8 @@ void methodSetup() { when(mockAcquireInitialTokenResponse.token()).thenReturn(mockRetryToken); when(mockAcquireInitialTokenResponse.delay()).thenReturn(Duration.ZERO); - when(mockRetryStrategy.acquireInitialToken(any())).thenReturn(mockAcquireInitialTokenResponse); + when(mockRetryStrategy.acquireInitialTokenAsync(any())) + .thenReturn(CompletableFuture.completedFuture(mockAcquireInitialTokenResponse)); mockDelegatePipeline = mock(RequestPipeline.class); } @@ -141,17 +143,22 @@ void execute_acquireDelay_behavesCorrectly(AcquireDelayTestCase testCase) throws when(mockDelegatePipeline.execute(any(), any())).thenReturn(CompletableFuture.completedFuture(response)); if (testCase.isFailure()) { - when(mockRetryStrategy.refreshRetryToken(any())).thenThrow( - new TokenAcquisitionFailedException("Acquire failed", mockRetryToken, null, testCase.failureDelay()) + when(mockRetryStrategy.refreshRetryTokenAsync(any())).thenReturn( + CompletableFutureUtils.failedFuture(new TokenAcquisitionFailedException("Acquire failed", mockRetryToken, null, + testCase.failureDelay())) ); } else { // only retry once, otherwise we'll get into an infinite loop AtomicBoolean first = new AtomicBoolean(); - when(mockRetryStrategy.refreshRetryToken(any())).thenAnswer(i -> { + when(mockRetryStrategy.refreshRetryTokenAsync(any())).thenAnswer(i -> { if (first.compareAndSet(false, true)) { - return RefreshRetryTokenResponse.create(mockRetryToken, testCase.successDelay()); + return CompletableFuture.completedFuture(RefreshRetryTokenResponse.create(mockRetryToken, + testCase.successDelay())); } - throw new TokenAcquisitionFailedException("Acquire failed", mockRetryToken, null, Duration.ZERO); + return CompletableFutureUtils.failedFuture(new TokenAcquisitionFailedException("Acquire failed", + mockRetryToken, + null, + Duration.ZERO)); }); } @@ -229,7 +236,7 @@ void execute_retryableException_treatsRetryAfterCorrectly(RetryAfterTestCase tes ArgumentCaptor refreshRequestCaptor = ArgumentCaptor.forClass(RefreshRetryTokenRequest.class); - verify(mockRetryStrategy).refreshRetryToken(refreshRequestCaptor.capture()); + verify(mockRetryStrategy).refreshRetryTokenAsync(refreshRequestCaptor.capture()); RefreshRetryTokenRequest refreshRequest = refreshRequestCaptor.getValue(); @@ -292,7 +299,7 @@ void execute_delegateThrows_noHttpResponse_uses0SuggestedDelay(boolean newRetrie ArgumentCaptor refreshRequestCaptor = ArgumentCaptor.forClass(RefreshRetryTokenRequest.class); - verify(mockRetryStrategy).refreshRetryToken(refreshRequestCaptor.capture()); + verify(mockRetryStrategy).refreshRetryTokenAsync(refreshRequestCaptor.capture()); RefreshRetryTokenRequest refreshRequest = refreshRequestCaptor.getValue(); diff --git a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/AsyncRequestCompressionTest.java b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/AsyncRequestCompressionTest.java index df5b327a5442..b6d0ec381a7d 100644 --- a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/AsyncRequestCompressionTest.java +++ b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/AsyncRequestCompressionTest.java @@ -83,7 +83,7 @@ public void asyncNonStreamingOperation_compressionEnabledThresholdOverridden_com c -> c.minimumCompressionThresholdInBytes(1))) .build(); - asyncClient.putOperationWithRequestCompression(request); + asyncClient.putOperationWithRequestCompression(request).join(); SdkHttpFullRequest loggedRequest = (SdkHttpFullRequest) mockAsyncHttpClient.getLastRequest(); InputStream loggedStream = loggedRequest.contentStreamProvider().get().newStream(); @@ -104,7 +104,7 @@ public void asyncNonStreamingOperation_payloadSizeLessThanCompressionThreshold_d .body(SdkBytes.fromUtf8String(UNCOMPRESSED_BODY)) .build(); - asyncClient.putOperationWithRequestCompression(request); + asyncClient.putOperationWithRequestCompression(request).join(); SdkHttpFullRequest loggedRequest = (SdkHttpFullRequest) mockAsyncHttpClient.getLastRequest(); InputStream loggedStream = loggedRequest.contentStreamProvider().get().newStream(); @@ -148,7 +148,7 @@ public void asyncNonStreamingOperation_compressionEnabledThresholdOverriddenWith c -> c.minimumCompressionThresholdInBytes(1))) .build(); - asyncClient.putOperationWithRequestCompression(request); + asyncClient.putOperationWithRequestCompression(request).join(); SdkHttpFullRequest loggedRequest = (SdkHttpFullRequest) mockAsyncHttpClient.getLastRequest(); InputStream loggedStream = loggedRequest.contentStreamProvider().get().newStream(); diff --git a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/BusinessMetricsUserAgentTest.java b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/BusinessMetricsUserAgentTest.java index ab9394e70ac0..f95312956612 100644 --- a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/BusinessMetricsUserAgentTest.java +++ b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/BusinessMetricsUserAgentTest.java @@ -246,7 +246,7 @@ void validate_serviceUserAgent_format() { .overrideConfiguration(c -> c.addExecutionInterceptor(interceptor)) .build(); - client.headOperation(); + assertThatThrownBy(client.headOperation()::join); String userAgent = assertAndGetUserAgentString(); assertThat(userAgent).contains("AmazonProtocolRestJson#" + ServiceVersionInfo.VERSION); diff --git a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/HttpChecksumInHeaderTest.java b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/HttpChecksumInHeaderTest.java index f24f7f35d1c6..1424545509bc 100644 --- a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/HttpChecksumInHeaderTest.java +++ b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/HttpChecksumInHeaderTest.java @@ -105,9 +105,9 @@ public void sync_json_nonStreaming_unsignedPayload_with_Sha1_in_header() { } @Test - public void aync_json_nonStreaming_unsignedPayload_with_Sha1_in_header() { + public void async_json_nonStreaming_unsignedPayload_with_Sha1_in_header() { jsonAsyncClient.operationWithChecksumNonStreaming( - r -> r.checksumAlgorithm(ChecksumAlgorithm.SHA1).stringMember("Hello world").build()); + r -> r.checksumAlgorithm(ChecksumAlgorithm.SHA1).stringMember("Hello world").build()).join(); assertThat(getAsyncRequest().firstMatchingHeader("Content-MD5")).isNotPresent(); //Note that content will be of form "{"stringMember":"Hello world"}" assertThat(getAsyncRequest().firstMatchingHeader("x-amz-checksum-sha1")).hasValue("M68rRwFal7o7B3KEMt3m0w39TaA="); @@ -132,7 +132,7 @@ public void sync_xml_nonStreaming_unsignedEmptyPayload_with_Sha1_in_header() { } @Test - public void aync_xml_nonStreaming_unsignedPayload_with_Sha1_in_header() { + public void async_xml_nonStreaming_unsignedPayload_with_Sha1_in_header() { xmlAsyncClient.operationWithChecksumNonStreaming(r -> r.stringMember("Hello world") .checksumAlgorithm(software.amazon.awssdk.services.protocolrestxml.model.ChecksumAlgorithm.SHA1).build()).join(); assertThat(getAsyncRequest().firstMatchingHeader("Content-MD5")).isNotPresent(); @@ -141,7 +141,7 @@ public void aync_xml_nonStreaming_unsignedPayload_with_Sha1_in_header() { } @Test - public void aync_xml_nonStreaming_unsignedEmptyPayload_with_Sha1_in_header() { + public void async_xml_nonStreaming_unsignedEmptyPayload_with_Sha1_in_header() { xmlAsyncClient.operationWithChecksumNonStreaming(r -> r.checksumAlgorithm(software.amazon.awssdk.services.protocolrestxml.model.ChecksumAlgorithm.SHA1).build()).join(); assertThat(getAsyncRequest().firstMatchingHeader("Content-MD5")).isNotPresent(); diff --git a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/NoneAuthTypeRequestTest.java b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/NoneAuthTypeRequestTest.java index 20ceeee91815..d193c7087f11 100644 --- a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/NoneAuthTypeRequestTest.java +++ b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/NoneAuthTypeRequestTest.java @@ -113,14 +113,14 @@ public void sync_json_authorization_is_present_for_defaultAuth() { @Test public void async_json_authorization_is_absent_for_noneAuthType() { - jsonAsyncClient.operationWithNoneAuthType(o -> o.booleanMember(true)); + jsonAsyncClient.operationWithNoneAuthType(o -> o.booleanMember(true)).join(); assertThat(getAsyncRequest().firstMatchingHeader("Authorization")).isNotPresent(); verify(credentialsProvider, times(0)).resolveIdentity(any(ResolveIdentityRequest.class)); } @Test public void async_json_authorization_is_present_for_defaultAuth() { - jsonAsyncClient.jsonValuesOperation(); + jsonAsyncClient.jsonValuesOperation().join(); assertThat(getAsyncRequest().firstMatchingHeader("Authorization")).isPresent(); verify(credentialsProvider, times(1)).resolveIdentity(any(ResolveIdentityRequest.class)); } @@ -141,14 +141,14 @@ public void sync_xml_authorization_is_present_for_defaultAuth() { @Test public void async_xml_authorization_is_absent_for_noneAuthType() { - xmlAsyncClient.operationWithNoneAuthType(o -> o.booleanMember(true)); + xmlAsyncClient.operationWithNoneAuthType(o -> o.booleanMember(true)).join(); assertThat(getAsyncRequest().firstMatchingHeader("Authorization")).isNotPresent(); verify(credentialsProvider, times(0)).resolveIdentity(any(ResolveIdentityRequest.class)); } @Test public void async_xml_authorization_is_present_for_defaultAuth() { - xmlAsyncClient.jsonValuesOperation(json -> json.jsonValueMember("one")); + xmlAsyncClient.jsonValuesOperation(json -> json.jsonValueMember("one")).join(); assertThat(getAsyncRequest().firstMatchingHeader("Authorization")).isPresent(); verify(credentialsProvider, times(1)).resolveIdentity(any(ResolveIdentityRequest.class)); } diff --git a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/ProfileFileConfigurationTest.java b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/ProfileFileConfigurationTest.java index d8a11a0b76b2..21df46df84ff 100644 --- a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/ProfileFileConfigurationTest.java +++ b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/ProfileFileConfigurationTest.java @@ -41,6 +41,7 @@ import software.amazon.awssdk.http.auth.aws.scheme.AwsV4AuthScheme; import software.amazon.awssdk.http.auth.aws.signer.AwsV4HttpSigner; import software.amazon.awssdk.http.auth.spi.signer.AsyncSignRequest; +import software.amazon.awssdk.http.auth.spi.signer.AsyncSignedRequest; import software.amazon.awssdk.http.auth.spi.signer.SignRequest; import software.amazon.awssdk.http.auth.spi.signer.SignedRequest; import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity; @@ -197,7 +198,7 @@ public void streamingOperation_asyncHttpSigner_profileIsHonoredForCredentialsAnd env.remove(SdkSystemSetting.AWS_ACCESS_KEY_ID); env.remove(SdkSystemSetting.AWS_SECRET_ACCESS_KEY); - Mockito.when(signer.signAsync(any(AsyncSignRequest.class))).thenReturn(CompletableFuture.completedFuture(any(AsyncSignRequest.class))); + Mockito.when(signer.signAsync(any(AsyncSignRequest.class))).thenReturn(CompletableFuture.completedFuture(mock(AsyncSignedRequest.class))); ProtocolRestJsonAsyncClient asyncClient = asyncClientWithHttpSignerOverride(); diff --git a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolquery/AsyncOperationCancelTest.java b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolquery/AsyncOperationCancelTest.java index e243a489ddbb..4498222bd438 100644 --- a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolquery/AsyncOperationCancelTest.java +++ b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolquery/AsyncOperationCancelTest.java @@ -15,6 +15,8 @@ package software.amazon.awssdk.services.protocolquery; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -35,6 +37,8 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; /** @@ -49,13 +53,24 @@ public class AsyncOperationCancelTest { private CompletableFuture executeFuture; + private ScheduledExecutorService scheduledExec; + @Before public void setUp() { + scheduledExec = mock(ScheduledExecutorService.class); + + when(scheduledExec.schedule(any(Runnable.class), anyLong(), any(TimeUnit.class))).thenAnswer(i -> { + Runnable r = i.getArgument(0); + r.run(); + return null; + }); + client = ProtocolQueryAsyncClient.builder() .region(Region.US_EAST_1) .credentialsProvider(StaticCredentialsProvider.create( AwsBasicCredentials.create("foo", "bar"))) .httpClient(mockHttpClient) + .overrideConfiguration(o -> o.scheduledExecutorService(scheduledExec)) .build(); executeFuture = new CompletableFuture(); diff --git a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolquery/MoveQueryParamsToBodyTest.java b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolquery/MoveQueryParamsToBodyTest.java index 559c36a8019b..9532d4bd83c6 100644 --- a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolquery/MoveQueryParamsToBodyTest.java +++ b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolquery/MoveQueryParamsToBodyTest.java @@ -129,7 +129,7 @@ public void customInterceptor_asyncClient_additionalQueryParamsAdded_paramsAlsoM .build(); ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(AsyncExecuteRequest.class); - asyncClient.membersInQueryParams(r -> r.stringQueryParam("hello")); + assertThatThrownBy(asyncClient.membersInQueryParams(r -> r.stringQueryParam("hello"))::join); verify(asyncMockHttpClient, atLeast(1)).execute(requestCaptor.capture()); verifyParametersMovedToBody_asyncClient(requestCaptor); @@ -164,7 +164,7 @@ public void requestOverrideConfiguration_asyncClient_additionalQueryParamsAdded_ ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(AsyncExecuteRequest.class); - asyncClient.membersInQueryParams(r -> r.stringQueryParam("hello").overrideConfiguration(createOverrideConfigWithQueryParams())); + assertThatThrownBy(asyncClient.membersInQueryParams(r -> r.stringQueryParam("hello").overrideConfiguration(createOverrideConfigWithQueryParams()))::join); verify(asyncMockHttpClient, atLeast(1)).execute(requestCaptor.capture()); verifyParametersMovedToBody_asyncClient(requestCaptor); } @@ -199,7 +199,7 @@ public void asyncClient_noQueryParamsAdded_onlyContainsOriginalContent() throws .build(); ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(AsyncExecuteRequest.class); - asyncClient.allTypes(r -> r.stringMember("hello")); + assertThatThrownBy(asyncClient.allTypes(r -> r.stringMember("hello"))::join); verify(asyncMockHttpClient, atLeast(1)).execute(requestCaptor.capture()); diff --git a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolrestjson/AsyncOperationCancelTest.java b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolrestjson/AsyncOperationCancelTest.java index 932f4a9032ac..0fceef6d1a11 100644 --- a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolrestjson/AsyncOperationCancelTest.java +++ b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolrestjson/AsyncOperationCancelTest.java @@ -17,10 +17,14 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import io.reactivex.Flowable; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -43,7 +47,8 @@ import software.amazon.awssdk.services.protocolrestjson.model.StreamingOutputOperationResponse; /** - * Test to ensure that cancelling the future returned for an async operation will cancel the future returned by the async HTTP client. + * Test to ensure that cancelling the future returned for an async operation will cancel the future returned by the async HTTP + * client. */ @RunWith(MockitoJUnitRunner.class) public class AsyncOperationCancelTest { @@ -54,30 +59,45 @@ public class AsyncOperationCancelTest { private CompletableFuture executeFuture; + private ScheduledExecutorService scheduledExec; + @Before public void setUp() { + scheduledExec = mock(ScheduledExecutorService.class); + + when(scheduledExec.schedule(any(Runnable.class), anyLong(), any(TimeUnit.class))).thenAnswer(i -> { + Runnable r = i.getArgument(0); + r.run(); + return null; + }); + client = ProtocolRestJsonAsyncClient.builder() - .region(Region.US_EAST_1) - .credentialsProvider(StaticCredentialsProvider.create( - AwsBasicCredentials.create("foo", "bar"))) - .httpClient(mockHttpClient) - .build(); + .region(Region.US_EAST_1) + .credentialsProvider(StaticCredentialsProvider.create( + AwsBasicCredentials.create("foo", "bar"))) + .httpClient(mockHttpClient) + .overrideConfiguration(o -> o.scheduledExecutorService(scheduledExec)) + .build(); executeFuture = new CompletableFuture<>(); + when(mockHttpClient.execute(any())).thenReturn(executeFuture); } @Test - public void testNonStreamingOperation() { - CompletableFuture responseFuture = client.allTypes(r -> {}); + public void testNonStreamingOperation() throws InterruptedException { + CompletableFuture responseFuture = client.allTypes(r -> { + }); responseFuture.cancel(true); + // Thread.sleep(1000); assertThat(executeFuture.isCompletedExceptionally()).isTrue(); assertThat(executeFuture.isCancelled()).isTrue(); } @Test public void testStreamingOperation() { - CompletableFuture responseFuture = client.streamingInputOperation(r -> {}, AsyncRequestBody.empty()); + CompletableFuture responseFuture = client.streamingInputOperation(r -> { + }, AsyncRequestBody.empty()); responseFuture.cancel(true); assertThat(executeFuture.isCompletedExceptionally()).isTrue(); assertThat(executeFuture.isCancelled()).isTrue(); @@ -95,25 +115,26 @@ public void testStreamingOutputOperation() { @Test public void testEventStreamingOperation() throws InterruptedException { CompletableFuture responseFuture = - client.eventStreamOperation(r -> {}, + client.eventStreamOperation(r -> { + }, Flowable.just(InputEventStream.inputEventBuilder().build()), - new EventStreamOperationResponseHandler() { - @Override - public void responseReceived(EventStreamOperationResponse response) { - } - - @Override - public void onEventStream(SdkPublisher publisher) { - } - - @Override - public void exceptionOccurred(Throwable throwable) { - } - - @Override - public void complete() { - } - }); + new EventStreamOperationResponseHandler() { + @Override + public void responseReceived(EventStreamOperationResponse response) { + } + + @Override + public void onEventStream(SdkPublisher publisher) { + } + + @Override + public void exceptionOccurred(Throwable throwable) { + } + + @Override + public void complete() { + } + }); responseFuture.cancel(true); assertThat(executeFuture.isCompletedExceptionally()).isTrue(); assertThat(executeFuture.isCancelled()).isTrue(); diff --git a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolrestxml/AsyncOperationCancelTest.java b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolrestxml/AsyncOperationCancelTest.java index 51deef21c7e5..5de704f296f7 100644 --- a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolrestxml/AsyncOperationCancelTest.java +++ b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/protocolrestxml/AsyncOperationCancelTest.java @@ -15,6 +15,8 @@ package software.amazon.awssdk.services.protocolrestxml; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -39,6 +41,8 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; /** @@ -53,13 +57,24 @@ public class AsyncOperationCancelTest { private CompletableFuture executeFuture; + private ScheduledExecutorService scheduledExec; + @Before public void setUp() { + scheduledExec = mock(ScheduledExecutorService.class); + + when(scheduledExec.schedule(any(Runnable.class), anyLong(), any(TimeUnit.class))).thenAnswer(i -> { + Runnable r = i.getArgument(0); + r.run(); + return null; + }); + client = ProtocolRestXmlAsyncClient.builder() .region(Region.US_EAST_1) .credentialsProvider(StaticCredentialsProvider.create( AwsBasicCredentials.create("foo", "bar"))) .httpClient(mockHttpClient) + .overrideConfiguration(o -> o.scheduledExecutorService(scheduledExec)) .build(); executeFuture = new CompletableFuture(); diff --git a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/retry/AdaptiveRetryRateLimitingTest.java b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/retry/AdaptiveRetryRateLimitingTest.java index 39ded28482b9..24bc160e9f9f 100644 --- a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/retry/AdaptiveRetryRateLimitingTest.java +++ b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/retry/AdaptiveRetryRateLimitingTest.java @@ -32,6 +32,7 @@ import java.time.Duration; import java.time.Instant; import java.util.List; +import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -47,11 +48,16 @@ import org.junit.jupiter.api.extension.RegisterExtension; import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.awscore.client.builder.AwsClientBuilder; +import software.amazon.awssdk.awscore.client.builder.AwsDefaultClientBuilder; import software.amazon.awssdk.awscore.retry.AwsRetryStrategy; import software.amazon.awssdk.core.SdkSystemSetting; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.retries.AdaptiveRetryStrategy; +import software.amazon.awssdk.services.json.JsonAsyncClient; +import software.amazon.awssdk.services.json.JsonAsyncClientBuilder; import software.amazon.awssdk.services.json.JsonClient; +import software.amazon.awssdk.services.json.JsonClientBuilder; /** * End-to-end testing for testing the correctness of the rate limiting implementation of the adaptive retry strategy. @@ -93,24 +99,61 @@ static void teardown() { } } + @Test + void staticServerThrottling_sync() throws Exception { + JsonClientBuilder builder = JsonClient.builder(); + configure(builder); + JsonClient client = builder.build(); + + Callable c = () -> { + client.allType(r -> {}); + return null; + }; + + try { + testStaticServerThrottling(c); + } finally { + client.close(); + } + } + + @Test + void staticServerThrottling_async() throws Exception { + JsonAsyncClientBuilder builder = JsonAsyncClient.builder(); + configure(builder); + JsonAsyncClient client = builder.build(); + + Callable c = () -> { + client.allType(r -> {}).join(); + return null; + }; + + try { + testStaticServerThrottling(c); + } finally { + client.close(); + } + } + + private void configure(AwsClientBuilder builder) { + AdaptiveRetryStrategy adaptiveRetryStrategy = AwsRetryStrategy.adaptiveRetryStrategy(true); + + builder.endpointOverride(URI.create(wm.baseUrl())) + .region(Region.US_WEST_2) + .credentialsProvider(StaticCredentialsProvider.create( + AwsBasicCredentials.create("akid", "secret"))) + .overrideConfiguration(o -> o.retryStrategy(adaptiveRetryStrategy)) + .build(); + } + /** * The server has a static max TPS of 20 before responding with throttling exceptions. The client sending rate should match * this within an acceptable margin. */ - @Test - void staticServerThrottling() throws Exception { + void testStaticServerThrottling(Callable operationInvoker) throws Exception { wm.stubFor(any(anyUrl()).willReturn(aResponse().withTransformers("throttling"))); - AdaptiveRetryStrategy adaptiveRetryStrategy = AwsRetryStrategy.adaptiveRetryStrategy(true); - - JsonClient client = JsonClient.builder() - .endpointOverride(URI.create(wm.baseUrl())) - .region(Region.US_WEST_2) - .credentialsProvider(StaticCredentialsProvider.create( - AwsBasicCredentials.create("akid", "secret"))) - .overrideConfiguration(o -> o.retryStrategy(adaptiveRetryStrategy)) - .build(); Instant startT = Instant.now(); Instant endAt = startT.plus(testDuration); @@ -122,8 +165,7 @@ void staticServerThrottling() throws Exception { .mapToObj(i -> pool.submit(() -> { while (!stop.get() && Instant.now().isBefore(endAt)) { try { - client.allType(r -> { - }); + operationInvoker.call(); } catch (Exception e) { // We measure wire-level TPS, not call success. // Throttling exceptions surface here once retries are exhausted. From 70480b393a468de79ed37c9352a7e9b73230a6c0 Mon Sep 17 00:00:00 2001 From: Dongie Agnir Date: Fri, 24 Jul 2026 13:13:12 -0700 Subject: [PATCH 2/2] Handle race condition in CRT tx pause If S3MetaRequestPauseObservable#pause is invoked before S3MetaRequestPauseObservable#subscribe is executed, an NPE can occur. This can happen if the transfer hasn't actually begin before the pause() was invoked. Fix this by returning `null` in this case, which `CrtFileUpload` treats as transfer not yet started. --- .../s3/internal/crt/S3MetaRequestPauseObservable.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3MetaRequestPauseObservable.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3MetaRequestPauseObservable.java index 5ddb41219d39..9307b38b79eb 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3MetaRequestPauseObservable.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3MetaRequestPauseObservable.java @@ -43,7 +43,11 @@ public void subscribe(S3MetaRequestWrapper request) { * Pause the request */ public ResumeToken pause() { - return pause.apply(request); + S3MetaRequestWrapper r = this.request; + if (r != null) { + return pause.apply(r); + } + return null; } }