fix(storage): fix flaky storage integration tests

Author: nidhiii-27

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

@@ -165,6 +165,9 @@ public void flush() {
   public boolean finishWrite(long length) {
     lock.lock();
     try {
+      if (state.getState() == State.TERMINAL_SUCCESS) {
+        return true;
+      }
       // if we're already finalizing, ack rather than enqueueing again
       if (state.isFinalizing() && state.getTotalSentBytes() == length) {
         return true;
@@ -192,6 +195,10 @@ public boolean closeStream(long length) {
     lock.lock();
     try {
 
+      if (state.getState() == State.TERMINAL_SUCCESS) {
+        return true;
+      }
+
       boolean offer = state.finalFlush(length);
       if (offer) {
         internalSend();

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

@@ -1356,6 +1356,11 @@ boolean isFinalizing() {
               return false;
             }
 
+            @Override
+            State getState() {
+              return State.RUNNING;
+            }
+
             @Override
             long getTotalSentBytes() {
               return 0;

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

@@ -380,12 +380,9 @@ static class DirectWriteService extends StorageImplBase {
     private static final Logger LOGGER = LoggerFactory.getLogger(DirectWriteService.class);
     private final BiConsumer<StreamObserver<WriteObjectResponse>, List<WriteObjectRequest>> c;
 
-    private ImmutableList.Builder<WriteObjectRequest> requests;
-
     DirectWriteService(
         BiConsumer<StreamObserver<WriteObjectResponse>, List<WriteObjectRequest>> c) {
       this.c = c;
-      this.requests = new ImmutableList.Builder<>();
     }
 
     DirectWriteService(ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> writes) {
@@ -420,6 +417,9 @@ private static void logUnexpectedRequest(
     @Override
     public StreamObserver<WriteObjectRequest> writeObject(StreamObserver<WriteObjectResponse> obs) {
       return new Adapter() {
+        private final ImmutableList.Builder<WriteObjectRequest> requests =
+            new ImmutableList.Builder<>();
+
         @Override
         public void onNext(WriteObjectRequest value) {
           requests.add(value);
@@ -432,7 +432,6 @@ public void onError(Throwable t) {}
         public void onCompleted() {
           ImmutableList<WriteObjectRequest> build = requests.build();
           c.accept(obs, build);
-          requests = new ImmutableList.Builder<>();
         }
       };
     }