fix(bigtable): do not retry on negative RetryInfo delay in RetryingVRpc

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

@@ -249,6 +249,8 @@ public void onClose(VRpcResult result) {
                   Scheduled scheduled = new Scheduled(retryDelay);
                   onStateChange(scheduled);
                 } else {
+                  // Immediate retry for client-initiated retries or server RetryInfo with 0s
+                  // delay (e.g. after GOAWAY).
                   onStateChange(new Idle());
                 }
                 return;
@@ -297,9 +299,17 @@ boolean shouldRetry(VRpcResult result) {
           context.getOperationInfo().getDeadline().timeRemaining(TimeUnit.MILLISECONDS);
 
       // If the error has RetryInfo, it means it comes from the server and should
-      // be retried as long as there's time remaining after the backoff.
+      // be retried as long as the delay is non-negative and there's time remaining
+      // after the backoff. The server caps RetryInfo-directed attempts and may send
+      // a 0s delay (e.g. after GOAWAY) to request an immediate retry on another session.
       if (result.getRetryInfo() != null && result.getRetryInfo().hasRetryDelay()) {
-        long retryDelayMs = Durations.toMillis(result.getRetryInfo().getRetryDelay());
+        Duration retryDelay = result.getRetryInfo().getRetryDelay();
+        // A negative retry_delay should never happen in practice, but guard against it so it
+        // cannot invert the remaining-deadline check below.
+        if (Durations.compare(retryDelay, Durations.ZERO) < 0) {
+          return false;
+        }
+        long retryDelayMs = Durations.toMillis(retryDelay);
         return timeRemainingMs - retryDelayMs > 1;
       }
 

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

@@ -521,4 +521,62 @@ public void requestNext() {}
     assertThat(vRpcException.getStatus().getCode()).isEqualTo(Status.Code.UNAVAILABLE);
     assertThat(attempts.get()).isEqualTo(1);
   }
+
+  @Test
+  void negativeRetryDelayNotRetried() {
+    AtomicInteger attempts = new AtomicInteger();
+    ErrorResponse.Builder errorBuilder =
+        ErrorResponse.newBuilder()
+            .setRpcId(1)
+            .setRetryInfo(RetryInfo.newBuilder().setRetryDelay(Durations.fromSeconds(-10)).build())
+            .setClusterInfo(ClusterInformation.getDefaultInstance());
+    errorBuilder
+        .getStatusBuilder()
+        .setCode(Code.UNAVAILABLE.getNumber())
+        .setMessage("server unavailable");
+    VRpc.VRpcResult negativeDelayResult = VRpc.VRpcResult.createServerError(errorBuilder.build());
+
+    RetryingVRpc<SessionFakeScriptedRequest, SessionFakeScriptedResponse> retrying =
+        new RetryingVRpc<>(
+            () ->
+                new VRpc<SessionFakeScriptedRequest, SessionFakeScriptedResponse>() {
+                  private boolean done;
+
+                  @Override
+                  public void start(
+                      SessionFakeScriptedRequest req,
+                      VRpcCallContext ctx,
+                      VRpcListener<SessionFakeScriptedResponse> listener) {
+                    attempts.incrementAndGet();
+                    done = true;
+                    ctx.getExecutor().execute(() -> listener.onClose(negativeDelayResult));
+                  }
+
+                  @Override
+                  public void cancel(String message, Throwable cause) {}
+
+                  @Override
+                  public boolean isDone() {
+                    return done;
+                  }
+
+                  @Override
+                  public void requestNext() {}
+                },
+            timer);
+
+    UnaryResponseFuture<SessionFakeScriptedResponse> f = new UnaryResponseFuture<>();
+    retrying.start(
+        SessionFakeScriptedRequest.getDefaultInstance(),
+        VRpc.VRpcCallContext.create(Deadline.after(1, TimeUnit.HOURS), true, tracer),
+        f);
+
+    ExecutionException executionException = assertThrows(ExecutionException.class, f::get);
+    assertThat(executionException.getCause()).isInstanceOf(VRpcException.class);
+    VRpcException vRpcException = (VRpcException) executionException.getCause();
+    assertThat(vRpcException.getStatus().getCode()).isEqualTo(Status.Code.UNAVAILABLE);
+    assertThat(vRpcException.getStatus().getDescription()).isEqualTo("server unavailable");
+    assertThat(vRpcException.getResult().getState()).isEqualTo(VRpc.VRpcResult.State.SERVER_RESULT);
+    assertThat(attempts.get()).isEqualTo(1);
+  }
 }