Repository: googleapis/java-storage (java)

Issue:
allow specifying the content length for resumable uploads
**Is your feature request related to a problem? Please describe.**
The resumable upload API allows [specifying a final length of the object](https://cloud.google.com/storage/docs/json_api/v1/parameters#xuploadcontentlength) if known in a HTTP header. There is no way to pass this value into the Java GCS API for resumable uploads, though.

**Describe the solution you'd like**
A `Storage.writer` method overload that takes the final content length.


Pull request description:
feat: allow specifying an expected object size for resumable operations.
Update resumable upload failure detection to be more specific about classifying a request as SCENARIO_5

Fixes #2511



Patch:
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index a3178ac7..bf702133 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -225,14 +225,14 @@
     <dependency>
       <groupId>com.google.api.grpc</groupId>
       <artifactId>proto-google-cloud-kms-v1</artifactId>
-      <version>0.141.0</version>
+      <version>0.142.0</version>
       <scope>test</scope>
     </dependency>
 
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-kms</artifactId>
-      <version>2.50.0</version>
+      <version>2.51.0</version>
       <scope>test</scope>
     </dependency>
     <dependency>
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiBlobWriteSessionConfig.java
index 7a80fed4..e8ba2f3c 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiBlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiBlobWriteSessionConfig.java
@@ -114,7 +114,7 @@ public final class BidiBlobWriteSessionConfig extends BlobWriteSessionConfig
                       BidiWriteObjectRequest req = grpc.getBidiWriteObjectRequest(info, opts);
 
                       ApiFuture<BidiResumableWrite> startResumableWrite =
-                          grpc.startResumableWrite(grpcCallContext, req);
+                          grpc.startResumableWrite(grpcCallContext, req, opts);
                       return ResumableMedia.gapic()
                           .write()
                           .bidiByteChannel(grpc.storageClient.bidiWriteObjectCallable())
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
index 5a7ef181..9ab421bd 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
@@ -156,7 +156,7 @@ public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig
                       WriteObjectRequest req = grpc.getWriteObjectRequest(info, opts);
 
                       ApiFuture<ResumableWrite> startResumableWrite =
-                          grpc.startResumableWrite(grpcCallContext, req);
+                          grpc.startResumableWrite(grpcCallContext, req, opts);
                       return ResumableMedia.gapic()
                           .write()
                           .byteChannel(
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java
index 15278616..c3abe7b0 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java
@@ -24,6 +24,7 @@ import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.api.gax.rpc.BidiStreamingCallable;
+import com.google.api.gax.rpc.ErrorDetails;
 import com.google.api.gax.rpc.OutOfRangeException;
 import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
 import com.google.cloud.storage.Conversions.Decoder;
@@ -345,10 +346,17 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
     public void onError(Throwable t) {
       if (t instanceof OutOfRangeException) {
         OutOfRangeException oore = (OutOfRangeException) t;
-        clientDetectedError(
-            ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
-                ImmutableList.of(lastWrittenRequest), null, context, oore));
-      } else if (t instanceof ApiException) {
+        ErrorDetails ed = oore.getErrorDetails();
+        if (!(ed != null
+            && ed.getErrorInfo() != null
+            && ed.getErrorInfo().getReason().equals("GRPC_MISMATCHED_UPLOAD_SIZE"))) {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), null, context, oore));
+          return;
+        }
+      }
+      if (t instanceof ApiException) {
         // use StorageExceptions logic to translate from ApiException to our status codes ensuring
         // things fall in line with our retry handlers.
         // This is suboptimal, as it will initialize a second exception, however this is the
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java
index 6acac1e6..e7697ed7 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java
@@ -24,6 +24,7 @@ import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.api.gax.rpc.ClientStreamingCallable;
+import com.google.api.gax.rpc.ErrorDetails;
 import com.google.api.gax.rpc.OutOfRangeException;
 import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
 import com.google.cloud.storage.Conversions.Decoder;
@@ -267,11 +268,18 @@ final class GapicUnbufferedChunkedResumableWritableByteChannel
       if (t instanceof OutOfRangeException) {
         OutOfRangeException oore = (OutOfRangeException) t;
         open = false;
-        StorageException storageException =
-            ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
-                segments, null, context, oore);
-        invocationHandle.setException(storageException);
-      } else if (t instanceof ApiException) {
+        ErrorDetails ed = oore.getErrorDetails();
+        if (!(ed != null
+            && ed.getErrorInfo() != null
+            && ed.getErrorInfo().getReason().equals("GRPC_MISMATCHED_UPLOAD_SIZE"))) {
+          StorageException storageException =
+              ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
+                  segments, null, context, oore);
+          invocationHandle.setException(storageException);
+          return;
+        }
+      }
+      if (t instanceof ApiException) {
         // use StorageExceptions logic to translate from ApiException to our status codes ensuring
         // things fall in line with our retry handlers.
         // This is suboptimal, as it will initialize a second exception, however this is the
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUploadSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUploadSessionBuilder.java
index 2d1daa54..3ae9b8bc 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUploadSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUploadSessionBuilder.java
@@ -21,6 +21,8 @@ import com.google.api.core.ApiFutures;
 import com.google.api.gax.rpc.BidiStreamingCallable;
 import com.google.api.gax.rpc.ClientStreamingCallable;
 import com.google.api.gax.rpc.UnaryCallable;
+import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
+import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.storage.v2.BidiWriteObjectRequest;
 import com.google.storage.v2.BidiWriteObjectResponse;
@@ -50,7 +52,8 @@ final class GapicUploadSessionBuilder {
 
   ApiFuture<ResumableWrite> resumableWrite(
       UnaryCallable<StartResumableWriteRequest, StartResumableWriteResponse> callable,
-      WriteObjectRequest writeObjectRequest) {
+      WriteObjectRequest writeObjectRequest,
+      Opts<ObjectTargetOpt> opts) {
     StartResumableWriteRequest.Builder b = StartResumableWriteRequest.newBuilder();
     if (writeObjectRequest.hasWriteObjectSpec()) {
       b.setWriteObjectSpec(writeObjectRequest.getWriteObjectSpec());
@@ -61,7 +64,7 @@ final class GapicUploadSessionBuilder {
     if (writeObjectRequest.hasObjectChecksums()) {
       b.setObjectChecksums(writeObjectRequest.getObjectChecksums());
     }
-    StartResumableWriteRequest req = b.build();
+    StartResumableWriteRequest req = opts.startResumableWriteRequest().apply(b).build();
     Function<String, WriteObjectRequest> f =
         uploadId ->
             writeObjectRequest.toBuilder().clearWriteObjectSpec().setUploadId(uploadId).build();
@@ -80,7 +83,8 @@ final class GapicUploadSessionBuilder {
 
   ApiFuture<BidiResumableWrite> bidiResumableWrite(
       UnaryCallable<StartResumableWriteRequest, StartResumableWriteResponse> x,
-      BidiWriteObjectRequest writeObjectRequest) {
+      BidiWriteObjectRequest writeObjectRequest,
+      Opts<ObjectTargetOpt> opts) {
     StartResumableWriteRequest.Builder b = StartResumableWriteRequest.newBuilder();
     if (writeObjectRequest.hasWriteObjectSpec()) {
       b.setWriteObjectSpec(writeObjectRequest.getWriteObjectSpec());
@@ -91,7 +95,7 @@ final class GapicUploadSessionBuilder {
     if (writeObjectRequest.hasObjectChecksums()) {
       b.setObjectChecksums(writeObjectRequest.getObjectChecksums());
     }
-    StartResumableWriteRequest req = b.build();
+    StartResumableWriteRequest req = opts.startResumableWriteRequest().apply(b).build();
     Function<String, BidiWriteObjectRequest> f =
         uploadId ->
             writeObjectRequest.toBuilder().clearWriteObjectSpec().setUploadId(uploadId).build();
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index 0d0904ca..022a3cc1 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -320,7 +320,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write =
         storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext);
 
-    ApiFuture<ResumableWrite> start = startResumableWrite(grpcCallContext, req);
+    ApiFuture<ResumableWrite> start = startResumableWrite(grpcCallContext, req, opts);
     ApiFuture<GrpcResumableSession> session2 =
         ApiFutures.transform(
             start,
@@ -365,7 +365,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
 
-    ApiFuture<ResumableWrite> start = startResumableWrite(grpcCallContext, req);
+    ApiFuture<ResumableWrite> start = startResumableWrite(grpcCallContext, req, opts);
 
     BufferedWritableByteChannelSession<WriteObjectResponse> session =
         ResumableMedia.gapic()
@@ -790,7 +790,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     // in JSON, the starting of the resumable session happens before the invocation of write can
     // happen. Emulate the same thing here.
     //  1. create the future
-    ApiFuture<ResumableWrite> startResumableWrite = startResumableWrite(grpcCallContext, req);
+    ApiFuture<ResumableWrite> startResumableWrite = startResumableWrite(grpcCallContext, req, opts);
     //  2. await the result of the future
     ResumableWrite resumableWrite = ApiFutureUtils.await(startResumableWrite);
     //  3. wrap the result in another future container before constructing the BlobWriteChannel
@@ -1919,7 +1919,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
 
   @VisibleForTesting
   ApiFuture<ResumableWrite> startResumableWrite(
-      GrpcCallContext grpcCallContext, WriteObjectRequest req) {
+      GrpcCallContext grpcCallContext, WriteObjectRequest req, Opts<ObjectTargetOpt> opts) {
     Set<StatusCode.Code> codes = resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(req));
     GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return ResumableMedia.gapic()
@@ -1928,11 +1928,12 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
             storageClient
                 .startResumableWriteCallable()
                 .withDefaultCallContext(merge.withRetryableCodes(codes)),
-            req);
+            req,
+            opts);
   }
 
   ApiFuture<BidiResumableWrite> startResumableWrite(
-      GrpcCallContext grpcCallContext, BidiWriteObjectRequest req) {
+      GrpcCallContext grpcCallContext, BidiWriteObjectRequest req, Opts<ObjectTargetOpt> opts) {
     Set<StatusCode.Code> codes = resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(req));
     GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return ResumableMedia.gapic()
@@ -1941,7 +1942,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
             storageClient
                 .startResumableWriteCallable()
                 .withDefaultCallContext(merge.withRetryableCodes(codes)),
-            req);
+            req,
+            opts);
   }
 
   private SourceObject sourceObjectEncode(SourceBlob from) {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java
index 7d0e46ce..499b9dce 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java
@@ -190,7 +190,7 @@ public final class JournalingBlobWriteSessionConfig extends BlobWriteSessionConf
             opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
         ApiFuture<ResumableWrite> f =
             grpcStorage.startResumableWrite(
-                grpcCallContext, grpcStorage.getWriteObjectRequest(info, opts));
+                grpcCallContext, grpcStorage.getWriteObjectRequest(info, opts), opts);
         ApiFuture<WriteCtx<ResumableWrite>> start =
             ApiFutures.transform(f, WriteCtx::new, MoreExecutors.directExecutor());
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java
index 9d3afa5a..149c7ad9 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java
@@ -205,7 +205,8 @@ final class JsonResumableSessionPutTask
             && contentLength != null
             && contentLength > 0) {
           String errorMessage = cause.getContent().toLowerCase(Locale.US);
-          if (errorMessage.contains("content-range")) {
+          if (errorMessage.contains("content-range")
+              && !errorMessage.contains("earlier")) { // TODO: exclude "earlier request"
             StorageException se =
                 ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
                     uploadId, response, cause, cause::getContent);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannel.java
index 58e9bad0..db807ce6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannel.java
@@ -43,6 +43,7 @@ import com.google.cloud.storage.UnifiedOpts.MetagenerationNotMatch;
 import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
+import com.google.cloud.storage.UnifiedOpts.ResumableUploadExpectedObjectSize;
 import com.google.cloud.storage.UnifiedOpts.SourceGenerationMatch;
 import com.google.cloud.storage.UnifiedOpts.SourceGenerationNotMatch;
 import com.google.cloud.storage.UnifiedOpts.SourceMetagenerationMatch;
@@ -102,7 +103,8 @@ final class ParallelCompositeUploadWritableByteChannel implements BufferedWritab
                 || o instanceof SourceMetagenerationMatch
                 || o instanceof SourceMetagenerationNotMatch
                 || o instanceof Crc32cMatch
-                || o instanceof Md5Match;
+                || o instanceof Md5Match
+                || o instanceof ResumableUploadExpectedObjectSize;
     TO_EXCLUDE_FROM_PARTS = tmp.negate();
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
index 89624d9b..70604593 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
@@ -1352,6 +1352,23 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new BlobWriteOption(UnifiedOpts.detectContentType());
     }
 
+    /**
+     * Set a precondition on the number of bytes that GCS should expect for a resumable upload. See
+     * the docs for <a
+     * href="https://cloud.google.com/storage/docs/json_api/v1/parameters#xuploadcontentlength">X-Upload-Content-Length</a>
+     * for more detail.
+     *
+     * <p>If the method invoked with this option does not perform a resumable upload, this option
+     * will be ignored.
+     *
+     * @since 2.42.0
+     */
+    @BetaApi
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static BlobWriteOption expectedObjectSize(long objectContentSize) {
+      return new BlobWriteOption(UnifiedOpts.resumableUploadExpectedObjectSize(objectContentSize));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
index 15476cfc..bcc987ea 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
@@ -55,6 +55,7 @@ import com.google.storage.v2.LockBucketRetentionPolicyRequest;
 import com.google.storage.v2.ReadObjectRequest;
 import com.google.storage.v2.RestoreObjectRequest;
 import com.google.storage.v2.RewriteObjectRequest;
+import com.google.storage.v2.StartResumableWriteRequest;
 import com.google.storage.v2.UpdateBucketRequest;
 import com.google.storage.v2.UpdateHmacKeyRequest;
 import com.google.storage.v2.UpdateObjectRequest;
@@ -196,6 +197,10 @@ final class UnifiedOpts {
     default Mapper<RewriteObjectRequest.Builder> rewriteObject() {
       return Mapper.identity();
     }
+
+    default Mapper<StartResumableWriteRequest.Builder> startResumableWrite() {
+      return Mapper.identity();
+    }
   }
 
   /** Base interface for those Opts which are applicable to Bucket List operations */
@@ -487,6 +492,12 @@ final class UnifiedOpts {
     return new Projection(projection);
   }
 
+  static ResumableUploadExpectedObjectSize resumableUploadExpectedObjectSize(
+      long expectedObjectSize) {
+    checkArgument(expectedObjectSize >= 0, "expectedObjectSize >= 0 (%s >= 0)", expectedObjectSize);
+    return new ResumableUploadExpectedObjectSize(expectedObjectSize);
+  }
+
   static SoftDeleted softDeleted(boolean softDeleted) {
     return new SoftDeleted(softDeleted);
   }
@@ -1832,6 +1843,25 @@ final class UnifiedOpts {
     }
   }
 
+  static final class ResumableUploadExpectedObjectSize extends RpcOptVal<@NonNull Long>
+      implements ObjectTargetOpt {
+    private static final long serialVersionUID = 3640126281492196357L;
+
+    private ResumableUploadExpectedObjectSize(@NonNull Long val) {
+      super(StorageRpc.Option.X_UPLOAD_CONTENT_LENGTH, val);
+    }
+
+    @Override
+    public Mapper<StartResumableWriteRequest.Builder> startResumableWrite() {
+      return b -> {
+        if (val > 0) {
+          b.getWriteObjectSpecBuilder().setObjectSize(val);
+        }
+        return b;
+      };
+    }
+  }
+
   static final class ShowDeletedKeys extends RpcOptVal<@NonNull Boolean> implements HmacKeyListOpt {
     private static final long serialVersionUID = -6604176744362903487L;
 
@@ -2426,6 +2456,10 @@ final class UnifiedOpts {
       return fuseMappers(ObjectTargetOpt.class, ObjectTargetOpt::bidiWriteObject);
     }
 
+    Mapper<StartResumableWriteRequest.Builder> startResumableWriteRequest() {
+      return fuseMappers(ObjectTargetOpt.class, ObjectTargetOpt::startResumableWrite);
+    }
+
     Mapper<GetObjectRequest.Builder> getObjectsRequest() {
       return fuseMappers(ObjectSourceOpt.class, ObjectSourceOpt::getObject);
     }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
index 89a09867..5341051a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
@@ -1090,6 +1090,10 @@ public class HttpStorageRpc implements StorageRpc {
           requestFactory.buildPostRequest(url, new JsonHttpContent(jsonFactory, object));
       HttpHeaders requestHeaders = httpRequest.getHeaders();
       requestHeaders.set("X-Upload-Content-Type", detectContentType(object, options));
+      Long xUploadContentLength = Option.X_UPLOAD_CONTENT_LENGTH.getLong(options);
+      if (xUploadContentLength != null) {
+        requestHeaders.set("X-Upload-Content-Length", xUploadContentLength);
+      }
       setEncryptionHeaders(requestHeaders, "x-goog-encryption-", options);
       HttpResponse response = httpRequest.execute();
       if (response.getStatusCode() != 200) {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java
index d4e0abbf..78747d42 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java
@@ -77,7 +77,8 @@ public interface StorageRpc extends ServiceRpc {
     SOFT_DELETED("softDeleted"),
     COPY_SOURCE_ACL("copySourceAcl"),
     GENERATION("generation"),
-    INCLUDE_FOLDERS_AS_PREFIXES("includeFoldersAsPrefixes");
+    INCLUDE_FOLDERS_AS_PREFIXES("includeFoldersAsPrefixes"),
+    X_UPLOAD_CONTENT_LENGTH("x-upload-content-length");
 
     private final String value;
 


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUploadSessionBuilderSyntaxTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUploadSessionBuilderSyntaxTest.java
index a8afe473..e35278b5 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUploadSessionBuilderSyntaxTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUploadSessionBuilderSyntaxTest.java
@@ -23,6 +23,7 @@ import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
 import com.google.api.gax.rpc.ClientStreamingCallable;
 import com.google.api.gax.rpc.UnaryCallable;
+import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.storage.v2.StartResumableWriteRequest;
 import com.google.storage.v2.StartResumableWriteResponse;
 import com.google.storage.v2.WriteObjectRequest;
@@ -94,7 +95,7 @@ public final class GapicUploadSessionBuilderSyntaxTest {
   @Test
   public void syntax_resumableUnbuffered_fluent() {
     ApiFuture<ResumableWrite> startAsync =
-        ResumableMedia.gapic().write().resumableWrite(startResumableWrite, req);
+        ResumableMedia.gapic().write().resumableWrite(startResumableWrite, req, Opts.empty());
     UnbufferedWritableByteChannelSession<WriteObjectResponse> session =
         ResumableMedia.gapic()
             .write()
@@ -110,7 +111,7 @@ public final class GapicUploadSessionBuilderSyntaxTest {
   @Test
   public void syntax_resumableBuffered_fluent() {
     ApiFuture<ResumableWrite> startAsync =
-        ResumableMedia.gapic().write().resumableWrite(startResumableWrite, req);
+        ResumableMedia.gapic().write().resumableWrite(startResumableWrite, req, Opts.empty());
     BufferedWritableByteChannelSession<WriteObjectResponse> session =
         ResumableMedia.gapic()
             .write()
@@ -150,7 +151,7 @@ public final class GapicUploadSessionBuilderSyntaxTest {
   @Test
   public void syntax_resumableUnbuffered_incremental() {
     ApiFuture<ResumableWrite> startAsync =
-        ResumableMedia.gapic().write().resumableWrite(startResumableWrite, req);
+        ResumableMedia.gapic().write().resumableWrite(startResumableWrite, req, Opts.empty());
     GapicWritableByteChannelSessionBuilder b1 =
         ResumableMedia.gapic()
             .write()
@@ -164,7 +165,7 @@ public final class GapicUploadSessionBuilderSyntaxTest {
   @Test
   public void syntax_resumableBuffered_incremental() {
     ApiFuture<ResumableWrite> startAsync =
-        ResumableMedia.gapic().write().resumableWrite(startResumableWrite, req);
+        ResumableMedia.gapic().write().resumableWrite(startResumableWrite, req, Opts.empty());
     GapicWritableByteChannelSessionBuilder b1 =
         ResumableMedia.gapic()
             .write()
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITSyncAndUploadUnbufferedWritableByteChannelPropertyTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITSyncAndUploadUnbufferedWritableByteChannelPropertyTest.java
index 240ae1d5..5055c1e2 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITSyncAndUploadUnbufferedWritableByteChannelPropertyTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITSyncAndUploadUnbufferedWritableByteChannelPropertyTest.java
@@ -310,7 +310,9 @@ public class ITSyncAndUploadUnbufferedWritableByteChannelPropertyTest {
 
       ApiFuture<ResumableWrite> f =
           storage.startResumableWrite(
-              GrpcCallContext.createDefault(), storage.getWriteObjectRequest(info, Opts.empty()));
+              GrpcCallContext.createDefault(),
+              storage.getWriteObjectRequest(info, Opts.empty()),
+              Opts.empty());
       ResumableWrite resumableWrite = ApiExceptions.callAndTranslateApiException(f);
 
       UploadCtx uploadCtx =
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITUnbufferedResumableUploadTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITUnbufferedResumableUploadTest.java
index 34b600d7..8dce2090 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITUnbufferedResumableUploadTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITUnbufferedResumableUploadTest.java
@@ -161,7 +161,9 @@ public final class ITUnbufferedResumableUploadTest {
         ResumableMedia.gapic()
             .write()
             .resumableWrite(
-                storageClient.startResumableWriteCallable().withDefaultCallContext(merge), request);
+                storageClient.startResumableWriteCallable().withDefaultCallContext(merge),
+                request,
+                opts);
 
     UnbufferedWritableByteChannelSession<WriteObjectResponse> session =
         ResumableMedia.gapic()
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITResumableUploadTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITResumableUploadTest.java
new file mode 100644
index 00000000..cb0524f7
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITResumableUploadTest.java
@@ -0,0 +1,229 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.it;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BlobWriteSession;
+import com.google.cloud.storage.BlobWriteSessionConfigs;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.DataGenerator;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import com.google.cloud.storage.TmpFile;
+import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.cloud.storage.it.runner.StorageITRunner;
+import com.google.cloud.storage.it.runner.annotations.Backend;
+import com.google.cloud.storage.it.runner.annotations.CrossRun;
+import com.google.cloud.storage.it.runner.annotations.Inject;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.WritableByteChannel;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@CrossRun(
+    backends = {Backend.PROD},
+    transports = {Transport.HTTP, Transport.GRPC})
+public final class ITResumableUploadTest {
+  @Rule public final TemporaryFolder temp = new TemporaryFolder();
+
+  @Inject public Storage storage;
+  @Inject public BucketInfo bucket;
+  @Inject public Generator generator;
+
+  @Test
+  public void expectedUploadSize_chunked_doesMatch()
+      throws IOException, ExecutionException, InterruptedException, TimeoutException {
+    doTestDoesMatch(storage);
+  }
+
+  @Test
+  public void expectedUploadSize_chunked_doesNotMatch() throws IOException {
+    doTestDoesNotMatch(storage);
+  }
+
+  @Test
+  @CrossRun.Exclude(transports = Transport.HTTP)
+  public void expectedUploadSize_bidi_doesMatch() throws Exception {
+    StorageOptions options =
+        storage
+            .getOptions()
+            .toBuilder()
+            .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bidiWrite())
+            .build();
+    try (Storage storage = options.getService()) {
+      doTestDoesMatch(storage);
+    }
+  }
+
+  @Test
+  @CrossRun.Exclude(transports = Transport.HTTP)
+  public void expectedUploadSize_bidi_doesNotMatch() throws Exception {
+    StorageOptions options =
+        storage
+            .getOptions()
+            .toBuilder()
+            .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bidiWrite())
+            .build();
+    try (Storage storage = options.getService()) {
+      doTestDoesNotMatch(storage);
+    }
+  }
+
+  @Test
+  public void expectedUploadSize_ignored_pcu() throws Exception {
+    StorageOptions options =
+        storage
+            .getOptions()
+            .toBuilder()
+            .setBlobWriteSessionConfig(BlobWriteSessionConfigs.parallelCompositeUpload())
+            .build();
+    try (Storage storage = options.getService()) {
+      int objectContentSize = 10;
+      byte[] bytes = DataGenerator.base64Characters().genBytes(objectContentSize);
+      BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+      BlobWriteSession session =
+          storage.blobWriteSession(info, BlobWriteOption.expectedObjectSize(objectContentSize + 1));
+      try (WritableByteChannel open = session.open()) {
+        open.write(ByteBuffer.wrap(bytes));
+      }
+
+      BlobInfo gen1 = session.getResult().get(5, TimeUnit.SECONDS);
+      assertThat(gen1.getSize()).isEqualTo(objectContentSize);
+    }
+  }
+
+  @Test
+  public void expectedUploadSize_createFrom_inputStream_doesMatch() throws Exception {
+    StorageOptions options =
+        storage
+            .getOptions()
+            .toBuilder()
+            .setBlobWriteSessionConfig(BlobWriteSessionConfigs.parallelCompositeUpload())
+            .build();
+    try (Storage storage = options.getService()) {
+      int objectContentSize = 10;
+      byte[] bytes = DataGenerator.base64Characters().genBytes(objectContentSize);
+      BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+      BlobInfo gen1 =
+          storage.createFrom(
+              info,
+              new ByteArrayInputStream(bytes),
+              BlobWriteOption.expectedObjectSize(objectContentSize));
+      assertThat(gen1.getSize()).isEqualTo(objectContentSize);
+    }
+  }
+
+  @Test
+  public void expectedUploadSize_createFrom_inputStream_doesNotMatch() throws Exception {
+    StorageOptions options =
+        storage
+            .getOptions()
+            .toBuilder()
+            .setBlobWriteSessionConfig(BlobWriteSessionConfigs.parallelCompositeUpload())
+            .build();
+    try (Storage storage = options.getService()) {
+      int objectContentSize = 10;
+      byte[] bytes = DataGenerator.base64Characters().genBytes(objectContentSize);
+      BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+      StorageException se =
+          assertThrows(
+              StorageException.class,
+              () ->
+                  storage.createFrom(
+                      info,
+                      new ByteArrayInputStream(bytes),
+                      BlobWriteOption.expectedObjectSize(objectContentSize + 1)));
+
+      assertThat(se.getCode()).isEqualTo(400);
+    }
+  }
+
+  @Test
+  public void expectedUploadSize_createFrom_path_doesMatch() throws IOException {
+    int objectContentSize = 10;
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    try (TmpFile tmpFile =
+        DataGenerator.base64Characters().tempFile(temp.getRoot().toPath(), objectContentSize)) {
+      BlobInfo gen1 =
+          storage.createFrom(
+              info, tmpFile.getPath(), BlobWriteOption.expectedObjectSize(objectContentSize));
+      assertThat(gen1.getSize()).isEqualTo(objectContentSize);
+    }
+  }
+
+  @Test
+  public void expectedUploadSize_createFrom_path_doesNotMatch() throws IOException {
+    int objectContentSize = 10;
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    try (TmpFile tmpFile =
+        DataGenerator.base64Characters().tempFile(temp.getRoot().toPath(), objectContentSize)) {
+      StorageException se =
+          assertThrows(
+              StorageException.class,
+              () ->
+                  storage.createFrom(
+                      info,
+                      tmpFile.getPath(),
+                      BlobWriteOption.expectedObjectSize(objectContentSize + 1)));
+
+      assertThat(se.getCode()).isEqualTo(400);
+    }
+  }
+
+  private void doTestDoesMatch(Storage storage)
+      throws IOException, InterruptedException, ExecutionException, TimeoutException {
+    int objectContentSize = 10;
+    byte[] bytes = DataGenerator.base64Characters().genBytes(objectContentSize);
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    BlobWriteSession session =
+        storage.blobWriteSession(info, BlobWriteOption.expectedObjectSize(objectContentSize));
+    try (WritableByteChannel open = session.open()) {
+      open.write(ByteBuffer.wrap(bytes));
+    }
+
+    BlobInfo gen1 = session.getResult().get(5, TimeUnit.SECONDS);
+    assertThat(gen1.getSize()).isEqualTo(objectContentSize);
+  }
+
+  private void doTestDoesNotMatch(Storage storage) throws IOException {
+    int objectContentSize = 10;
+    byte[] bytes = DataGenerator.base64Characters().genBytes(objectContentSize);
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    BlobWriteSession session =
+        storage.blobWriteSession(info, BlobWriteOption.expectedObjectSize(objectContentSize + 1));
+
+    WritableByteChannel open = session.open();
+    open.write(ByteBuffer.wrap(bytes));
+    StorageException se = assertThrows(StorageException.class, open::close);
+
+    assertThat(se.getCode()).isEqualTo(400);
+  }
+}