Skip to content

Commit 5feba0f

Browse files
sugmanuesugmanue
authored
Remove legacy retry stages and supporting classes (#5784)
1 parent fc289f0 commit 5feba0f

File tree

10 files changed

+237
-1452
lines changed

10 files changed

+237
-1452
lines changed

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@
3737
import software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncApiCallTimeoutTrackingStage;
3838
import software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncBeforeTransmissionExecutionInterceptorsStage;
3939
import software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncExecutionFailureExceptionReportingStage;
40-
import software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncRetryableStage2;
40+
import software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncRetryableStage;
4141
import software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncSigningStage;
4242
import software.amazon.awssdk.core.internal.http.pipeline.stages.CompressRequestStage;
4343
import software.amazon.awssdk.core.internal.http.pipeline.stages.HttpChecksumStage;
@@ -205,8 +205,8 @@ public <OutputT> CompletableFuture<OutputT> execute(
205205
.then(AsyncBeforeTransmissionExecutionInterceptorsStage::new)
206206
.then(d -> new MakeAsyncHttpRequestStage<>(responseHandler, d))
207207
.wrappedWith(AsyncApiCallAttemptMetricCollectionStage::new)
208-
.wrappedWith((deps, wrapped) -> new AsyncRetryableStage2<>(responseHandler, deps,
209-
wrapped))
208+
.wrappedWith((deps, wrapped) -> new AsyncRetryableStage<>(responseHandler, deps,
209+
wrapped))
210210
.then(async(() -> new UnwrapResponseContainer<>()))
211211
.then(async(() -> new AfterExecutionInterceptorsStage<>()))
212212
.wrappedWith(AsyncExecutionFailureExceptionReportingStage::new)

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@
4646
import software.amazon.awssdk.core.internal.http.pipeline.stages.MergeCustomHeadersStage;
4747
import software.amazon.awssdk.core.internal.http.pipeline.stages.MergeCustomQueryParamsStage;
4848
import software.amazon.awssdk.core.internal.http.pipeline.stages.QueryParametersToBodyStage;
49-
import software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage2;
49+
import software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage;
5050
import software.amazon.awssdk.core.internal.http.pipeline.stages.SigningStage;
5151
import software.amazon.awssdk.core.internal.http.pipeline.stages.TimeoutExceptionHandlingStage;
5252
import software.amazon.awssdk.core.internal.http.pipeline.stages.UnwrapResponseContainer;
@@ -199,7 +199,7 @@ public <OutputT> OutputT execute(HttpResponseHandler<Response<OutputT>> response
199199
.wrappedWith(ApiCallAttemptTimeoutTrackingStage::new)
200200
.wrappedWith(TimeoutExceptionHandlingStage::new)
201201
.wrappedWith((deps, wrapped) -> new ApiCallAttemptMetricCollectionStage<>(wrapped))
202-
.wrappedWith(RetryableStage2::new)::build)
202+
.wrappedWith(RetryableStage::new)::build)
203203
.wrappedWith(StreamManagingStage::new)
204204
.wrappedWith(ApiCallTimeoutTrackingStage::new)::build)
205205
.wrappedWith((deps, wrapped) -> new ApiCallMetricCollectionStage<>(wrapped))

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

Lines changed: 31 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -19,22 +19,19 @@
1919

2020
import java.io.IOException;
2121
import java.time.Duration;
22-
import java.util.OptionalDouble;
22+
import java.util.Optional;
2323
import java.util.concurrent.CompletableFuture;
2424
import java.util.concurrent.ScheduledExecutorService;
2525
import software.amazon.awssdk.annotations.SdkInternalApi;
26-
import software.amazon.awssdk.annotations.SdkTestInternalApi;
2726
import software.amazon.awssdk.core.Response;
2827
import software.amazon.awssdk.core.async.AsyncRequestBody;
2928
import software.amazon.awssdk.core.client.config.SdkClientOption;
30-
import software.amazon.awssdk.core.exception.SdkClientException;
3129
import software.amazon.awssdk.core.exception.SdkException;
3230
import software.amazon.awssdk.core.internal.http.HttpClientDependencies;
3331
import software.amazon.awssdk.core.internal.http.RequestExecutionContext;
3432
import software.amazon.awssdk.core.internal.http.TransformingAsyncResponseHandler;
3533
import software.amazon.awssdk.core.internal.http.pipeline.RequestPipeline;
3634
import software.amazon.awssdk.core.internal.http.pipeline.stages.utils.RetryableStageHelper;
37-
import software.amazon.awssdk.core.internal.retry.RateLimitingTokenBucket;
3835
import software.amazon.awssdk.http.SdkHttpFullRequest;
3936
import software.amazon.awssdk.utils.CompletableFutureUtils;
4037

