From 89df3785b240a117724b017fae8067e3b2ed5fc5 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Mon, 19 Oct 2026 18:34:50 +0000 Subject: [PATCH] fix(bigtable): add exponential backoff on client transport failure retries --- .../v2/internal/middleware/RetryingVRpc.java | 43 ++++-- .../internal/middleware/RetryingVRpcTest.java | 126 ++++++++++++++++++ 2 files changed, 155 insertions(+), 14 deletions(-) diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/RetryingVRpc.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/RetryingVRpc.java index 206ff052bbdc..b9428a2debaa 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/RetryingVRpc.java +++ b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/RetryingVRpc.java @@ -21,10 +21,8 @@ import com.google.common.base.Stopwatch; import com.google.protobuf.Duration; import com.google.protobuf.util.Durations; -import com.google.rpc.RetryInfo; import io.grpc.Context; import io.grpc.Status; -import java.util.Optional; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; import java.util.logging.Level; @@ -37,6 +35,9 @@ public class RetryingVRpc implements VRpc { private static final Logger LOG = Logger.getLogger(RetryingVRpc.class.getName()); + static final long INITIAL_TRANSPORT_FAILURE_BACKOFF_MS = 10L; + static final long MAX_TRANSPORT_FAILURE_BACKOFF_MS = 1000L; + private final Context grpcContext; private final io.opentelemetry.context.Context otelContext; @@ -239,12 +240,9 @@ public void onClose(VRpcResult result) { return; } finishAttempt(result); - if (shouldRetry(result)) { + Duration retryDelay = getRetryDelay(result); + if (shouldRetry(result, retryDelay)) { context = context.createForNextAttempt(); - Duration retryDelay = - Optional.ofNullable(result.getRetryInfo()) - .map(RetryInfo::getRetryDelay) - .orElse(Durations.ZERO); if (Durations.compare(retryDelay, Durations.ZERO) > 0) { Scheduled scheduled = new Scheduled(retryDelay); onStateChange(scheduled); @@ -284,17 +282,34 @@ public void onExit() { Status.CANCELLED.withDescription("attempt abandoned during transition"))); } - boolean shouldRetry(VRpcResult result) { + Duration getRetryDelay(VRpcResult result) { + if (result.getRetryInfo() != null && result.getRetryInfo().hasRetryDelay()) { + return result.getRetryInfo().getRetryDelay(); + } + if (result.getState() == VRpcResult.State.TRANSPORT_FAILURE) { + int attemptNumber = context.getOperationInfo().getAttemptNumber(); + long delayMs = + Math.min( + MAX_TRANSPORT_FAILURE_BACKOFF_MS, + INITIAL_TRANSPORT_FAILURE_BACKOFF_MS * (1L << Math.min(attemptNumber, 16))); + return Durations.fromMillis(delayMs); + } + return Durations.ZERO; + } + + boolean shouldRetry(VRpcResult result, Duration retryDelay) { + long retryDelayMs = Durations.toMillis(retryDelay); + // only schedule retry if there's still a chance for the request to succeed + boolean hasRemainingTime = + context.getOperationInfo().getDeadline().timeRemaining(TimeUnit.MILLISECONDS) + - retryDelayMs + > 1; // If the error has RetryInfo, it means it comes from the server and should // be retried. if (!result.getStatus().isOk() && result.getRetryInfo() != null && result.getRetryInfo().hasRetryDelay()) { - long retryDelay = Durations.toMillis(result.getRetryInfo().getRetryDelay()); - // only schedule retry if there's still a chance for the request to succeed - return context.getOperationInfo().getDeadline().timeRemaining(TimeUnit.MILLISECONDS) - - retryDelay - > 1; + return hasRemainingTime; } // Do not retry result that is explicitly rejected if (result.getRejected()) { @@ -307,7 +322,7 @@ boolean shouldRetry(VRpcResult result) { || (context.getOperationInfo().isIdempotent() && result.getState() == VRpcResult.State.TRANSPORT_FAILURE); if (isRetryable && context.getOperationInfo().getAttemptNumber() < 3) { - return true; + return hasRemainingTime; } return false; } diff --git a/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/middleware/RetryingVRpcTest.java b/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/middleware/RetryingVRpcTest.java index 0109785ffe98..f08f43339efd 100644 --- a/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/middleware/RetryingVRpcTest.java +++ b/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/middleware/RetryingVRpcTest.java @@ -61,10 +61,16 @@ import io.grpc.Server; import io.grpc.Status; import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import javax.annotation.Nullable; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -414,4 +420,124 @@ void cancelInScheduledState() throws Exception { VRpcException vRpcException = (VRpcException) executionException.getCause(); assertThat(vRpcException.getStatus().getCode()).isEqualTo(Status.Code.CANCELLED); } + + @Test + void retryTransportFailureWithExponentialBackoff() throws Exception { + List scheduledDelaysMs = Collections.synchronizedList(new ArrayList<>()); + BigtableTimer recordingTimer = + new BigtableTimer() { + @Override + public Timeout newTimeout(Runnable task, Executor exec, long delay, TimeUnit unit) { + scheduledDelaysMs.add(unit.toMillis(delay)); + return timer.newTimeout(task, exec, delay, unit); + } + + @Override + public void stop() { + timer.stop(); + } + + @Override + public Registration onStop(Runnable hook) { + return timer.onStop(hook); + } + }; + + AtomicInteger attemptCount = new AtomicInteger(); + RetryingVRpc retrying = + new RetryingVRpc<>( + () -> + new VRpc() { + @Override + public void start( + SessionFakeScriptedRequest req, + VRpc.VRpcCallContext ctx, + VRpc.VRpcListener listener) { + int attempt = attemptCount.getAndIncrement(); + ctx.getExecutor() + .execute( + () -> { + if (attempt < 3) { + listener.onClose( + VRpc.VRpcResult.createLocalTransportError( + Status.UNAVAILABLE.withDescription("missed heartbeat"))); + } else { + listener.onMessage( + SessionFakeScriptedResponse.getDefaultInstance()); + listener.onClose( + VRpc.VRpcResult.createServerOk( + VirtualRpcResponse.getDefaultInstance())); + } + }); + } + + @Override + public void cancel(@Nullable String message, @Nullable Throwable cause) {} + + @Override + public boolean isDone() { + return true; + } + + @Override + public void requestNext() {} + }, + recordingTimer); + + UnaryResponseFuture f = new UnaryResponseFuture<>(); + retrying.start( + SessionFakeScriptedRequest.getDefaultInstance(), + VRpc.VRpcCallContext.create(Deadline.after(1, TimeUnit.MINUTES), true, tracer), + f); + + assertThat(f.get()).isEqualTo(SessionFakeScriptedResponse.getDefaultInstance()); + assertThat(attemptCount.get()).isEqualTo(4); + assertThat(scheduledDelaysMs).containsExactly(10L, 20L, 40L).inOrder(); + } + + @Test + void retryTransportFailureDeadlineExceeded() { + AtomicInteger attemptCount = new AtomicInteger(); + RetryingVRpc retrying = + new RetryingVRpc<>( + () -> + new VRpc() { + @Override + public void start( + SessionFakeScriptedRequest req, + VRpc.VRpcCallContext ctx, + VRpc.VRpcListener listener) { + attemptCount.incrementAndGet(); + ctx.getExecutor() + .execute( + () -> + listener.onClose( + VRpc.VRpcResult.createLocalTransportError( + Status.UNAVAILABLE.withDescription("missed heartbeat")))); + } + + @Override + public void cancel(@Nullable String message, @Nullable Throwable cause) {} + + @Override + public boolean isDone() { + return true; + } + + @Override + public void requestNext() {} + }, + timer); + + UnaryResponseFuture f = new UnaryResponseFuture<>(); + // Initial transport failure retry delay is 10ms, which exceeds a 5ms deadline. + retrying.start( + SessionFakeScriptedRequest.getDefaultInstance(), + VRpc.VRpcCallContext.create(Deadline.after(5, TimeUnit.MILLISECONDS), true, tracer), + f); + + ExecutionException exception = assertThrows(ExecutionException.class, f::get); + assertThat(exception).hasMessageThat().contains("missed heartbeat"); + assertThat(attemptCount.get()).isEqualTo(1); + } }