fix: Create next attempt after first attempt to initialize exponentia…

…l backoff settings. (#2316)

---------

Co-authored-by: Evan Greco <egreco@google.com>

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java

@@ -971,15 +971,22 @@ private Boolean retryOnRetryableError(Code errorCode, AppendRequestAndResponse r
       try {
         requestWrapper.retryCount++;
         if (this.retrySettings != null && errorCode == Code.RESOURCE_EXHAUSTED) {
-          // Trigger exponential backoff in append loop when request is resent for quota errors
-          if (requestWrapper.attemptSettings == null) {
-            requestWrapper.attemptSettings = requestWrapper.retryAlgorithm.createFirstAttempt();
-          } else {
-            requestWrapper.attemptSettings =
-                requestWrapper.retryAlgorithm.createNextAttempt(requestWrapper.attemptSettings);
-          }
+          // Trigger exponential backoff in append loop when request is resent for quota errors.
+          // createNextAttempt correctly initializes the retry delay; createfirstAttempt does not
+          // include a positive delay, just 0.
+          requestWrapper.attemptSettings =
+              requestWrapper.retryAlgorithm.createNextAttempt(
+                  requestWrapper.attemptSettings == null
+                      ? requestWrapper.retryAlgorithm.createFirstAttempt()
+                      : requestWrapper.attemptSettings);
           requestWrapper.blockMessageSendDeadline =
               Instant.now().plusMillis(requestWrapper.attemptSettings.getRetryDelay().toMillis());
+          log.info(
+              "Messages blocked for retry for "
+                  + java.time.Duration.between(
+                      java.time.Instant.now(), requestWrapper.blockMessageSendDeadline)
+                  + " until "
+                  + requestWrapper.blockMessageSendDeadline);
         }
 
         Long offset =

google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/FakeBigQueryWrite.java

@@ -19,6 +19,8 @@
 import com.google.protobuf.AbstractMessage;
 import io.grpc.ServerServiceDefinition;
 import io.grpc.Status;
+import java.time.Instant;
+import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
@@ -128,4 +130,8 @@ public void setReturnErrorDuringExclusiveStreamRetry(boolean retryOnError) {
   public void setVerifyOffset(boolean verifyOffset) {
     serviceImpl.setVerifyOffset(verifyOffset);
   }
+
+  public ArrayList<Instant> getLatestRequestReceivedInstants() {
+    return serviceImpl.getLatestRequestReceivedInstants();
+  }
 }

google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/FakeBigQueryWriteImpl.java

@@ -20,6 +20,7 @@
 import com.google.rpc.Code;
 import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -73,6 +74,7 @@ class FakeBigQueryWriteImpl extends BigQueryWriteGrpc.BigQueryWriteImplBase {
   private final Map<StreamObserver<AppendRowsResponse>, Boolean> connectionToFirstRequest =
       new ConcurrentHashMap<>();
   private Status failedStatus = Status.ABORTED;
+  private ArrayList<Instant> requestReceivedInstants = new ArrayList<>();
 
   /** Class used to save the state of a possible response. */
   public static class Response {
@@ -111,6 +113,10 @@ public String toString() {
     }
   }
 
+  public ArrayList<Instant> getLatestRequestReceivedInstants() {
+    return requestReceivedInstants;
+  }
+
   @Override
   public void getWriteStream(
       GetWriteStreamRequest request, StreamObserver<WriteStream> responseObserver) {
@@ -197,6 +203,7 @@ public StreamObserver<AppendRowsRequest> appendRows(
         new StreamObserver<AppendRowsRequest>() {
           @Override
           public void onNext(AppendRowsRequest value) {
+            requestReceivedInstants.add(Instant.now());
             recordCount++;
             requests.add(value);
             long offset = value.getOffset().getValue();

google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/StreamWriterTest.java

@@ -54,6 +54,7 @@
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import java.io.IOException;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -86,12 +87,15 @@ public class StreamWriterTest {
   private static final String EXPLICIT_STREAM = "projects/p/datasets/d1/tables/t1/streams/s1";
   private static final String TEST_TRACE_ID = "DATAFLOW:job_id";
   private static final int MAX_RETRY_NUM_ATTEMPTS = 3;
+  private static final long INITIAL_RETRY_MILLIS = 500;
+  private static final double RETRY_MULTIPLIER = 1.3;
+  private static final int MAX_RETRY_DELAY_MINUTES = 5;
   private static final RetrySettings retrySettings =
       RetrySettings.newBuilder()
-          .setInitialRetryDelay(Duration.ofMillis(500))
-          .setRetryDelayMultiplier(1.1)
+          .setInitialRetryDelay(Duration.ofMillis(INITIAL_RETRY_MILLIS))
+          .setRetryDelayMultiplier(RETRY_MULTIPLIER)
           .setMaxAttempts(MAX_RETRY_NUM_ATTEMPTS)
-          .setMaxRetryDelay(org.threeten.bp.Duration.ofMinutes(5))
+          .setMaxRetryDelay(org.threeten.bp.Duration.ofMinutes(MAX_RETRY_DELAY_MINUTES))
           .build();
   private FakeScheduledExecutorService fakeExecutor;
   private FakeBigQueryWrite testBigQueryWrite;
@@ -2003,6 +2007,46 @@ public void testExclusiveAppendSuccessAndQuotaErrorRetryMaxRetry() throws Except
         ((StatusRuntimeException) ex.getCause()).getStatus().getCode());
   }
 
+  @Test
+  public void testExclusiveAppendQuotaErrorRetryExponentialBackoff() throws Exception {
+    testBigQueryWrite.setReturnErrorDuringExclusiveStreamRetry(true);
+    StreamWriter writer = getTestStreamWriterExclusiveRetryEnabled();
+
+    testBigQueryWrite.addResponse(
+        new DummyResponseSupplierWillFailThenSucceed(
+            new FakeBigQueryWriteImpl.Response(createAppendResponse(0)),
+            /* totalFailCount= */ MAX_RETRY_NUM_ATTEMPTS + 1,
+            com.google.rpc.Status.newBuilder().setCode(Code.RESOURCE_EXHAUSTED.ordinal()).build()));
+
+    ApiFuture<AppendRowsResponse> future =
+        writer.append(createProtoRows(new String[] {String.valueOf(0)}), 0);
+
+    ExecutionException ex =
+        assertThrows(
+            ExecutionException.class,
+            () -> {
+              future.get();
+            });
+    assertEquals(
+        Status.Code.RESOURCE_EXHAUSTED,
+        ((StatusRuntimeException) ex.getCause()).getStatus().getCode());
+
+    ArrayList<Instant> instants = testBigQueryWrite.getLatestRequestReceivedInstants();
+    Instant previousInstant = instants.get(0);
+    // Include initial attempt
+    assertEquals(instants.size(), MAX_RETRY_NUM_ATTEMPTS + 1);
+    double minExpectedDelay = INITIAL_RETRY_MILLIS * 0.95;
+    for (int i = 1; i < instants.size(); i++) {
+      Instant currentInstant = instants.get(i);
+      double differenceInMillis =
+          java.time.Duration.between(previousInstant, currentInstant).toMillis();
+      assertThat(differenceInMillis).isAtLeast((double) INITIAL_RETRY_MILLIS);
+      assertThat(differenceInMillis).isGreaterThan(minExpectedDelay);
+      minExpectedDelay = minExpectedDelay * RETRY_MULTIPLIER;
+      previousInstant = currentInstant;
+    }
+  }
+
   @Test
   public void testAppendSuccessAndNonRetryableError() throws Exception {
     StreamWriter writer = getTestStreamWriterRetryEnabled();