fix(storage): BidiAppendableUpload Takeover operation fixes (#13776)

BidiAppendableUpload Takeover operation fixes

Co-authored-by: Dhriti Chopra <dhritichopra@google.com>

Author: Dhriti07

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiUploadState.java

@@ -544,6 +544,7 @@ final void updateStateFromResponse(BidiWriteObjectResponse response) {
         if (state == State.INITIALIZING || state == State.RETRYING) {
           transitionTo(
               stateToReturnToAfterRetry != null ? stateToReturnToAfterRetry : State.RUNNING);
+          stateToReturnToAfterRetry = null;
         }
 
         boolean signalTerminalSuccess = false;
@@ -679,7 +680,9 @@ final void pendingRetry() {
       lock.lock();
       try {
         validateCurrentStateIsOneOf(State.allNonTerminal);
-        stateToReturnToAfterRetry = state;
+        if (state != State.RETRYING && state != State.PENDING_RETRY) {
+          stateToReturnToAfterRetry = state;
+        }
         transitionTo(State.PENDING_RETRY);
       } finally {
         lock.unlock();
@@ -887,10 +890,11 @@ void awaitState(State... anyOf) throws InterruptedException {
       lock.lock();
       try {
         ImmutableSet<State> states = ImmutableSet.copyOf(anyOf);
-        while (!states.contains(this.state) && !stateUpdated.await(5, TimeUnit.MILLISECONDS)) {
+        while (!states.contains(this.state)) {
           if (resultFuture.isDone()) {
             return;
           }
+          stateUpdated.await(5, TimeUnit.MILLISECONDS);
         }
       } finally {
         lock.unlock();
@@ -913,11 +917,11 @@ public void awaitTakeoverStateReconciliation(Runnable restart) {
     public void awaitAck(long writeOffset) throws InterruptedException {
       lock.lock();
       try {
-        while (confirmedBytes < writeOffset
-            && !confirmedBytesUpdated.await(5, TimeUnit.MILLISECONDS)) {
+        while (confirmedBytes < writeOffset) {
           if (resultFuture.isDone()) {
             return;
           }
+          confirmedBytesUpdated.await(5, TimeUnit.MILLISECONDS);
         }
       } finally {
         lock.unlock();

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadTest.java

@@ -58,7 +58,7 @@
 
 @RunWith(StorageITRunner.class)
 @CrossRun(
-    backends = {Backend.TEST_BENCH},
+    backends = {Backend.TEST_BENCH, Backend.PROD},
     transports = Transport.GRPC)
 @Parameterized(UploadConfigParameters.class)
 public final class ITAppendableUploadTest {

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java

@@ -259,10 +259,10 @@ static BackendResources of(
               protectedBucketNames.add(bucketName);
               return new BucketInfoShim(
                   BucketInfo.newBuilder(bucketName)
-                      .setLocation(zone.get().get().getRegion())
+                      .setLocation("us-central1")
                       .setCustomPlacementConfig(
                           CustomPlacementConfig.newBuilder()
-                              .setDataLocations(ImmutableList.of(zone.get().get().getZone()))
+                              .setDataLocations(ImmutableList.of("us-central1-c"))
                               .build())
                       .setStorageClass(StorageClass.valueOf("RAPID"))
                       .setHierarchicalNamespace(