Skip to content

Commit 4db9525

Browse files
schiemonschiemon
committed
[WIP] feat, test: make sure each attempt is started with a consistent attempt number
Also add more tests to Retrying(Rpc)Client
1 parent 6a3db67 commit 4db9525

5 files changed

Lines changed: 312 additions & 58 deletions

File tree

‎core/src/main/java/com/linecorp/armeria/client/retry/AbstractRetryingClient.java‎

Lines changed: 17 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -180,7 +180,8 @@ protected final RetryRuleWithContent<O> retryRuleWithContent() {
180180
protected void startRetryAttempt(ClientRequestContext ctx,
181181
ClientRequestContext attemptCtx,
182182
BiConsumer<ClientRequestContext, O> onAcceptHandler,
183-
BiConsumer<ClientRequestContext, @Nullable Throwable> onAttemptAbortedHandler
183+
BiConsumer<ClientRequestContext, @Nullable Throwable>
184+
onAttemptAbortedHandler
184185
) {
185186
requireNonNull(ctx, "ctx");
186187
requireNonNull(attemptCtx, "attemptCtx");
@@ -231,11 +232,13 @@ protected static <O extends Response> void completeRetryAttempt(ClientRequestCon
231232
return;
232233
}
233234

234-
synchronized (state(ctx)) {
235-
state(ctx).completeAttempt(attemptCtx, attemptRes);
235+
final State<O> state = state(ctx);
236+
237+
synchronized (state) {
238+
state.completeAttempt(attemptCtx, attemptRes);
236239

237240
if (isWinning) {
238-
state(ctx).complete();
241+
state.complete();
239242
}
240243
}
241244
}
@@ -257,7 +260,7 @@ protected static boolean isRetryingComplete(ClientRequestContext ctx) {
257260
* todo(szymon): [doc].
258261
*/
259262
protected void scheduleNextRetry(ClientRequestContext ctx,
260-
Runnable retryTask,
263+
Consumer<Integer> retryTask,
261264
Backoff backoff,
262265
Consumer<? super Throwable> actionOnException) {
263266
scheduleNextRetry(ctx, retryTask, backoff, -1, actionOnException);
@@ -267,7 +270,7 @@ protected void scheduleNextRetry(ClientRequestContext ctx,
267270
* todo(szymon): [doc].
268271
*/
269272
protected static void scheduleNextRetry(ClientRequestContext ctx,
270-
Runnable retryTask,
273+
Consumer<Integer> retryTask,
271274
Backoff backoff,
272275
long retryDelayFromServerMillis,
273276
Consumer<? super Throwable> actionOnException) {
@@ -332,6 +335,8 @@ protected static void scheduleNextRetry(ClientRequestContext ctx,
332335

333336
state.startRetryTask();
334337
scheduler.schedule(() -> {
338+
339+
final int thisAttemptNo;
335340
// *, see comment above.
336341
synchronized (state) {
337342
if (isRetryingComplete(ctx)) {
@@ -340,10 +345,10 @@ protected static void scheduleNextRetry(ClientRequestContext ctx,
340345
return;
341346
}
342347

343-
state.acquireAttemptNoWithCurrentBackoff(backoff);
348+
thisAttemptNo = state.acquireAttemptNoWithCurrentBackoff(backoff);
344349
}
345350

346-
retryTask.run();
351+
retryTask.accept(thisAttemptNo);
347352

348353
synchronized (state) {
349354
state.completeRetryTask();
@@ -578,10 +583,10 @@ int nextAttemptNoWithBackoff(Backoff backoff) {
578583
return 1;
579584
}
580585

581-
return currentAttemptNoWithLastBackoff + 1;
586+
return currentAttemptNoWithLastBackoff;
582587
}
583588

584-
void acquireAttemptNoWithCurrentBackoff(Backoff backoff) {
589+
int acquireAttemptNoWithCurrentBackoff(Backoff backoff) {
585590
checkState(!retryingCompleteFuture.isDone());
586591
checkState((totalAttemptNo + 1) <= config.maxTotalAttempts(),
587592
"Exceeded the maximum number of attempts: %s", config.maxTotalAttempts());
@@ -591,10 +596,11 @@ void acquireAttemptNoWithCurrentBackoff(Backoff backoff) {
591596
if (lastBackoff != backoff) {
592597
lastBackoff = backoff;
593598
currentAttemptNoWithLastBackoff = 1;
594-
return;
595599
}
596600

597601
currentAttemptNoWithLastBackoff++;
602+
603+
return totalAttemptNo;
598604
}
599605

600606
int numPendingAttempts() {

‎core/src/main/java/com/linecorp/armeria/client/retry/RetryingClient.java‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -249,7 +249,7 @@ protected HttpResponse doExecute(ClientRequestContext ctx, HttpRequest req) thro
249249
if (ctx.exchangeType().isRequestStreaming()) {
250250
final HttpRequestDuplicator reqDuplicator = req.toDuplicator(ctx.eventLoop().withoutContext(), 0);
251251
doExecute0(new RetryingContext(mappedRetryConfig(ctx), ctx, reqDuplicator, req, res,
252-
responseFuture));
252+
responseFuture), 1);
253253
} else {
254254
req.aggregate(AggregationOptions.usePooledObjects(ctx.alloc(), ctx.eventLoop()))
255255
.handle((agg, cause) -> {
@@ -258,7 +258,7 @@ protected HttpResponse doExecute(ClientRequestContext ctx, HttpRequest req) thro
258258
} else {
259259
final HttpRequestDuplicator reqDuplicator = new AggregatedHttpRequestDuplicator(agg);
260260
doExecute0(new RetryingContext(mappedRetryConfig(ctx), ctx, reqDuplicator, req, res,
261-
responseFuture));
261+
responseFuture), 1);
262262
}
263263
return null;
264264
});
@@ -267,7 +267,7 @@ protected HttpResponse doExecute(ClientRequestContext ctx, HttpRequest req) thro
267267
return res;
268268
}
269269

270-
private void doExecute0(RetryingContext retryingContext) {
270+
private void doExecute0(RetryingContext retryingContext, int attemptNo) {
271271

272272
final RetryConfig<HttpResponse> config = retryingContext.config();
273273
final ClientRequestContext ctx = retryingContext.ctx();
@@ -277,8 +277,7 @@ private void doExecute0(RetryingContext retryingContext) {
277277

278278
// todo(szymon): we need to inject the attempt number as there may be concurrent attempts
279279
// starting and acquiring an attempt number.
280-
final int totalAttempts = getTotalAttempts(ctx);
281-
final boolean initialAttempt = totalAttempts <= 1;
280+
final boolean initialAttempt = attemptNo <= 1;
282281
// The request or attemptRes has been aborted by the client before it receives a attemptRes,
283282
// so stop retrying.
284283
if (originalReq.whenComplete().isCompletedExceptionally()) {
@@ -312,7 +311,7 @@ private void doExecute0(RetryingContext retryingContext) {
312311
attemptReq = rootReqDuplicator.duplicate();
313312
} else {
314313
final RequestHeadersBuilder newHeaders = originalReq.headers().toBuilder();
315-
newHeaders.setInt(ARMERIA_RETRY_COUNT, totalAttempts - 1);
314+
newHeaders.setInt(ARMERIA_RETRY_COUNT, attemptNo - 1);
316315
attemptReq = rootReqDuplicator.duplicate(newHeaders.build());
317316
}
318317

@@ -400,7 +399,7 @@ private void doExecute0(RetryingContext retryingContext) {
400399

401400
if (hedgingDelayMillis >= 0) {
402401
final Backoff hedgingDelayBackoff = Backoff.fixed(hedgingDelayMillis);
403-
scheduleNextRetry(ctx, () -> doExecute0(retryingContext),
402+
scheduleNextRetry(ctx, hedgingAttemptNo -> doExecute0(retryingContext, hedgingAttemptNo),
404403
hedgingDelayBackoff,
405404
cause -> handleExceptionAfterScheduling(retryingContext, cause));
406405
}
@@ -645,7 +644,7 @@ private void handleRetryDecision(RetryingContext retryingContext, @Nullable Retr
645644
if (backoff != null) {
646645
final long millisAfter = useRetryAfter ? getRetryAfterMillis(attemptCtx) : -1;
647646
scheduleNextRetry(retryingContext.ctx(),
648-
() -> doExecute0(retryingContext),
647+
attemptNo -> doExecute0(retryingContext, attemptNo),
649648
backoff,
650649
millisAfter,
651650
cause -> handleExceptionAfterScheduling(retryingContext, cause));

‎core/src/main/java/com/linecorp/armeria/client/retry/RetryingRpcClient.java‎

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -145,14 +145,14 @@ public static RetryingRpcClientBuilder builder(RetryConfigMapping<RpcResponse> m
145145
protected RpcResponse doExecute(ClientRequestContext ctx, RpcRequest req) throws Exception {
146146
final CompletableFuture<RpcResponse> returnedResFuture = new CompletableFuture<>();
147147
final RpcResponse res = RpcResponse.from(returnedResFuture);
148-
doExecute0(ctx, req, res, returnedResFuture);
148+
doExecute0(ctx, req, res, returnedResFuture, 1);
149149
return res;
150150
}
151151

152152
private void doExecute0(ClientRequestContext ctx, RpcRequest req,
153-
RpcResponse returnedRes, CompletableFuture<RpcResponse> returnedResFuture) {
154-
final int totalAttempts = getTotalAttempts(ctx);
155-
final boolean initialAttempt = totalAttempts <= 1;
153+
RpcResponse returnedRes, CompletableFuture<RpcResponse> returnedResFuture,
154+
int attemptNo) {
155+
final boolean initialAttempt = attemptNo <= 1;
156156
if (returnedRes.isDone()) {
157157
// The response has been cancelled by the client before it receives a response, so stop retrying.
158158
completeRetryingExceptionally(ctx, returnedResFuture, new CancellationException(
@@ -169,7 +169,7 @@ private void doExecute0(ClientRequestContext ctx, RpcRequest req,
169169

170170
if (!initialAttempt) {
171171
attemptCtx.mutateAdditionalRequestHeaders(
172-
mutator -> mutator.add(ARMERIA_RETRY_COUNT, StringUtil.toString(totalAttempts - 1)));
172+
mutator -> mutator.add(ARMERIA_RETRY_COUNT, StringUtil.toString(attemptNo - 1)));
173173
}
174174

175175
final RpcResponse attemptRes;
@@ -218,7 +218,8 @@ private void doExecute0(ClientRequestContext ctx, RpcRequest req,
218218

219219
if (backoff != null) {
220220
scheduleNextRetry(ctx,
221-
() -> doExecute0(ctx, req, returnedRes, returnedResFuture),
221+
nextAttemptNo -> doExecute0(ctx, req, returnedRes,
222+
returnedResFuture, nextAttemptNo),
222223
backoff,
223224
cause0 -> handleExceptionAfterScheduling(ctx, returnedResFuture,
224225
cause0));
@@ -237,7 +238,9 @@ private void doExecute0(ClientRequestContext ctx, RpcRequest req,
237238

238239
if (hedgingDelayMillis >= 0) {
239240
final Backoff hedgingBackoff = Backoff.fixed(hedgingDelayMillis);
240-
scheduleNextRetry(ctx, () -> doExecute0(ctx, req, returnedRes, returnedResFuture),
241+
scheduleNextRetry(ctx, hedgingAttemptNo ->
242+
doExecute0(ctx, req, returnedRes, returnedResFuture,
243+
hedgingAttemptNo),
241244
hedgingBackoff,
242245
cause -> handleExceptionAfterScheduling(ctx, returnedResFuture, cause));
243246
}

0 commit comments

Comments
 (0)