Skip to content

Commit 89df378

Browse files
mutianfmutianf
committed
fix(bigtable): add exponential backoff on client transport failure retries
1 parent e666612 commit 89df378

2 files changed

Lines changed: 155 additions & 14 deletions

File tree

  • java-bigtable/google-cloud-bigtable/src

‎java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/RetryingVRpc.java‎

Lines changed: 29 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -21,10 +21,8 @@
2121
import com.google.common.base.Stopwatch;
2222
import com.google.protobuf.Duration;
2323
import com.google.protobuf.util.Durations;
24-
import com.google.rpc.RetryInfo;
2524
import io.grpc.Context;
2625
import io.grpc.Status;
27-
import java.util.Optional;
2826
import java.util.concurrent.TimeUnit;
2927
import java.util.function.Supplier;
3028
import java.util.logging.Level;
@@ -37,6 +35,9 @@ public class RetryingVRpc<ReqT, RespT> implements VRpc<ReqT, RespT> {
3735

3836
private static final Logger LOG = Logger.getLogger(RetryingVRpc.class.getName());
3937

38+
static final long INITIAL_TRANSPORT_FAILURE_BACKOFF_MS = 10L;
39+
static final long MAX_TRANSPORT_FAILURE_BACKOFF_MS = 1000L;
40+
4041
private final Context grpcContext;
4142
private final io.opentelemetry.context.Context otelContext;
4243

@@ -239,12 +240,9 @@ public void onClose(VRpcResult result) {
239240
return;
240241
}
241242
finishAttempt(result);
242-
if (shouldRetry(result)) {
243+
Duration retryDelay = getRetryDelay(result);
244+
if (shouldRetry(result, retryDelay)) {
243245
context = context.createForNextAttempt();
244-
Duration retryDelay =
245-
Optional.ofNullable(result.getRetryInfo())
246-
.map(RetryInfo::getRetryDelay)
247-
.orElse(Durations.ZERO);
248246
if (Durations.compare(retryDelay, Durations.ZERO) > 0) {
249247
Scheduled scheduled = new Scheduled(retryDelay);
250248
onStateChange(scheduled);
@@ -284,17 +282,34 @@ public void onExit() {
284282
Status.CANCELLED.withDescription("attempt abandoned during transition")));
285283
}
286284

287-
boolean shouldRetry(VRpcResult result) {
285+
Duration getRetryDelay(VRpcResult result) {
286+
if (result.getRetryInfo() != null && result.getRetryInfo().hasRetryDelay()) {
287+
return result.getRetryInfo().getRetryDelay();
288+
}
289+
if (result.getState() == VRpcResult.State.TRANSPORT_FAILURE) {
290+
int attemptNumber = context.getOperationInfo().getAttemptNumber();
291+
long delayMs =
292+
Math.min(
293+
MAX_TRANSPORT_FAILURE_BACKOFF_MS,
294+
INITIAL_TRANSPORT_FAILURE_BACKOFF_MS * (1L << Math.min(attemptNumber, 16)));
295+
return Durations.fromMillis(delayMs);
296+
}
297+
return Durations.ZERO;
298+
}
299+
300+
boolean shouldRetry(VRpcResult result, Duration retryDelay) {
301+
long retryDelayMs = Durations.toMillis(retryDelay);
302+
// only schedule retry if there's still a chance for the request to succeed
303+
boolean hasRemainingTime =
304+
context.getOperationInfo().getDeadline().timeRemaining(TimeUnit.MILLISECONDS)
305+
- retryDelayMs
306+
> 1;
288307
// If the error has RetryInfo, it means it comes from the server and should
289308
// be retried.
290309
if (!result.getStatus().isOk()
291310
&& result.getRetryInfo() != null
292311
&& result.getRetryInfo().hasRetryDelay()) {
293-
long retryDelay = Durations.toMillis(result.getRetryInfo().getRetryDelay());
294-
// only schedule retry if there's still a chance for the request to succeed
295-
return context.getOperationInfo().getDeadline().timeRemaining(TimeUnit.MILLISECONDS)
296-
- retryDelay
297-
> 1;
312+
return hasRemainingTime;
298313
}
299314
// Do not retry result that is explicitly rejected
300315
if (result.getRejected()) {
@@ -307,7 +322,7 @@ boolean shouldRetry(VRpcResult result) {
307322
|| (context.getOperationInfo().isIdempotent()
308323
&& result.getState() == VRpcResult.State.TRANSPORT_FAILURE);
309324
if (isRetryable && context.getOperationInfo().getAttemptNumber() < 3) {
310-
return true;
325+
return hasRemainingTime;
311326
}
312327
return false;
313328
}

‎java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/middleware/RetryingVRpcTest.java‎

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,10 +61,16 @@
6161
import io.grpc.Server;
6262
import io.grpc.Status;
6363
import java.io.IOException;
64+
import java.util.ArrayList;
65+
import java.util.Collections;
66+
import java.util.List;
6467
import java.util.concurrent.ExecutionException;
68+
import java.util.concurrent.Executor;
6569
import java.util.concurrent.Executors;
6670
import java.util.concurrent.ScheduledExecutorService;
6771
import java.util.concurrent.TimeUnit;
72+
import java.util.concurrent.atomic.AtomicInteger;
73+
import javax.annotation.Nullable;
6874
import org.junit.jupiter.api.AfterEach;
6975
import org.junit.jupiter.api.BeforeEach;
7076
import org.junit.jupiter.api.Test;
@@ -414,4 +420,124 @@ void cancelInScheduledState() throws Exception {
414420
VRpcException vRpcException = (VRpcException) executionException.getCause();
415421
assertThat(vRpcException.getStatus().getCode()).isEqualTo(Status.Code.CANCELLED);
416422
}
423+
424+
@Test
425+
void retryTransportFailureWithExponentialBackoff() throws Exception {
426+
List<Long> scheduledDelaysMs = Collections.synchronizedList(new ArrayList<>());
427+
BigtableTimer recordingTimer =
428+
new BigtableTimer() {
429+
@Override
430+
public Timeout newTimeout(Runnable task, Executor exec, long delay, TimeUnit unit) {
431+
scheduledDelaysMs.add(unit.toMillis(delay));
432+
return timer.newTimeout(task, exec, delay, unit);
433+
}
434+
435+
@Override
436+
public void stop() {
437+
timer.stop();
438+
}
439+
440+
@Override
441+
public Registration onStop(Runnable hook) {
442+
return timer.onStop(hook);
443+
}
444+
};
445+
446+
AtomicInteger attemptCount = new AtomicInteger();
447+
RetryingVRpc<SessionFakeScriptedRequest, SessionFakeScriptedResponse> retrying =
448+
new RetryingVRpc<>(
449+
() ->
450+
new VRpc<SessionFakeScriptedRequest, SessionFakeScriptedResponse>() {
451+
@Override
452+
public void start(
453+
SessionFakeScriptedRequest req,
454+
VRpc.VRpcCallContext ctx,
455+
VRpc.VRpcListener<SessionFakeScriptedResponse> listener) {
456+
int attempt = attemptCount.getAndIncrement();
457+
ctx.getExecutor()
458+
.execute(
459+
() -> {
460+
if (attempt < 3) {
461+
listener.onClose(
462+
VRpc.VRpcResult.createLocalTransportError(
463+
Status.UNAVAILABLE.withDescription("missed heartbeat")));
464+
} else {
465+
listener.onMessage(
466+
SessionFakeScriptedResponse.getDefaultInstance());
467+
listener.onClose(
468+
VRpc.VRpcResult.createServerOk(
469+
VirtualRpcResponse.getDefaultInstance()));
470+
}
471+
});
472+
}
473+
474+
@Override
475+
public void cancel(@Nullable String message, @Nullable Throwable cause) {}
476+
477+
@Override
478+
public boolean isDone() {
479+
return true;
480+
}
481+
482+
@Override
483+
public void requestNext() {}
484+
},
485+
recordingTimer);
486+
487+
UnaryResponseFuture<SessionFakeScriptedResponse> f = new UnaryResponseFuture<>();
488+
retrying.start(
489+
SessionFakeScriptedRequest.getDefaultInstance(),
490+
VRpc.VRpcCallContext.create(Deadline.after(1, TimeUnit.MINUTES), true, tracer),
491+
f);
492+
493+
assertThat(f.get()).isEqualTo(SessionFakeScriptedResponse.getDefaultInstance());
494+
assertThat(attemptCount.get()).isEqualTo(4);
495+
assertThat(scheduledDelaysMs).containsExactly(10L, 20L, 40L).inOrder();
496+
}
497+
498+
@Test
499+
void retryTransportFailureDeadlineExceeded() {
500+
AtomicInteger attemptCount = new AtomicInteger();
501+
RetryingVRpc<SessionFakeScriptedRequest, SessionFakeScriptedResponse> retrying =
502+
new RetryingVRpc<>(
503+
() ->
504+
new VRpc<SessionFakeScriptedRequest, SessionFakeScriptedResponse>() {
505+
@Override
506+
public void start(
507+
SessionFakeScriptedRequest req,
508+
VRpc.VRpcCallContext ctx,
509+
VRpc.VRpcListener<SessionFakeScriptedResponse> listener) {
510+
attemptCount.incrementAndGet();
511+
ctx.getExecutor()
512+
.execute(
513+
() ->
514+
listener.onClose(
515+
VRpc.VRpcResult.createLocalTransportError(
516+
Status.UNAVAILABLE.withDescription("missed heartbeat"))));
517+
}
518+
519+
@Override
520+
public void cancel(@Nullable String message, @Nullable Throwable cause) {}
521+
522+
@Override
523+
public boolean isDone() {
524+
return true;
525+
}
526+
527+
@Override
528+
public void requestNext() {}
529+
},
530+
timer);
531+
532+
UnaryResponseFuture<SessionFakeScriptedResponse> f = new UnaryResponseFuture<>();
533+
// Initial transport failure retry delay is 10ms, which exceeds a 5ms deadline.
534+
retrying.start(
535+
SessionFakeScriptedRequest.getDefaultInstance(),
536+
VRpc.VRpcCallContext.create(Deadline.after(5, TimeUnit.MILLISECONDS), true, tracer),
537+
f);
538+
539+
ExecutionException exception = assertThrows(ExecutionException.class, f::get);
540+
assertThat(exception).hasMessageThat().contains("missed heartbeat");
541+
assertThat(attemptCount.get()).isEqualTo(1);
542+
}
417543
}

0 commit comments

Comments
 (0)