@@ -49,97 +46,54 @@ public final class AsyncRetryableStage<OutputT> implements RequestPipeline<SdkHt
4946
private final RequestPipeline<SdkHttpFullRequest, CompletableFuture<Response<OutputT>>> requestPipeline;
5047
private final ScheduledExecutorService scheduledExecutor;
5148
private final HttpClientDependencies dependencies;
52-
private final RateLimitingTokenBucket rateLimitingTokenBucket;
5349

5450
public AsyncRetryableStage(TransformingAsyncResponseHandler<Response<OutputT>> responseHandler,
5551
HttpClientDependencies dependencies,
5652
RequestPipeline<SdkHttpFullRequest, CompletableFuture<Response<OutputT>>> requestPipeline) {
5753
this.responseHandler = responseHandler;
5854
this.dependencies = dependencies;
5955
this.scheduledExecutor = dependencies.clientConfiguration().option(SdkClientOption.SCHEDULED_EXECUTOR_SERVICE);
60-
this.rateLimitingTokenBucket = new RateLimitingTokenBucket();
6156
this.requestPipeline = requestPipeline;
6257
}
6358

64-
@SdkTestInternalApi
65-
public AsyncRetryableStage(TransformingAsyncResponseHandler<Response<OutputT>> responseHandler,
66-
HttpClientDependencies dependencies,
67-
RequestPipeline<SdkHttpFullRequest, CompletableFuture<Response<OutputT>>> requestPipeline,
68-
RateLimitingTokenBucket rateLimitingTokenBucket) {
69-
this.responseHandler = responseHandler;
70-
this.dependencies = dependencies;
71-
this.scheduledExecutor = dependencies.clientConfiguration().option(SdkClientOption.SCHEDULED_EXECUTOR_SERVICE);
72-
this.requestPipeline = requestPipeline;
73-
this.rateLimitingTokenBucket = rateLimitingTokenBucket;
74-
}
75-
7659
@Override
7760
public CompletableFuture<Response<OutputT>> execute(SdkHttpFullRequest request,
7861
RequestExecutionContext context) throws Exception {
7962
return new RetryingExecutor(request, context).execute();
8063
}
8164

