fix: update grpc bidi resumable uploads to validate ack'd object size

Follow up to #2527

Move request trimming inline rather than at the end

Author: BenWhitehead

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

@@ -52,7 +52,7 @@ public StartResumableWriteResponse getRes() {
 
   @Override
   public BidiWriteObjectRequest.Builder newBuilder() {
-    return writeRequest.toBuilder();
+    return writeRequest.toBuilder().clearWriteObjectSpec();
   }
 
   @Override

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

@@ -136,7 +136,8 @@ private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength, bo
     List<BidiWriteObjectRequest> messages = new ArrayList<>();
 
     int bytesConsumed = 0;
-    for (ChunkSegment datum : data) {
+    for (int i = 0; i < data.length; i++) {
+      ChunkSegment datum = data[i];
       Crc32cLengthKnown crc32c = datum.getCrc32c();
       ByteString b = datum.getB();
       int contentSize = b.size();
@@ -149,11 +150,14 @@ private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength, bo
       if (crc32c != null) {
         checksummedData.setCrc32C(crc32c.getValue());
       }
-      BidiWriteObjectRequest.Builder builder =
-          writeCtx
-              .newRequestBuilder()
-              .setWriteOffset(offset)
-              .setChecksummedData(checksummedData.build());
+      BidiWriteObjectRequest.Builder builder = writeCtx.newRequestBuilder();
+      if (!first) {
+        builder.clearUploadId();
+        builder.clearObjectChecksums();
+      } else {
+        first = false;
+      }
+      builder.setWriteOffset(offset).setChecksummedData(checksummedData.build());
       if (!datum.isOnlyFullBlocks()) {
         builder.setFinishWrite(true);
         if (cumulative != null) {
@@ -163,8 +167,11 @@ private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength, bo
         finished = true;
       }
 
-      BidiWriteObjectRequest build = possiblyPairDownBidiRequest(builder, first).build();
-      first = false;
+      if (i == data.length - 1 && !finished) {
+        builder.setFlush(true).setStateLookup(true);
+      }
+
+      BidiWriteObjectRequest build = builder.build();
       messages.add(build);
       bytesConsumed += contentSize;
     }
@@ -224,11 +231,6 @@ private void flush(@NonNull List<BidiWriteObjectRequest> segments) {
             for (BidiWriteObjectRequest message : segments) {
               opened.onNext(message);
             }
-            if (!finished) {
-              BidiWriteObjectRequest message =
-                  BidiWriteObjectRequest.newBuilder().setFlush(true).setStateLookup(true).build();
-              opened.onNext(message);
-            }
             responseObserver.await();
             return null;
           } catch (Exception e) {
@@ -240,26 +242,6 @@ private void flush(@NonNull List<BidiWriteObjectRequest> segments) {
         Decoder.identity());
   }
 
-  private static BidiWriteObjectRequest.Builder possiblyPairDownBidiRequest(
-      BidiWriteObjectRequest.Builder b, boolean firstMessageOfStream) {
-    if (firstMessageOfStream && b.getWriteOffset() == 0) {
-      return b;
-    }
-
-    if (!firstMessageOfStream) {
-      b.clearUploadId();
-    }
-
-    if (b.getWriteOffset() > 0) {
-      b.clearWriteObjectSpec();
-    }
-
-    if (b.getWriteOffset() > 0 && !b.getFinishWrite()) {
-      b.clearObjectChecksums();
-    }
-    return b;
-  }
-
   private class BidiObserver implements ApiStreamObserver<BidiWriteObjectResponse> {
 
     private final Semaphore sem;