82-
private class RetryingExecutor {
65+
private final class RetryingExecutor {
8366
private final AsyncRequestBody originalRequestBody;
8467
private final RequestExecutionContext context;
8568
private final RetryableStageHelper retryableStageHelper;
8669

8770
private RetryingExecutor(SdkHttpFullRequest request, RequestExecutionContext context) {
8871
this.originalRequestBody = context.requestProvider();
8972
this.context = context;
90-
this.retryableStageHelper = new RetryableStageHelper(request, context, rateLimitingTokenBucket, dependencies);
73+
this.retryableStageHelper = new RetryableStageHelper(request, context, dependencies);
9174
}
9275

93-
public CompletableFuture<Response<OutputT>> execute() throws Exception {
76+
public CompletableFuture<Response<OutputT>> execute() {
9477
CompletableFuture<Response<OutputT>> future = new CompletableFuture<>();
95-
maybeAttemptExecute(future);
78+
attemptFirstExecute(future);
9679
return future;
9780
}
9881

99-
public void maybeAttemptExecute(CompletableFuture<Response<OutputT>> future) {
100-
retryableStageHelper.startingAttempt();
101-
102-
if (!retryableStageHelper.retryPolicyAllowsRetry()) {
103-
future.completeExceptionally(retryableStageHelper.retryPolicyDisallowedRetryException());
104-
return;
105-
}
106-
107-
if (retryableStageHelper.getAttemptNumber() > 1) {
108-
// We failed the last attempt, but will retry. The response handler wants to know when that happens.
109-
responseHandler.onError(retryableStageHelper.getLastException());
110-
111-
// Reset the request provider to the original one before retries, in case it was modified downstream.
112-
context.requestProvider(originalRequestBody);
113-
}
114-
115-
Duration backoffDelay = retryableStageHelper.getBackoffDelay();
116-
117-
OptionalDouble tokenAcquireTimeSeconds = retryableStageHelper.getSendTokenNonBlocking();
118-
if (!tokenAcquireTimeSeconds.isPresent()) {
119-
String errorMessage = "Unable to acquire a send token immediately without waiting. This indicates that ADAPTIVE "
120-
+ "retry mode is enabled, fast fail rate limiting is enabled, and that rate limiting is "
121-
+ "engaged because of prior throttled requests. The request will not be executed.";
122-
future.completeExceptionally(SdkClientException.create(errorMessage));
123-
return;
124-
}
125-
long tokenAcquireTimeMillis = (long) (tokenAcquireTimeSeconds.getAsDouble() * 1000);
126-
127-
if (!backoffDelay.isZero()) {
82+
public void attemptFirstExecute(CompletableFuture<Response<OutputT>> future) {
83+
Duration backoffDelay = retryableStageHelper.acquireInitialToken();
84+
if (backoffDelay.isZero()) {
85+
attemptExecute(future);
86+
} else {
12887
retryableStageHelper.logBackingOff(backoffDelay);
129-
}
130-
131-
long totalDelayMillis = backoffDelay.toMillis() + tokenAcquireTimeMillis;
132-
133-
if (totalDelayMillis > 0) {
88+
long totalDelayMillis = backoffDelay.toMillis();
13489
scheduledExecutor.schedule(() -> attemptExecute(future), totalDelayMillis, MILLISECONDS);
135-
} else {
136-
attemptExecute(future);
13790
}
13891
}
13992

14093
private void attemptExecute(CompletableFuture<Response<OutputT>> future) {
14194
CompletableFuture<Response<OutputT>> responseFuture;
14295
try {
96+
retryableStageHelper.startingAttempt();
14397
retryableStageHelper.logSendingRequest();
14498
responseFuture = requestPipeline.execute(retryableStageHelper.requestToSend(), context);
14599

@@ -164,23 +118,37 @@ private void attemptExecute(CompletableFuture<Response<OutputT>> future) {
164118
}
165119

166120
retryableStageHelper.setLastResponse(response.httpResponse());
167-
168121
if (!response.isSuccess()) {
169122
retryableStageHelper.adjustClockIfClockSkew(response);
170123
maybeRetryExecute(future, response.exception());
171124
return;
172125
}
173126

174-
retryableStageHelper.updateClientSendingRateForSuccessResponse();
175-
176-
retryableStageHelper.attemptSucceeded();
127+
retryableStageHelper.recordAttemptSucceeded();
177128
future.complete(response);
178129
});
179130
}
180131

132+
public void maybeAttemptExecute(CompletableFuture<Response<OutputT>> future) {
133+
Optional<Duration> delay = retryableStageHelper.tryRefreshToken(Duration.ZERO);
134+
if (!delay.isPresent()) {
135+
future.completeExceptionally(retryableStageHelper.retryPolicyDisallowedRetryException());
136+
return;
137+
}
138+
// We failed the last attempt, but will retry. The response handler wants to know when that happens.
139+
responseHandler.onError(retryableStageHelper.getLastException());
140+
141+
// Reset the request provider to the original one before retries, in case it was modified downstream.
142+
context.requestProvider(originalRequestBody);
143+
144+
Duration backoffDelay = delay.get();
145+
retryableStageHelper.logBackingOff(backoffDelay);
146+
long totalDelayMillis = backoffDelay.toMillis();
147+
scheduledExecutor.schedule(() -> attemptExecute(future), totalDelayMillis, MILLISECONDS);
148+
}
149+
181150
private void maybeRetryExecute(CompletableFuture<Response<OutputT>> future, Exception exception) {
182151
retryableStageHelper.setLastException(exception);
183-
retryableStageHelper.updateClientSendingRateForErrorResponse();
184152
maybeAttemptExecute(future);
185153
}
186154
}

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

Lines changed: 0 additions & 155 deletions
This file was deleted.

0 commit comments

Comments
 (0)