Repository: googleapis/java-storage
Pull request #2941 (2025-02-19 19:35:39)

feat: add new Options to allow per method header values
Add new "Option"s for those methods which already have option types to allow providing an ImmutableMap<String, String> to be applied as extra headers to all requests sent as part of that operation.

If an operation has multiple sources of input Options (rewrite) the "first" (i.e. source option) will be the one added to the request.

The following resources do not have "Option"s and therefor do not have extra headers support at this time:
* Acl
* DefaultAcl
* ServiceAccount
* Notification

### Overview of the changes

This blind number of lines is large, but the number of differences is actually fairly low.

* Add `Headers` Opt` in UnifiedOpts, this is the primary guts of the new feature and is responsible for holding onto the extra header values and validating the provided headers to ensure no collisions with headers otherwise set by our library/restricted by gcs.
* The ~800 lines of change in Storage.java are the same "add new option `extraHeaders`" to each of the 15 option classes along with accompanying javadocs. All javadocs are the exact same, and were copy-pasted for each method.
* Create new utility method `Utils#headerNameToLowerCase` to centralize the logic for transforming header names across all locations
* Add new `StorageRpc.Options#EXTRA_HEADERS` to support extra headers for JSON calls
  * update JSON request building code in `HttpStorageRpc` (433 lines), `ApiaryUnbufferedReadableByteChannel` (13 lines) and `JsonResumable*` (~45 lines + ~50 lines for existing tests) classes to take `EXTRA_HEADERS` into consideration
* test: add new suite of tests `ITExtraHeadersOptionTest` to validate headers are properly added to each request of various operations
* test: introduce new interface `AssertRequestHeaders` which both `GrpcRequestAuditing` and `RequestAuditing` now implement -- this interface makes it easy for each transport to share common methods for asserting request headers



Patch:
diff --git a/README.md b/README.md
index 6ca40a1a..87185f7c 100644
--- a/README.md
+++ b/README.md
@@ -19,7 +19,7 @@ If you are using Maven with [BOM][libraries-bom], add this to your pom.xml file:
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>libraries-bom</artifactId>
-      <version>26.54.0</version>
+      <version>26.55.0</version>
       <type>pom</type>
       <scope>import</scope>
     </dependency>
@@ -59,7 +59,7 @@ If you are using Maven without the BOM, add this to your dependencies:
 If you are using Gradle 5.x or later, add this to your dependencies:
 
 ```Groovy
-implementation platform('com.google.cloud:libraries-bom:26.54.0')
+implementation platform('com.google.cloud:libraries-bom:26.55.0')
 
 implementation 'com.google.cloud:google-cloud-storage'
 ```
diff --git a/generation_config.yaml b/generation_config.yaml
index 0268d85f..1a742391 100644
--- a/generation_config.yaml
+++ b/generation_config.yaml
@@ -1,6 +1,6 @@
 gapic_generator_version: 2.53.0
-googleapis_commitish: a003cab30e2a263e16e9252256041f8934f40e2c
-libraries_bom_version: 26.54.0
+googleapis_commitish: 6bc8e91bf92cc985da5ed0c227b48f12315cb695
+libraries_bom_version: 26.55.0
 libraries:
   - api_shortname: storage
     name_pretty: Cloud Storage
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index ec307a7d..c37a0008 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -334,6 +334,18 @@
       <version>4.8.179</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>jul-to-slf4j</artifactId>
+      <version>2.0.16</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>ch.qos.logback</groupId>
+      <artifactId>logback-classic</artifactId>
+      <version>1.3.15</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>
@@ -386,6 +398,22 @@
               -->
               <dependency>io.opentelemetry:opentelemetry-sdk-extension-autoconfigure-spi</dependency>
               <dependency>io.opentelemetry.semconv:opentelemetry-semconv</dependency>
+              <!--
+              com.google.cloud.opentelemetry:exporter-metrics depends on slf4j
+              We are also using slf4j and logback in our test scope
+              list things out here to appease the dependency checker that gets confused like the following
+
+              [ERROR] Used undeclared dependencies found:
+              [ERROR]    org.slf4j:slf4j-api:jar:2.0.16:compile
+              [ERROR] Unused declared dependencies found:
+              [ERROR]    org.slf4j:jul-to-slf4j:jar:2.0.16:test
+              [ERROR]    ch.qos.logback:logback-classic:jar:1.3.15:test
+              [ERROR] Non-test scoped test only dependencies found:
+              [ERROR]    org.slf4j:slf4j-api:jar:2.0.16:compile
+              -->
+              <dependency>org.slf4j:slf4j-api</dependency>
+              <dependency>org.slf4j:jul-to-slf4j</dependency>
+              <dependency>ch.qos.logback:logback-classic</dependency>
             </ignoredDependencies>
           </configuration>
         </plugin>
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedReadableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedReadableByteChannel.java
index f61814d4..86afe7a9 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedReadableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedReadableByteChannel.java
@@ -33,6 +33,7 @@ import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedR
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.hash.HashFunction;
 import com.google.common.hash.Hashing;
 import com.google.common.io.BaseEncoding;
@@ -51,8 +52,8 @@ import java.nio.channels.ClosedChannelException;
 import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.ScatteringByteChannel;
 import java.util.List;
-import java.util.Locale;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.function.Function;
 import javax.annotation.concurrent.Immutable;
 import org.checkerframework.checker.nullness.qual.NonNull;
@@ -273,6 +274,14 @@ class ApiaryUnbufferedReadableByteChannel implements UnbufferedReadableByteChann
               "x-goog-encryption-key-sha256",
               base64.encode(hashFunction.hashBytes(base64.decode(key)).asBytes()));
         });
+    ifNonNull(
+        options.get(StorageRpc.Option.EXTRA_HEADERS),
+        ApiaryUnbufferedReadableByteChannel::cast,
+        (ImmutableMap<String, String> extraHeaders) -> {
+          for (Entry<String, String> e : extraHeaders.entrySet()) {
+            headers.set(e.getKey(), e.getValue());
+          }
+        });
 
     // gzip handling is performed upstream of here. Ensure we always get the raw input stream from
     // the request
@@ -302,7 +311,7 @@ class ApiaryUnbufferedReadableByteChannel implements UnbufferedReadableByteChann
       if (list.isEmpty()) {
         return null;
       } else {
-        return list.get(0).trim().toLowerCase(Locale.ENGLISH);
+        return Utils.headerNameToLowerCase(list.get(0).trim());
       }
     } else if (o instanceof String) {
       return (String) o;
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 f8221308..3211210f 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
@@ -650,8 +650,10 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     Opts<ObjectTargetOpt> dstOpts =
         Opts.unwrap(copyRequest.getTargetOptions()).resolveFrom(dst).prepend(defaultOpts);
 
-    Mapper<RewriteObjectRequest.Builder> mapper =
+    Mapper<RewriteObjectRequest.Builder> requestBuilderMapper =
         srcOpts.rewriteObjectsRequest().andThen(dstOpts.rewriteObjectsRequest());
+    Mapper<GrpcCallContext> grpcCallContextMapper =
+        srcOpts.grpcMetadataMapper().andThen(dstOpts.grpcMetadataMapper());
 
     Object srcProto = codecs.blobId().encode(src);
     Object dstProto = codecs.blobInfo().encode(dst);
@@ -686,9 +688,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
       b.setMaxBytesRewrittenPerCall(copyRequest.getMegabytesCopiedPerChunk() * _1MiB);
     }
 
-    RewriteObjectRequest req = mapper.apply(b).build();
-    GrpcCallContext grpcCallContext =
-        srcOpts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
+    RewriteObjectRequest req = requestBuilderMapper.apply(b).build();
+    GrpcCallContext grpcCallContext = grpcCallContextMapper.apply(GrpcCallContext.createDefault());
     UnaryCallable<RewriteObjectRequest, RewriteResponse> callable =
         storageClient.rewriteObjectCallable().withDefaultCallContext(grpcCallContext);
     GrpcCallContext retryContext = Retrying.newCallContext();
@@ -733,7 +734,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
   public GrpcBlobReadChannel reader(BlobId blob, BlobSourceOption... options) {
     Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob).prepend(defaultOpts);
     ReadObjectRequest request = getReadObjectRequest(blob, opts);
-    GrpcCallContext grpcCallContext = Retrying.newCallContext();
+    GrpcCallContext grpcCallContext = opts.grpcMetadataMapper().apply(Retrying.newCallContext());
 
     return new GrpcBlobReadChannel(
         storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext),
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
index 279418c5..43e575a2 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
@@ -96,7 +96,6 @@ import java.util.Collections;
 import java.util.IdentityHashMap;
 import java.util.Iterator;
 import java.util.List;
-import java.util.Locale;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
@@ -161,6 +160,11 @@ public final class GrpcStorageOptions extends StorageOptions
     return resolveSettingsAndOpts().x();
   }
 
+  @InternalApi
+  GrpcInterceptorProvider getGrpcInterceptorProvider() {
+    return grpcInterceptorProvider;
+  }
+
   private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
     in.defaultReadObject();
     this.openTelemetry = HttpStorageOptions.getDefaultInstance().getOpenTelemetry();
@@ -225,7 +229,7 @@ public final class GrpcStorageOptions extends StorageOptions
           Map<String, List<String>> requestMetadata = credentials.getRequestMetadata(uri);
           for (Entry<String, List<String>> e : requestMetadata.entrySet()) {
             String key = e.getKey();
-            if ("x-goog-user-project".equals(key.trim().toLowerCase(Locale.ENGLISH))) {
+            if ("x-goog-user-project".equals(Utils.headerNameToLowerCase(key.trim()))) {
               List<String> value = e.getValue();
               if (!value.isEmpty()) {
                 foundQuotaProject = true;
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSession.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSession.java
index d997400d..e29094fd 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSession.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSession.java
@@ -54,14 +54,13 @@ final class JsonResumableSession {
    * have the concept of nested retry handling.
    */
   ResumableOperationResult<@Nullable StorageObject> query() {
-    return new JsonResumableSessionQueryTask(context, resumableWrite.getUploadId()).call();
+    return new JsonResumableSessionQueryTask(context, resumableWrite).call();
   }
 
   ResumableOperationResult<@Nullable StorageObject> put(
       RewindableContent content, HttpContentRange contentRange) {
     JsonResumableSessionPutTask task =
-        new JsonResumableSessionPutTask(
-            context, resumableWrite.getUploadId(), content, contentRange);
+        new JsonResumableSessionPutTask(context, resumableWrite, content, contentRange);
     HttpRpcContext httpRpcContext = HttpRpcContext.getInstance();
     try {
       httpRpcContext.newInvocationId();
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 149c7ad9..9ebd8e58 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
@@ -17,6 +17,7 @@
 package com.google.cloud.storage;
 
 import com.google.api.client.http.GenericUrl;
+import com.google.api.client.http.HttpHeaders;
 import com.google.api.client.http.HttpRequest;
 import com.google.api.client.http.HttpResponse;
 import com.google.api.client.http.HttpResponseException;
@@ -31,6 +32,7 @@ import io.opencensus.trace.Status;
 import java.io.IOException;
 import java.math.BigInteger;
 import java.util.Locale;
+import java.util.Map.Entry;
 import java.util.concurrent.Callable;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
@@ -38,7 +40,7 @@ final class JsonResumableSessionPutTask
     implements Callable<ResumableOperationResult<@Nullable StorageObject>> {
 
   private final HttpClientContext context;
-  private final String uploadId;
+  private final JsonResumableWrite jsonResumableWrite;
   private final RewindableContent content;
   private final HttpContentRange originalContentRange;
 
@@ -47,11 +49,11 @@ final class JsonResumableSessionPutTask
   @VisibleForTesting
   JsonResumableSessionPutTask(
       HttpClientContext httpClientContext,
-      String uploadId,
+      JsonResumableWrite jsonResumableWrite,
       RewindableContent content,
       HttpContentRange originalContentRange) {
     this.context = httpClientContext;
-    this.uploadId = uploadId;
+    this.jsonResumableWrite = jsonResumableWrite;
     this.content = content;
     this.originalContentRange = originalContentRange;
     this.contentRange = originalContentRange;
@@ -87,13 +89,18 @@ final class JsonResumableSessionPutTask
     boolean success = false;
     boolean finalizing = originalContentRange.isFinalizing();
 
+    String uploadId = jsonResumableWrite.getUploadId();
     HttpRequest req =
         context
             .getRequestFactory()
             .buildPutRequest(new GenericUrl(uploadId), content)
             .setParser(context.getObjectParser());
     req.setThrowExceptionOnExecuteError(false);
-    req.getHeaders().setContentRange(contentRange.getHeaderValue());
+    HttpHeaders headers = req.getHeaders();
+    headers.setContentRange(contentRange.getHeaderValue());
+    for (Entry<String, String> e : jsonResumableWrite.getExtraHeaders().entrySet()) {
+      headers.set(e.getKey(), e.getValue());
+    }
 
     HttpResponse response = null;
     try {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java
index 57c5868c..dba12171 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java
@@ -20,6 +20,7 @@ import static com.google.cloud.storage.HttpClientContext.firstHeaderValue;
 
 import com.google.api.client.http.EmptyContent;
 import com.google.api.client.http.GenericUrl;
+import com.google.api.client.http.HttpHeaders;
 import com.google.api.client.http.HttpRequest;
 import com.google.api.client.http.HttpResponse;
 import com.google.api.client.http.HttpResponseException;
@@ -27,6 +28,7 @@ import com.google.api.services.storage.model.StorageObject;
 import java.io.IOException;
 import java.math.BigInteger;
 import java.util.Locale;
+import java.util.Map.Entry;
 import java.util.concurrent.Callable;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
@@ -34,15 +36,16 @@ final class JsonResumableSessionQueryTask
     implements Callable<ResumableOperationResult<@Nullable StorageObject>> {
 
   private final HttpClientContext context;
-  private final String uploadId;
+  private final JsonResumableWrite jsonResumableWrite;
 
-  JsonResumableSessionQueryTask(HttpClientContext context, String uploadId) {
+  JsonResumableSessionQueryTask(HttpClientContext context, JsonResumableWrite jsonResumableWrite) {
     this.context = context;
-    this.uploadId = uploadId;
+    this.jsonResumableWrite = jsonResumableWrite;
   }
 
   public ResumableOperationResult<@Nullable StorageObject> call() {
     HttpResponse response = null;
+    String uploadId = jsonResumableWrite.getUploadId();
     try {
       HttpRequest req =
           context
@@ -50,7 +53,11 @@ final class JsonResumableSessionQueryTask
               .buildPutRequest(new GenericUrl(uploadId), new EmptyContent())
               .setParser(context.getObjectParser());
       req.setThrowExceptionOnExecuteError(false);
-      req.getHeaders().setContentRange(HttpContentRange.query().getHeaderValue());
+      HttpHeaders headers = req.getHeaders();
+      headers.setContentRange(HttpContentRange.query().getHeaderValue());
+      for (Entry<String, String> e : jsonResumableWrite.getExtraHeaders().entrySet()) {
+        headers.set(e.getKey(), e.getValue());
+      }
 
       response = req.execute();
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableWrite.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableWrite.java
index 336ce0e4..41bbec72 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableWrite.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableWrite.java
@@ -21,6 +21,7 @@ import static com.google.common.base.Preconditions.checkArgument;
 import com.google.api.services.storage.model.StorageObject;
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableMap;
 import com.google.gson.Gson;
 import com.google.gson.stream.JsonReader;
 import java.io.IOException;
@@ -60,6 +61,16 @@ final class JsonResumableWrite implements Serializable {
     this.beginOffset = beginOffset;
   }
 
+  ImmutableMap<String, String> getExtraHeaders() {
+    if (options != null) {
+      Object tmp = options.get(StorageRpc.Option.EXTRA_HEADERS);
+      if (tmp != null) {
+        return (ImmutableMap<String, String>) tmp;
+      }
+    }
+    return ImmutableMap.of();
+  }
+
   public @NonNull String getUploadId() {
     return uploadId;
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java
index aab52633..390b90dc 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java
@@ -35,7 +35,6 @@ import io.grpc.StatusRuntimeException;
 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.util.List;
-import java.util.Locale;
 import java.util.Map;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
@@ -295,12 +294,12 @@ enum ResumableSessionFailureScenario {
   // The header names from HttpHeaders are lower cased, define some utility methods to create
   // predicates where we can specify values ignoring case
   private static Predicate<String> matches(String expected) {
-    String lower = expected.toLowerCase(Locale.US);
+    String lower = Utils.headerNameToLowerCase(expected);
     return lower::equals;
   }
 
   private static Predicate<String> startsWith(String prefix) {
-    String lower = prefix.toLowerCase(Locale.US);
+    String lower = Utils.headerNameToLowerCase(prefix);
     return s -> s.startsWith(lower);
   }
 
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 8d6f479e..6ed3c5af 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
@@ -471,6 +471,59 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new BucketTargetOption(UnifiedOpts.projection(projection));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static BucketTargetOption extraHeaders(
+        @NonNull ImmutableMap<String, String> extraHeaders) {
+      return new BucketTargetOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -545,6 +598,59 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new BucketSourceOption(UnifiedOpts.requestedPolicyVersion(version));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static BucketSourceOption extraHeaders(
+        @NonNull ImmutableMap<String, String> extraHeaders) {
+      return new BucketSourceOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -633,6 +739,59 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new ListHmacKeysOption(UnifiedOpts.projectId(projectId));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static ListHmacKeysOption extraHeaders(
+        @NonNull ImmutableMap<String, String> extraHeaders) {
+      return new ListHmacKeysOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -691,6 +850,59 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new CreateHmacKeyOption(UnifiedOpts.projectId(projectId));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static CreateHmacKeyOption extraHeaders(
+        @NonNull ImmutableMap<String, String> extraHeaders) {
+      return new CreateHmacKeyOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -726,6 +938,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
 
   /** Class for specifying getHmacKey options */
   class GetHmacKeyOption extends Option<HmacKeySourceOpt> {
+
     private GetHmacKeyOption(HmacKeySourceOpt opt) {
       super(opt);
     }
@@ -748,6 +961,59 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new GetHmacKeyOption(UnifiedOpts.projectId(projectId));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static GetHmacKeyOption extraHeaders(
+        @NonNull ImmutableMap<String, String> extraHeaders) {
+      return new GetHmacKeyOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -782,6 +1048,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
 
   /** Class for specifying deleteHmacKey options */
   class DeleteHmacKeyOption extends Option<HmacKeyTargetOpt> {
+
     private DeleteHmacKeyOption(HmacKeyTargetOpt opt) {
       super(opt);
     }
@@ -795,6 +1062,59 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new DeleteHmacKeyOption(UnifiedOpts.userProject(userProject));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static DeleteHmacKeyOption extraHeaders(
+        @NonNull ImmutableMap<String, String> extraHeaders) {
+      return new DeleteHmacKeyOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -830,6 +1150,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
 
   /** Class for specifying updateHmacKey options */
   class UpdateHmacKeyOption extends Option<HmacKeyTargetOpt> {
+
     private UpdateHmacKeyOption(HmacKeyTargetOpt opt) {
       super(opt);
     }
@@ -843,6 +1164,59 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new UpdateHmacKeyOption(UnifiedOpts.userProject(userProject));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static UpdateHmacKeyOption extraHeaders(
+        @NonNull ImmutableMap<String, String> extraHeaders) {
+      return new UpdateHmacKeyOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -929,6 +1303,58 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new BucketGetOption(UnifiedOpts.fields(set));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static BucketGetOption extraHeaders(@NonNull ImmutableMap<String, String> extraHeaders) {
+      return new BucketGetOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -1120,6 +1546,59 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new BlobTargetOption(UnifiedOpts.overrideUnlockedRetention(overrideUnlockedRetention));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static BlobTargetOption extraHeaders(
+        @NonNull ImmutableMap<String, String> extraHeaders) {
+      return new BlobTargetOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -1346,6 +1825,58 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new BlobWriteOption(UnifiedOpts.resumableUploadExpectedObjectSize(objectContentSize));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static BlobWriteOption extraHeaders(@NonNull ImmutableMap<String, String> extraHeaders) {
+      return new BlobWriteOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -1490,6 +2021,59 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new BlobSourceOption(UnifiedOpts.returnRawInputStream(shouldReturnRawInputStream));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static BlobSourceOption extraHeaders(
+        @NonNull ImmutableMap<String, String> extraHeaders) {
+      return new BlobSourceOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -1658,6 +2242,58 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new BlobGetOption(UnifiedOpts.softDeleted(softDeleted));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static BlobGetOption extraHeaders(@NonNull ImmutableMap<String, String> extraHeaders) {
+      return new BlobGetOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -1743,6 +2379,59 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
     public static BlobRestoreOption copySourceAcl(boolean copySourceAcl) {
       return new BlobRestoreOption(UnifiedOpts.copySourceAcl(copySourceAcl));
     }
+
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static BlobRestoreOption extraHeaders(
+        @NonNull ImmutableMap<String, String> extraHeaders) {
+      return new BlobRestoreOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
   }
 
   /** Class for specifying bucket list options. */
@@ -1802,6 +2491,59 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new BucketListOption(UnifiedOpts.fields(set));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static BucketListOption extraHeaders(
+        @NonNull ImmutableMap<String, String> extraHeaders) {
+      return new BucketListOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -1979,6 +2721,58 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
       return new BlobListOption(UnifiedOpts.softDeleted(softDeleted));
     }
 
+    /**
+     * A set of extra headers to be set for all requests performed within the scope of the operation
+     * this option is passed to (a get, read, resumable upload etc).
+     *
+     * <p>If the same header name is specified across multiple options provided to a method, the
+     * first occurrence will be the value included in the request(s).
+     *
+     * <p>The following headers are not allowed to be specified, and will result in an {@link
+     * IllegalArgumentException}.
+     *
+     * <ol>
+     *   <li>{@code Accept-Encoding}
+     *   <li>{@code Cache-Control}
+     *   <li>{@code Connection}
+     *   <li>{@code Content-ID}
+     *   <li>{@code Content-Length}
+     *   <li>{@code Content-Range}
+     *   <li>{@code Content-Transfer-Encoding}
+     *   <li>{@code Content-Type}
+     *   <li>{@code Date}
+     *   <li>{@code ETag}
+     *   <li>{@code If-Match}
+     *   <li>{@code If-None-Match}
+     *   <li>{@code Keep-Alive}
+     *   <li>{@code Range}
+     *   <li>{@code TE}
+     *   <li>{@code Trailer}
+     *   <li>{@code Transfer-Encoding}
+     *   <li>{@code User-Agent}
+     *   <li>{@code X-Goog-Api-Client}
+     *   <li>{@code X-Goog-Content-Length-Range}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key}
+     *   <li>{@code X-Goog-Copy-Source-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Encryption-Algorithm}
+     *   <li>{@code X-Goog-Encryption-Key}
+     *   <li>{@code X-Goog-Encryption-Key-Sha256}
+     *   <li>{@code X-Goog-Gcs-Idempotency-Token}
+     *   <li>{@code X-Goog-Meta-*}
+     *   <li>{@code X-Goog-User-Project}
+     *   <li>{@code X-HTTP-Method-Override}
+     *   <li>{@code X-Upload-Content-Length}
+     *   <li>{@code X-Upload-Content-Type}
+     * </ol>
+     *
+     * @since 2.49.0
+     */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
+    public static BlobListOption extraHeaders(@NonNull ImmutableMap<String, String> extraHeaders) {
+      return new BlobListOption(UnifiedOpts.extraHeaders(extraHeaders));
+    }
+
     /**
      * Deduplicate any options which are the same parameter. The value which comes last in {@code
      * os} will be the value included in the return.
@@ -5036,8 +5830,6 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * {@link Storage#delete(BlobId)}, however without the ability to change metadata fields for the
    * target object.
    *
-   * <p>This feature is currently only supported for HNS (Hierarchical Namespace) buckets.
-   *
    * @since 2.48.0
    */
   @TransportCompatibility({Transport.HTTP, Transport.GRPC})
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 6c9828c1..09a335f3 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
@@ -62,7 +62,10 @@ import java.net.URLConnection;
 import java.security.Key;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.Locale;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Set;
@@ -565,6 +568,18 @@ final class UnifiedOpts {
     return Md5MatchExtractor.INSTANCE;
   }
 
+  static Headers extraHeaders(ImmutableMap<String, String> extraHeaders) {
+    requireNonNull(extraHeaders, "extraHeaders must be non null");
+    String blockedHeaders =
+        extraHeaders.keySet().stream()
+            .map(Utils::headerNameToLowerCase)
+            .filter(Headers.BLOCKLIST)
+            .sorted(Comparator.naturalOrder())
+            .collect(Collectors.joining(", ", "[", "]"));
+    checkArgument("[]".equals(blockedHeaders), "Disallowed headers: %s", blockedHeaders);
+    return new Headers(extraHeaders);
+  }
+
   static final class Crc32cMatch implements ObjectTargetOpt {
     private static final long serialVersionUID = 8172282701777561769L;
     private final int val;
@@ -1920,6 +1935,183 @@ final class UnifiedOpts {
     }
   }
 
+  static final class Headers extends RpcOptVal<ImmutableMap<String, String>>
+      implements BucketSourceOpt,
+          BucketTargetOpt,
+          BucketListOpt,
+          ObjectSourceOpt,
+          ObjectTargetOpt,
+          ObjectListOpt,
+          HmacKeySourceOpt,
+          HmacKeyTargetOpt,
+          HmacKeyListOpt {
+
+    /**
+     * The set of header names which are blocked from being able to be provided for an instance of
+     * this class.
+     *
+     * <p>Most values here are from <a
+     * href="https://cloud.google.com/storage/docs/json_api/v1/parameters">the json api
+     * parameters</a> list or general http headers our client otherwise sets during the course of
+     * normal operation.
+     */
+    private static final Predicate<String> BLOCKLIST;
+
+    static {
+      ImmutableSet<String> fullHeaderNames =
+          Stream.of(
+                  "Accept-Encoding",
+                  "Cache-Control",
+                  "Connection",
+                  "Content-ID",
+                  "Content-Length",
+                  "Content-Range",
+                  "Content-Transfer-Encoding",
+                  "Content-Type",
+                  "Date",
+                  "ETag",
+                  "If-Match",
+                  "If-None-Match",
+                  "Keep-Alive",
+                  "Range",
+                  "TE",
+                  "Trailer",
+                  "Transfer-Encoding",
+                  "User-Agent",
+                  "X-Goog-Api-Client",
+                  "X-Goog-Content-Length-Range",
+                  "X-Goog-Copy-Source-Encryption-Algorithm",
+                  "X-Goog-Copy-Source-Encryption-Key",
+                  "X-Goog-Copy-Source-Encryption-Key-Sha256",
+                  "X-Goog-Encryption-Algorithm",
+                  "X-Goog-Encryption-Key",
+                  "X-Goog-Encryption-Key-Sha256",
+                  "X-Goog-Gcs-Idempotency-Token",
+                  "X-Goog-Request-Params",
+                  "X-Goog-User-Project",
+                  "X-HTTP-Method-Override",
+                  "X-Upload-Content-Length",
+                  "X-Upload-Content-Type")
+              .map(Utils::headerNameToLowerCase)
+              .collect(ImmutableSet.toImmutableSet());
+
+      ImmutableSet<String> prefixes =
+          Stream.of("X-Goog-Meta-")
+              .map(Utils::headerNameToLowerCase)
+              .collect(ImmutableSet.toImmutableSet());
+
+      BLOCKLIST =
+          name -> {
+            if (fullHeaderNames.contains(name)) {
+              return true;
+            }
+
+            for (String prefix : prefixes) {
+              if (name.startsWith(prefix)) {
+                return true;
+              }
+            }
+            return false;
+          };
+    }
+
+    private Headers(ImmutableMap<String, String> val) {
+      super(StorageRpc.Option.EXTRA_HEADERS, val);
+    }
+
+    @Override
+    public Mapper<GrpcCallContext> getGrpcMetadataMapper() {
+      return ctx -> {
+        if (val.isEmpty()) {
+          return ctx;
+        }
+        Set<String> existingHeaderNames =
+            ctx.getExtraHeaders().keySet().stream()
+                .map(Utils::headerNameToLowerCase)
+                .collect(Collectors.toSet());
+        Map<String, List<String>> wrapped = new HashMap<>();
+        for (Entry<String, String> e : val.entrySet()) {
+          String key = Utils.headerNameToLowerCase(e.getKey());
+          if (existingHeaderNames.contains(key)) {
+            continue;
+          }
+          wrapped.put(key, ImmutableList.of(e.getValue()));
+        }
+        return ctx.withExtraHeaders(wrapped);
+      };
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public Mapper<ImmutableMap.Builder<StorageRpc.Option, Object>> mapper() {
+      return optionBuilder -> {
+        if (val.isEmpty()) {
+          return optionBuilder;
+        }
+        // not ideal, but ImmutableMap.Builder doesn't have any read methods so we can detect
+        // collision before build time.
+        ImmutableMap<StorageRpc.Option, Object> builtOptions = Utils.mapBuild(optionBuilder);
+        ImmutableMap<String, String> tmp =
+            (ImmutableMap<String, String>) builtOptions.get(StorageRpc.Option.EXTRA_HEADERS);
+        if (tmp == null) {
+          ImmutableMap.Builder<String, String> b = ImmutableMap.builder();
+          for (Entry<String, String> e : val.entrySet()) {
+            String key = Utils.headerNameToLowerCase(e.getKey());
+            b.put(key, e.getValue());
+          }
+          optionBuilder.put(key, Utils.mapBuild(b));
+          return optionBuilder;
+        } else {
+          ImmutableMap.Builder<StorageRpc.Option, Object> newOptionBuilder = ImmutableMap.builder();
+          for (Entry<StorageRpc.Option, Object> e : builtOptions.entrySet()) {
+            if (e.getKey() != key) {
+              newOptionBuilder.put(e.getKey(), e.getValue());
+            }
+          }
+
+          ImmutableMap.Builder<String, String> extraHeadersBuilder = ImmutableMap.builder();
+          copyEntries(tmp, extraHeadersBuilder);
+          newOptionBuilder.put(key, Utils.mapBuild(extraHeadersBuilder));
+          return newOptionBuilder;
+        }
+      };
+    }
+
+    public Mapper<ImmutableMap<String, String>> extraHeadersMapper() {
+      return map -> {
+        if (val.isEmpty()) {
+          return map;
+        }
+        ImmutableMap.Builder<String, String> b = ImmutableMap.builder();
+        copyEntries(map, b);
+        return Utils.mapBuild(b);
+      };
+    }
+
+    private void copyEntries(
+        ImmutableMap<String, String> map, ImmutableMap.Builder<String, String> b) {
+      Set<String> existingHeaderNames =
+          map.keySet().stream().map(Utils::headerNameToLowerCase).collect(Collectors.toSet());
+      b.putAll(map);
+      for (Entry<String, String> e : val.entrySet()) {
+        String key = Utils.headerNameToLowerCase(e.getKey());
+        if (!existingHeaderNames.contains(key)) {
+          b.put(key, e.getValue());
+        }
+      }
+    }
+
+    @Override
+    public Mapper<RewriteObjectRequest.Builder> rewriteObject() {
+      return Mapper.identity();
+    }
+
+    @Override
+    public Mapper<MoveObjectRequest.Builder> moveObject() {
+      return Mapper.identity();
+    }
+  }
+
   static final class VersionsFilter extends RpcOptVal<@NonNull Boolean> implements ObjectListOpt {
     private VersionsFilter(boolean val) {
       super(StorageRpc.Option.VERSIONS, val);
@@ -2003,7 +2195,7 @@ final class UnifiedOpts {
 
     private ObjectTargetOpt detectForName(String name) {
       if (name != null) {
-        String nameLower = name.toLowerCase(Locale.ENGLISH);
+        String nameLower = Utils.headerNameToLowerCase(name);
         String contentTypeFor = FILE_NAME_MAP.getContentTypeFor(nameLower);
         if (contentTypeFor != null) {
           return new SetContentType(contentTypeFor);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java
index 853294c1..a2de1ad5 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java
@@ -40,6 +40,7 @@ import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
 import java.util.Arrays;
+import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
@@ -319,4 +320,8 @@ final class Utils {
   static <K, V> ImmutableMap<K, V> mapBuild(ImmutableMap.Builder<K, V> b) {
     return (ImmutableMap<K, V>) mapBuild.apply(b);
   }
+
+  static String headerNameToLowerCase(String headerName) {
+    return headerName.toLowerCase(Locale.US);
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/XGoogApiClientHeaderProvider.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/XGoogApiClientHeaderProvider.java
index 263a6705..310c2b12 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/XGoogApiClientHeaderProvider.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/XGoogApiClientHeaderProvider.java
@@ -24,7 +24,6 @@ import com.google.common.collect.MapDifference;
 import com.google.common.collect.MapDifference.ValueDifference;
 import com.google.common.collect.Maps;
 import java.util.HashMap;
-import java.util.Locale;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.stream.Collector;
@@ -98,7 +97,7 @@ final class XGoogApiClientHeaderProvider implements HeaderProvider {
     for (Entry<String, String> e : orig.entrySet()) {
       String k = e.getKey();
       String v = e.getValue();
-      tmp.put(k.toLowerCase(Locale.US), v);
+      tmp.put(Utils.headerNameToLowerCase(k), v);
     }
     return tmp;
   }
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 fc8553e7..286654de 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
@@ -43,17 +43,28 @@ import com.google.api.client.json.JsonFactory;
 import com.google.api.client.json.jackson2.JacksonFactory;
 import com.google.api.client.util.Data;
 import com.google.api.services.storage.Storage;
+import com.google.api.services.storage.Storage.BucketAccessControls;
+import com.google.api.services.storage.Storage.Buckets;
+import com.google.api.services.storage.Storage.Buckets.LockRetentionPolicy;
+import com.google.api.services.storage.Storage.Buckets.SetIamPolicy;
+import com.google.api.services.storage.Storage.Buckets.TestIamPermissions;
+import com.google.api.services.storage.Storage.DefaultObjectAccessControls;
+import com.google.api.services.storage.Storage.Notifications;
+import com.google.api.services.storage.Storage.ObjectAccessControls;
 import com.google.api.services.storage.Storage.Objects.Compose;
 import com.google.api.services.storage.Storage.Objects.Delete;
 import com.google.api.services.storage.Storage.Objects.Get;
 import com.google.api.services.storage.Storage.Objects.Insert;
 import com.google.api.services.storage.Storage.Objects.Move;
 import com.google.api.services.storage.Storage.Objects.Patch;
+import com.google.api.services.storage.Storage.Projects;
+import com.google.api.services.storage.Storage.Projects.HmacKeys;
+import com.google.api.services.storage.Storage.Projects.HmacKeys.Create;
+import com.google.api.services.storage.Storage.Projects.HmacKeys.Update;
 import com.google.api.services.storage.StorageRequest;
 import com.google.api.services.storage.model.Bucket;
 import com.google.api.services.storage.model.Bucket.RetentionPolicy;
 import com.google.api.services.storage.model.BucketAccessControl;
-import com.google.api.services.storage.model.Buckets;
 import com.google.api.services.storage.model.ComposeRequest;
 import com.google.api.services.storage.model.ComposeRequest.SourceObjects.ObjectPreconditions;
 import com.google.api.services.storage.model.HmacKey;
@@ -73,6 +84,7 @@ import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
 import com.google.common.base.Function;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import com.google.common.hash.HashFunction;
@@ -97,6 +109,7 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.UUID;
 import java.util.stream.Collector;
 import java.util.stream.Collectors;
@@ -375,15 +388,18 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_CREATE_BUCKET);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .buckets()
-          .insert(this.options.getProjectId(), bucket)
-          .setProjection(DEFAULT_PROJECTION)
-          .setPredefinedAcl(Option.PREDEFINED_ACL.getString(options))
-          .setPredefinedDefaultObjectAcl(Option.PREDEFINED_DEFAULT_OBJECT_ACL.getString(options))
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .setEnableObjectRetention(Option.ENABLE_OBJECT_RETENTION.getBoolean(options))
-          .execute();
+      Storage.Buckets.Insert insert =
+          storage
+              .buckets()
+              .insert(this.options.getProjectId(), bucket)
+              .setProjection(DEFAULT_PROJECTION)
+              .setPredefinedAcl(Option.PREDEFINED_ACL.getString(options))
+              .setPredefinedDefaultObjectAcl(
+                  Option.PREDEFINED_DEFAULT_OBJECT_ACL.getString(options))
+              .setUserProject(Option.USER_PROJECT.getString(options))
+              .setEnableObjectRetention(Option.ENABLE_OBJECT_RETENTION.getBoolean(options));
+      setExtraHeaders(insert, options);
+      return insert.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -412,6 +428,7 @@ public class HttpStorageRpc implements StorageRpc {
         insert.setDisableGZipContent(disableGzipContent);
       }
       setEncryptionHeaders(insert.getRequestHeaders(), ENCRYPTION_KEY_PREFIX, options);
+      setExtraHeaders(insert, options);
       return insert
           .setProjection(DEFAULT_PROJECTION)
           .setPredefinedAcl(Option.PREDEFINED_ACL.getString(options))
@@ -436,7 +453,7 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_LIST_BUCKETS);
     Scope scope = tracer.withSpan(span);
     try {
-      Buckets buckets =
+      Storage.Buckets.List list =
           storage
               .buckets()
               .list(this.options.getProjectId())
@@ -445,8 +462,9 @@ public class HttpStorageRpc implements StorageRpc {
               .setMaxResults(Option.MAX_RESULTS.getLong(options))
               .setPageToken(Option.PAGE_TOKEN.getString(options))
               .setFields(Option.FIELDS.getString(options))
-              .setUserProject(Option.USER_PROJECT.getString(options))
-              .execute();
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(list, options);
+      com.google.api.services.storage.model.Buckets buckets = list.execute();
       return Tuple.<String, Iterable<Bucket>>of(buckets.getNextPageToken(), buckets.getItems());
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
@@ -462,7 +480,7 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_LIST_OBJECTS);
     Scope scope = tracer.withSpan(span);
     try {
-      Objects objects =
+      Storage.Objects.List list =
           storage
               .objects()
               .list(bucket)
@@ -478,8 +496,9 @@ public class HttpStorageRpc implements StorageRpc {
               .setFields(Option.FIELDS.getString(options))
               .setUserProject(Option.USER_PROJECT.getString(options))
               .setSoftDeleted(Option.SOFT_DELETED.getBoolean(options))
-              .setIncludeFoldersAsPrefixes(Option.INCLUDE_FOLDERS_AS_PREFIXES.getBoolean(options))
-              .execute();
+              .setIncludeFoldersAsPrefixes(Option.INCLUDE_FOLDERS_AS_PREFIXES.getBoolean(options));
+      setExtraHeaders(list, options);
+      Objects objects = list.execute();
       Iterable<StorageObject> storageObjects =
           Iterables.concat(
               firstNonNull(objects.getItems(), ImmutableList.<StorageObject>of()),
@@ -503,7 +522,7 @@ public class HttpStorageRpc implements StorageRpc {
     }
 
     if (Boolean.TRUE == Option.DETECT_CONTENT_TYPE.get(options)) {
-      contentType = FILE_NAME_MAP.getContentTypeFor(object.getName().toLowerCase(Locale.ENGLISH));
+      contentType = FILE_NAME_MAP.getContentTypeFor(object.getName().toLowerCase(Locale.US));
     }
 
     return firstNonNull(contentType, "application/octet-stream");
@@ -527,15 +546,17 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_GET_BUCKET);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .buckets()
-          .get(bucket.getName())
-          .setProjection(DEFAULT_PROJECTION)
-          .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
-          .setIfMetagenerationNotMatch(Option.IF_METAGENERATION_NOT_MATCH.getLong(options))
-          .setFields(Option.FIELDS.getString(options))
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .execute();
+      Storage.Buckets.Get get =
+          storage
+              .buckets()
+              .get(bucket.getName())
+              .setProjection(DEFAULT_PROJECTION)
+              .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
+              .setIfMetagenerationNotMatch(Option.IF_METAGENERATION_NOT_MATCH.getLong(options))
+              .setFields(Option.FIELDS.getString(options))
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(get, options);
+      return get.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       StorageException serviceException = translate(ex);
@@ -553,6 +574,7 @@ public class HttpStorageRpc implements StorageRpc {
       throws IOException {
     Storage.Objects.Get get = storage.objects().get(object.getBucket(), object.getName());
     setEncryptionHeaders(get.getRequestHeaders(), ENCRYPTION_KEY_PREFIX, options);
+    setExtraHeaders(get, options);
     return get.setGeneration(object.getGeneration())
         .setProjection(DEFAULT_PROJECTION)
         .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
@@ -590,6 +612,7 @@ public class HttpStorageRpc implements StorageRpc {
     try {
       Storage.Objects.Restore restore =
           storage.objects().restore(object.getBucket(), object.getName(), object.getGeneration());
+      setExtraHeaders(restore, options);
       return restore
           .setProjection(DEFAULT_PROJECTION)
           .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
@@ -644,16 +667,19 @@ public class HttpStorageRpc implements StorageRpc {
           projection = NO_ACL_PROJECTION;
         }
       }
-      return storage
-          .buckets()
-          .patch(bucket.getName(), bucket)
-          .setProjection(projection == null ? DEFAULT_PROJECTION : projection)
-          .setPredefinedAcl(Option.PREDEFINED_ACL.getString(options))
-          .setPredefinedDefaultObjectAcl(Option.PREDEFINED_DEFAULT_OBJECT_ACL.getString(options))
-          .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
-          .setIfMetagenerationNotMatch(Option.IF_METAGENERATION_NOT_MATCH.getLong(options))
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .execute();
+      Buckets.Patch patch =
+          storage
+              .buckets()
+              .patch(bucket.getName(), bucket)
+              .setProjection(projection == null ? DEFAULT_PROJECTION : projection)
+              .setPredefinedAcl(Option.PREDEFINED_ACL.getString(options))
+              .setPredefinedDefaultObjectAcl(
+                  Option.PREDEFINED_DEFAULT_OBJECT_ACL.getString(options))
+              .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
+              .setIfMetagenerationNotMatch(Option.IF_METAGENERATION_NOT_MATCH.getLong(options))
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(patch, options);
+      return patch.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -665,17 +691,20 @@ public class HttpStorageRpc implements StorageRpc {
 
   private Storage.Objects.Patch patchCall(StorageObject storageObject, Map<Option, ?> options)
       throws IOException {
-    return storage
-        .objects()
-        .patch(storageObject.getBucket(), storageObject.getName(), storageObject)
-        .setProjection(DEFAULT_PROJECTION)
-        .setPredefinedAcl(Option.PREDEFINED_ACL.getString(options))
-        .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
-        .setIfMetagenerationNotMatch(Option.IF_METAGENERATION_NOT_MATCH.getLong(options))
-        .setIfGenerationMatch(Option.IF_GENERATION_MATCH.getLong(options))
-        .setIfGenerationNotMatch(Option.IF_GENERATION_NOT_MATCH.getLong(options))
-        .setOverrideUnlockedRetention(Option.OVERRIDE_UNLOCKED_RETENTION.getBoolean(options))
-        .setUserProject(Option.USER_PROJECT.getString(options));
+    Storage.Objects.Patch patch =
+        storage
+            .objects()
+            .patch(storageObject.getBucket(), storageObject.getName(), storageObject)
+            .setProjection(DEFAULT_PROJECTION)
+            .setPredefinedAcl(Option.PREDEFINED_ACL.getString(options))
+            .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
+            .setIfMetagenerationNotMatch(Option.IF_METAGENERATION_NOT_MATCH.getLong(options))
+            .setIfGenerationMatch(Option.IF_GENERATION_MATCH.getLong(options))
+            .setIfGenerationNotMatch(Option.IF_GENERATION_NOT_MATCH.getLong(options))
+            .setOverrideUnlockedRetention(Option.OVERRIDE_UNLOCKED_RETENTION.getBoolean(options))
+            .setUserProject(Option.USER_PROJECT.getString(options));
+    setExtraHeaders(patch, options);
+    return patch;
   }
 
   @Override
@@ -698,13 +727,15 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_DELETE_BUCKET);
     Scope scope = tracer.withSpan(span);
     try {
-      storage
-          .buckets()
-          .delete(bucket.getName())
-          .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
-          .setIfMetagenerationNotMatch(Option.IF_METAGENERATION_NOT_MATCH.getLong(options))
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .execute();
+      Buckets.Delete delete =
+          storage
+              .buckets()
+              .delete(bucket.getName())
+              .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
+              .setIfMetagenerationNotMatch(Option.IF_METAGENERATION_NOT_MATCH.getLong(options))
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(delete, options);
+      delete.execute();
       return true;
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
@@ -721,15 +752,18 @@ public class HttpStorageRpc implements StorageRpc {
 
   private Storage.Objects.Delete deleteCall(StorageObject blob, Map<Option, ?> options)
       throws IOException {
-    return storage
-        .objects()
-        .delete(blob.getBucket(), blob.getName())
-        .setGeneration(blob.getGeneration())
-        .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
-        .setIfMetagenerationNotMatch(Option.IF_METAGENERATION_NOT_MATCH.getLong(options))
-        .setIfGenerationMatch(Option.IF_GENERATION_MATCH.getLong(options))
-        .setIfGenerationNotMatch(Option.IF_GENERATION_NOT_MATCH.getLong(options))
-        .setUserProject(Option.USER_PROJECT.getString(options));
+    Storage.Objects.Delete delete =
+        storage
+            .objects()
+            .delete(blob.getBucket(), blob.getName())
+            .setGeneration(blob.getGeneration())
+            .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
+            .setIfMetagenerationNotMatch(Option.IF_METAGENERATION_NOT_MATCH.getLong(options))
+            .setIfGenerationMatch(Option.IF_GENERATION_MATCH.getLong(options))
+            .setIfGenerationNotMatch(Option.IF_GENERATION_NOT_MATCH.getLong(options))
+            .setUserProject(Option.USER_PROJECT.getString(options));
+    setExtraHeaders(delete, options);
+    return delete;
   }
 
   @Override
@@ -781,6 +815,7 @@ public class HttpStorageRpc implements StorageRpc {
               .setIfGenerationMatch(Option.IF_GENERATION_MATCH.getLong(targetOptions))
               .setUserProject(Option.USER_PROJECT.getString(targetOptions));
       setEncryptionHeaders(compose.getRequestHeaders(), ENCRYPTION_KEY_PREFIX, targetOptions);
+      setExtraHeaders(compose, targetOptions);
       return compose.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
@@ -807,6 +842,7 @@ public class HttpStorageRpc implements StorageRpc {
               .setIfGenerationNotMatch(Option.IF_GENERATION_NOT_MATCH.getLong(options))
               .setUserProject(Option.USER_PROJECT.getString(options));
       setEncryptionHeaders(getRequest.getRequestHeaders(), ENCRYPTION_KEY_PREFIX, options);
+      setExtraHeaders(getRequest, options);
       if (Option.RETURN_RAW_INPUT_STREAM.getBoolean(options) != null) {
         getRequest.setReturnRawInputStream(Option.RETURN_RAW_INPUT_STREAM.getBoolean(options));
       }
@@ -846,6 +882,7 @@ public class HttpStorageRpc implements StorageRpc {
             .setIfGenerationNotMatch(Option.IF_GENERATION_NOT_MATCH.getLong(options))
             .setUserProject(Option.USER_PROJECT.getString(options));
     setEncryptionHeaders(req.getRequestHeaders(), ENCRYPTION_KEY_PREFIX, options);
+    setExtraHeaders(req, options);
     return req;
   }
 
@@ -1112,6 +1149,7 @@ public class HttpStorageRpc implements StorageRpc {
         requestHeaders.set("X-Upload-Content-Length", xUploadContentLength);
       }
       setEncryptionHeaders(requestHeaders, "x-goog-encryption-", options);
+      setExtraHeaders(Option.EXTRA_HEADERS.get(options), requestHeaders);
       HttpResponse response = httpRequest.execute();
       if (response.getStatusCode() != 200) {
         throw buildStorageException(response.getStatusCode(), response.getStatusMessage());
@@ -1263,6 +1301,8 @@ public class HttpStorageRpc implements StorageRpc {
       HttpHeaders requestHeaders = rewrite.getRequestHeaders();
       setEncryptionHeaders(requestHeaders, SOURCE_ENCRYPTION_KEY_PREFIX, req.sourceOptions);
       setEncryptionHeaders(requestHeaders, ENCRYPTION_KEY_PREFIX, req.targetOptions);
+      setExtraHeaders(rewrite, req.sourceOptions);
+      setExtraHeaders(rewrite, req.targetOptions);
       com.google.api.services.storage.model.RewriteResponse rewriteResponse = rewrite.execute();
       return new RewriteResponse(
           req,
@@ -1282,11 +1322,13 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_GET_BUCKET_ACL);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .bucketAccessControls()
-          .get(bucket, entity)
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .execute();
+      BucketAccessControls.Get get =
+          storage
+              .bucketAccessControls()
+              .get(bucket, entity)
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(get, options);
+      return get.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       StorageException serviceException = translate(ex);
@@ -1305,11 +1347,13 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_DELETE_BUCKET_ACL);
     Scope scope = tracer.withSpan(span);
     try {
-      storage
-          .bucketAccessControls()
-          .delete(bucket, entity)
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .execute();
+      BucketAccessControls.Delete delete =
+          storage
+              .bucketAccessControls()
+              .delete(bucket, entity)
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(delete, options);
+      delete.execute();
       return true;
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
@@ -1329,11 +1373,13 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_CREATE_BUCKET_ACL);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .bucketAccessControls()
-          .insert(acl.getBucket(), acl)
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .execute();
+      BucketAccessControls.Insert insert =
+          storage
+              .bucketAccessControls()
+              .insert(acl.getBucket(), acl)
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(insert, options);
+      return insert.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1348,11 +1394,13 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_PATCH_BUCKET_ACL);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .bucketAccessControls()
-          .patch(acl.getBucket(), acl.getEntity(), acl)
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .execute();
+      BucketAccessControls.Patch patch =
+          storage
+              .bucketAccessControls()
+              .patch(acl.getBucket(), acl.getEntity(), acl)
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(patch, options);
+      return patch.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1367,12 +1415,13 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_LIST_BUCKET_ACLS);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .bucketAccessControls()
-          .list(bucket)
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .execute()
-          .getItems();
+      BucketAccessControls.List list =
+          storage
+              .bucketAccessControls()
+              .list(bucket)
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(list, options);
+      return list.execute().getItems();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1387,7 +1436,9 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_GET_OBJECT_DEFAULT_ACL);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage.defaultObjectAccessControls().get(bucket, entity).execute();
+      DefaultObjectAccessControls.Get get =
+          storage.defaultObjectAccessControls().get(bucket, entity);
+      return get.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       StorageException serviceException = translate(ex);
@@ -1406,7 +1457,9 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_DELETE_OBJECT_DEFAULT_ACL);
     Scope scope = tracer.withSpan(span);
     try {
-      storage.defaultObjectAccessControls().delete(bucket, entity).execute();
+      DefaultObjectAccessControls.Delete delete =
+          storage.defaultObjectAccessControls().delete(bucket, entity);
+      delete.execute();
       return true;
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
@@ -1426,7 +1479,9 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_CREATE_OBJECT_DEFAULT_ACL);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage.defaultObjectAccessControls().insert(acl.getBucket(), acl).execute();
+      DefaultObjectAccessControls.Insert insert =
+          storage.defaultObjectAccessControls().insert(acl.getBucket(), acl);
+      return insert.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1441,10 +1496,9 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_PATCH_OBJECT_DEFAULT_ACL);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .defaultObjectAccessControls()
-          .patch(acl.getBucket(), acl.getEntity(), acl)
-          .execute();
+      DefaultObjectAccessControls.Patch patch =
+          storage.defaultObjectAccessControls().patch(acl.getBucket(), acl.getEntity(), acl);
+      return patch.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1459,7 +1513,8 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_LIST_OBJECT_DEFAULT_ACLS);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage.defaultObjectAccessControls().list(bucket).execute().getItems();
+      DefaultObjectAccessControls.List list = storage.defaultObjectAccessControls().list(bucket);
+      return list.execute().getItems();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1474,11 +1529,9 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_GET_OBJECT_ACL);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .objectAccessControls()
-          .get(bucket, object, entity)
-          .setGeneration(generation)
-          .execute();
+      ObjectAccessControls.Get get =
+          storage.objectAccessControls().get(bucket, object, entity).setGeneration(generation);
+      return get.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       StorageException serviceException = translate(ex);
@@ -1497,11 +1550,9 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_DELETE_OBJECT_ACL);
     Scope scope = tracer.withSpan(span);
     try {
-      storage
-          .objectAccessControls()
-          .delete(bucket, object, entity)
-          .setGeneration(generation)
-          .execute();
+      ObjectAccessControls.Delete delete =
+          storage.objectAccessControls().delete(bucket, object, entity).setGeneration(generation);
+      delete.execute();
       return true;
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
@@ -1521,11 +1572,12 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_CREATE_OBJECT_ACL);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .objectAccessControls()
-          .insert(acl.getBucket(), acl.getObject(), acl)
-          .setGeneration(acl.getGeneration())
-          .execute();
+      ObjectAccessControls.Insert insert =
+          storage
+              .objectAccessControls()
+              .insert(acl.getBucket(), acl.getObject(), acl)
+              .setGeneration(acl.getGeneration());
+      return insert.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1540,11 +1592,12 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_PATCH_OBJECT_ACL);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .objectAccessControls()
-          .patch(acl.getBucket(), acl.getObject(), acl.getEntity(), acl)
-          .setGeneration(acl.getGeneration())
-          .execute();
+      ObjectAccessControls.Patch patch =
+          storage
+              .objectAccessControls()
+              .patch(acl.getBucket(), acl.getObject(), acl.getEntity(), acl)
+              .setGeneration(acl.getGeneration());
+      return patch.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1559,12 +1612,9 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_LIST_OBJECT_ACLS);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .objectAccessControls()
-          .list(bucket, object)
-          .setGeneration(generation)
-          .execute()
-          .getItems();
+      ObjectAccessControls.List list =
+          storage.objectAccessControls().list(bucket, object).setGeneration(generation);
+      return list.execute().getItems();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1583,13 +1633,14 @@ public class HttpStorageRpc implements StorageRpc {
       projectId = this.options.getProjectId();
     }
     try {
-      return storage
-          .projects()
-          .hmacKeys()
-          .create(projectId, serviceAccountEmail)
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .setDisableGZipContent(true)
-          .execute();
+      Create create =
+          storage
+              .projects()
+              .hmacKeys()
+              .create(projectId, serviceAccountEmail)
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(create, options);
+      return create.setDisableGZipContent(true).execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1608,7 +1659,7 @@ public class HttpStorageRpc implements StorageRpc {
       projectId = this.options.getProjectId();
     }
     try {
-      HmacKeysMetadata hmacKeysMetadata =
+      HmacKeys.List list =
           storage
               .projects()
               .hmacKeys()
@@ -1617,8 +1668,9 @@ public class HttpStorageRpc implements StorageRpc {
               .setPageToken(Option.PAGE_TOKEN.getString(options))
               .setMaxResults(Option.MAX_RESULTS.getLong(options))
               .setShowDeletedKeys(Option.SHOW_DELETED_KEYS.getBoolean(options))
-              .setUserProject(Option.USER_PROJECT.getString(options))
-              .execute();
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(list, options);
+      HmacKeysMetadata hmacKeysMetadata = list.execute();
       return Tuple.<String, Iterable<HmacKeyMetadata>>of(
           hmacKeysMetadata.getNextPageToken(), hmacKeysMetadata.getItems());
     } catch (IOException ex) {
@@ -1639,12 +1691,14 @@ public class HttpStorageRpc implements StorageRpc {
       projectId = this.options.getProjectId();
     }
     try {
-      return storage
-          .projects()
-          .hmacKeys()
-          .get(projectId, accessId)
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .execute();
+      HmacKeys.Get get =
+          storage
+              .projects()
+              .hmacKeys()
+              .get(projectId, accessId)
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(get, options);
+      return get.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1663,12 +1717,14 @@ public class HttpStorageRpc implements StorageRpc {
       projectId = this.options.getProjectId();
     }
     try {
-      return storage
-          .projects()
-          .hmacKeys()
-          .update(projectId, hmacKeyMetadata.getAccessId(), hmacKeyMetadata)
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .execute();
+      Update update =
+          storage
+              .projects()
+              .hmacKeys()
+              .update(projectId, hmacKeyMetadata.getAccessId(), hmacKeyMetadata)
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(update, options);
+      return update.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1687,12 +1743,14 @@ public class HttpStorageRpc implements StorageRpc {
       projectId = this.options.getProjectId();
     }
     try {
-      storage
-          .projects()
-          .hmacKeys()
-          .delete(projectId, hmacKeyMetadata.getAccessId())
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .execute();
+      HmacKeys.Delete delete =
+          storage
+              .projects()
+              .hmacKeys()
+              .delete(projectId, hmacKeyMetadata.getAccessId())
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(delete, options);
+      delete.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1716,6 +1774,7 @@ public class HttpStorageRpc implements StorageRpc {
         getIamPolicy.setOptionsRequestedPolicyVersion(
             Option.REQUESTED_POLICY_VERSION.getLong(options).intValue());
       }
+      setExtraHeaders(getIamPolicy, options);
       return getIamPolicy.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
@@ -1731,11 +1790,13 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_SET_BUCKET_IAM_POLICY);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .buckets()
-          .setIamPolicy(bucket, policy)
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .execute();
+      SetIamPolicy setIamPolicy =
+          storage
+              .buckets()
+              .setIamPolicy(bucket, policy)
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(setIamPolicy, options);
+      return setIamPolicy.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1751,11 +1812,13 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_TEST_BUCKET_IAM_PERMISSIONS);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .buckets()
-          .testIamPermissions(bucket, permissions)
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .execute();
+      TestIamPermissions testIamPermissions =
+          storage
+              .buckets()
+              .testIamPermissions(bucket, permissions)
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(testIamPermissions, options);
+      return testIamPermissions.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1770,7 +1833,8 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_DELETE_NOTIFICATION);
     Scope scope = tracer.withSpan(span);
     try {
-      storage.notifications().delete(bucket, notification).execute();
+      Notifications.Delete delete = storage.notifications().delete(bucket, notification);
+      delete.execute();
       return true;
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
@@ -1790,7 +1854,8 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_LIST_NOTIFICATIONS);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage.notifications().list(bucket).execute().getItems();
+      Notifications.List list = storage.notifications().list(bucket);
+      return list.execute().getItems();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1805,7 +1870,8 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_CREATE_NOTIFICATION);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage.notifications().insert(bucket, notification).execute();
+      Notifications.Insert insert = storage.notifications().insert(bucket, notification);
+      return insert.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1820,7 +1886,8 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_GET_NOTIFICATION);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage.notifications().get(bucket, notification).execute();
+      Notifications.Get get = storage.notifications().get(bucket, notification);
+      return get.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       StorageException serviceException = translate(ex);
@@ -1839,12 +1906,14 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_LOCK_RETENTION_POLICY);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage
-          .buckets()
-          .lockRetentionPolicy(bucket.getName(), Option.IF_METAGENERATION_MATCH.getLong(options))
-          .setUserProject(Option.USER_PROJECT.getString(options))
-          .setDisableGZipContent(true)
-          .execute();
+      LockRetentionPolicy lockRetentionPolicy =
+          storage
+              .buckets()
+              .lockRetentionPolicy(
+                  bucket.getName(), Option.IF_METAGENERATION_MATCH.getLong(options))
+              .setUserProject(Option.USER_PROJECT.getString(options));
+      setExtraHeaders(lockRetentionPolicy, options);
+      return lockRetentionPolicy.setDisableGZipContent(true).execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1859,7 +1928,8 @@ public class HttpStorageRpc implements StorageRpc {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_GET_SERVICE_ACCOUNT);
     Scope scope = tracer.withSpan(span);
     try {
-      return storage.projects().serviceAccount().get(projectId).execute();
+      Projects.ServiceAccount.Get get = storage.projects().serviceAccount().get(projectId);
+      return get.execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
       throw translate(ex);
@@ -1875,4 +1945,23 @@ public class HttpStorageRpc implements StorageRpc {
     error.setMessage(statusMessage);
     return translate(error);
   }
+
+  private static <Resource, Request extends StorageRequest<Resource>> void setExtraHeaders(
+      Request req, Map<Option, ?> options) {
+    ImmutableMap<String, String> extraHeaders = Option.EXTRA_HEADERS.get(options);
+    HttpHeaders headers = req.getRequestHeaders();
+    setExtraHeaders(extraHeaders, headers);
+  }
+
+  private static void setExtraHeaders(
+      @Nullable ImmutableMap<String, String> extraHeaders, HttpHeaders headers) {
+    if (extraHeaders != null) {
+      for (Entry<String, String> e : extraHeaders.entrySet()) {
+        String key = e.getKey();
+        if (!headers.containsKey(key) || key.equals("authorization")) {
+          headers.set(key, e.getValue());
+        }
+      }
+    }
+  }
 }
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 ffd1766b..dc056fda 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
@@ -78,7 +78,12 @@ public interface StorageRpc extends ServiceRpc {
     COPY_SOURCE_ACL("copySourceAcl"),
     GENERATION("generation"),
     INCLUDE_FOLDERS_AS_PREFIXES("includeFoldersAsPrefixes"),
-    X_UPLOAD_CONTENT_LENGTH("x-upload-content-length");
+    X_UPLOAD_CONTENT_LENGTH("x-upload-content-length"),
+    /**
+     * An {@link com.google.common.collect.ImmutableMap ImmutableMap&lt;String, String>} of values
+     * which will be set as additional headers on the request.
+     */
+    EXTRA_HEADERS("extra_headers");
 
     private final String value;
 
diff --git a/pom.xml b/pom.xml
index cf4e8e8b..7dc31d20 100644
--- a/pom.xml
+++ b/pom.xml
@@ -69,7 +69,7 @@
       <dependency>
         <groupId>org.junit</groupId>
         <artifactId>junit-bom</artifactId>
-        <version>5.11.4</version>
+        <version>5.12.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
diff --git a/renovate.json b/renovate.json
index 48b00313..c4eb8ff7 100644
--- a/renovate.json
+++ b/renovate.json
@@ -94,7 +94,9 @@
         "^org.mockito:mockito-core",
         "^org.objenesis:objenesis",
         "^com.google.cloud:google-cloud-conformance-tests",
-        "^io.github.classgraph:classgraph"
+        "^io.github.classgraph:classgraph",
+        "^ch.qos.logback:logback-classic",
+        "^org.slf4j:jul-to-slf4j"
       ],
       "semanticCommitType": "test",
       "semanticCommitScope": "deps"
@@ -139,6 +141,12 @@
         "^com.google.cloud:sdk-platform-java-config"
       ],
       "groupName": "sdk-platform-java dependencies"
+    },
+    {
+      "packagePatterns": [
+        "^ch.qos.logback:logback-classic"
+      ],
+      "allowedVersions": "<1.4.0"
     }
   ],
   "semanticCommits": true,
diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml
index 7a517595..25f002fa 100644
--- a/samples/snippets/pom.xml
+++ b/samples/snippets/pom.xml
@@ -31,7 +31,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>26.54.0</version>
+        <version>26.55.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITExtraHeadersOptionTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITExtraHeadersOptionTest.java
new file mode 100644
index 00000000..d72be29a
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITExtraHeadersOptionTest.java
@@ -0,0 +1,377 @@
+/*
+ * Copyright 2025 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;
+
+import static com.google.cloud.storage.TestUtils.assertAll;
+import static com.google.cloud.storage.TestUtils.xxd;
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.api.gax.grpc.GrpcInterceptorProvider;
+import com.google.api.gax.paging.Page;
+import com.google.cloud.ReadChannel;
+import com.google.cloud.ServiceOptions;
+import com.google.cloud.WriteChannel;
+import com.google.cloud.storage.Storage.BlobGetOption;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.Storage.BucketGetOption;
+import com.google.cloud.storage.Storage.BucketListOption;
+import com.google.cloud.storage.Storage.ComposeRequest;
+import com.google.cloud.storage.Storage.CopyRequest;
+import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.cloud.storage.it.AssertRequestHeaders;
+import com.google.cloud.storage.it.AssertRequestHeaders.FilteringPolicy;
+import com.google.cloud.storage.it.GrpcRequestAuditing;
+import com.google.cloud.storage.it.RequestAuditing;
+import com.google.cloud.storage.it.TemporaryBucket;
+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 com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.io.ByteStreams;
+import com.google.common.truth.IterableSubject;
+import io.grpc.ClientInterceptor;
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.time.Clock;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@CrossRun(
+    backends = {Backend.PROD},
+    transports = {Transport.HTTP, Transport.GRPC})
+public final class ITExtraHeadersOptionTest {
+
+  private static final String HEADER_NAME = "x-my-header";
+  private static final String HEADER_VALUE = "value";
+  private static final ImmutableMap<String, String> EXTRA_HEADERS =
+      ImmutableMap.of(HEADER_NAME, HEADER_VALUE);
+  @Inject public BucketInfo bucket;
+  @Inject public Generator generator;
+  @Inject public Storage baseStorage;
+  @Inject public Transport transport;
+
+  private Storage storage;
+  private AssertRequestHeaders headers;
+
+  @Before
+  public void setUp() throws Exception {
+    switch (transport) {
+      case HTTP:
+        RequestAuditing requestAuditing = new RequestAuditing();
+        headers = requestAuditing;
+        storage =
+            ((HttpStorageOptions) baseStorage.getOptions())
+                .toBuilder()
+                .setTransportOptions(requestAuditing)
+                // we're counting requests, disable retries so that if a request fails it won't
+                // show up as a bad assertion of the test itself
+                .setRetrySettings(ServiceOptions.getNoRetrySettings())
+                .build()
+                .getService();
+        break;
+      case GRPC:
+        GrpcRequestAuditing grpcRequestAuditing = new GrpcRequestAuditing();
+        headers = grpcRequestAuditing;
+        GrpcStorageOptions grpcStorageOptions = (GrpcStorageOptions) baseStorage.getOptions();
+        GrpcInterceptorProvider grpcInterceptorProvider =
+            grpcStorageOptions.getGrpcInterceptorProvider();
+        storage =
+            grpcStorageOptions
+                .toBuilder()
+                // we're counting requests, disable retries so that if a request fails it won't
+                // show up as a bad assertion of the test itself
+                .setRetrySettings(ServiceOptions.getNoRetrySettings())
+                .setGrpcInterceptorProvider(
+                    () -> {
+                      List<ClientInterceptor> interceptors =
+                          grpcInterceptorProvider.getInterceptors();
+                      return ImmutableList.<ClientInterceptor>builder()
+                          .addAll(interceptors)
+                          .add(grpcRequestAuditing)
+                          .build();
+                    })
+                .build()
+                .getService();
+        break;
+    }
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    if (storage != null) {
+      storage.close();
+    }
+  }
+
+  @Test
+  public void simpleUnary() throws Exception {
+    Bucket gen1 = storage.get(bucket.getName(), BucketGetOption.extraHeaders(EXTRA_HEADERS));
+
+    IterableSubject subject = headers.assertRequestHeader(HEADER_NAME, FilteringPolicy.NO_FILTER);
+    assertAll(() -> subject.containsExactly(HEADER_VALUE));
+  }
+
+  @Test
+  public void pageObjects() throws Exception {
+    String baseName = generator.randomObjectName();
+    Blob blob1 = storage.create(BlobInfo.newBuilder(bucket, baseName + "1").build());
+    Blob blob2 = storage.create(BlobInfo.newBuilder(bucket, baseName + "2").build());
+
+    headers.clear();
+    ImmutableList<String> expectedNames = ImmutableList.of(blob1.getName(), blob2.getName());
+    Page<Blob> page =
+        storage.list(
+            bucket.getName(),
+            BlobListOption.prefix(baseName),
+            BlobListOption.pageSize(1),
+            BlobListOption.extraHeaders(EXTRA_HEADERS));
+
+    List<String> collect = page.streamAll().map(BlobInfo::getName).collect(Collectors.toList());
+    IterableSubject subject = headers.assertRequestHeader(HEADER_NAME, FilteringPolicy.NO_FILTER);
+    assertAll(
+        () -> assertThat(collect).hasSize(2),
+        () -> assertThat(collect).containsExactlyElementsIn(expectedNames),
+        () -> subject.containsExactly(HEADER_VALUE, HEADER_VALUE));
+  }
+
+  @Test
+  public void pageBucket() throws Exception {
+    String baseName = generator.randomBucketName();
+    BucketInfo info1 = BucketInfo.of(baseName + "1");
+    BucketInfo info2 = BucketInfo.of(baseName + "2");
+    try (TemporaryBucket tmp1 =
+            TemporaryBucket.newBuilder().setBucketInfo(info1).setStorage(storage).build();
+        TemporaryBucket tmp2 =
+            TemporaryBucket.newBuilder().setBucketInfo(info2).setStorage(storage).build()) {
+      headers.clear();
+      Page<Bucket> page =
+          storage.list(
+              BucketListOption.prefix(baseName),
+              BucketListOption.pageSize(1),
+              BucketListOption.extraHeaders(EXTRA_HEADERS));
+
+      List<String> collect = page.streamAll().map(BucketInfo::getName).collect(Collectors.toList());
+      IterableSubject subject = headers.assertRequestHeader(HEADER_NAME, FilteringPolicy.NO_FILTER);
+      assertAll(
+          () -> assertThat(collect).containsExactly(info1.getName(), info2.getName()),
+          () -> subject.containsExactly(HEADER_VALUE, HEADER_VALUE));
+    }
+  }
+
+  @Test
+  public void readAllBytes() throws Exception {
+    byte[] expected = DataGenerator.base64Characters().genBytes(512 * 1024 + 45);
+    String expectedXxd = xxd(expected);
+
+    Blob gen1 =
+        storage.create(
+            BlobInfo.newBuilder(bucket, generator.randomObjectName()).build(),
+            expected,
+            BlobTargetOption.doesNotExist());
+
+    headers.clear();
+    byte[] actual =
+        storage.readAllBytes(
+            gen1.getBlobId(),
+            BlobSourceOption.generationMatch(),
+            BlobSourceOption.extraHeaders(EXTRA_HEADERS));
+    IterableSubject subject = headers.assertRequestHeader(HEADER_NAME, FilteringPolicy.NO_FILTER);
+    String actualXxd = xxd(actual);
+
+    assertAll(
+        () -> subject.containsExactly(HEADER_VALUE),
+        () -> assertThat(actualXxd).isEqualTo(expectedXxd));
+  }
+
+  @Test
+  public void reader() throws Exception {
+    byte[] expected = DataGenerator.base64Characters().genBytes(512 * 1024 + 45);
+    String expectedXxd = xxd(expected);
+
+    Blob gen1 =
+        storage.create(
+            BlobInfo.newBuilder(bucket, generator.randomObjectName()).build(),
+            expected,
+            BlobTargetOption.doesNotExist());
+
+    headers.clear();
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    try (ReadChannel reader =
+        storage.reader(
+            gen1.getBlobId(),
+            BlobSourceOption.generationMatch(),
+            BlobSourceOption.extraHeaders(EXTRA_HEADERS))) {
+      ByteStreams.copy(reader, Channels.newChannel(baos));
+    }
+    byte[] actual = baos.toByteArray();
+    IterableSubject subject = headers.assertRequestHeader(HEADER_NAME, FilteringPolicy.NO_FILTER);
+    String actualXxd = xxd(actual);
+
+    assertAll(
+        () -> subject.containsExactly(HEADER_VALUE),
+        () -> assertThat(actualXxd).isEqualTo(expectedXxd));
+  }
+
+  @Test
+  public void directUpload() throws Exception {
+    byte[] expected = DataGenerator.base64Characters().genBytes(512 * 1024 + 45);
+    String expectedXxd = xxd(expected);
+
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    headers.clear();
+    Blob gen1 =
+        storage.create(
+            info,
+            expected,
+            BlobTargetOption.doesNotExist(),
+            BlobTargetOption.extraHeaders(EXTRA_HEADERS));
+    IterableSubject subject = headers.assertRequestHeader(HEADER_NAME, FilteringPolicy.NO_FILTER);
+
+    byte[] actual = storage.readAllBytes(gen1.getBlobId(), BlobSourceOption.generationMatch());
+    String actualXxd = xxd(actual);
+
+    assertAll(
+        () -> subject.containsExactly(HEADER_VALUE),
+        () -> assertThat(actualXxd).isEqualTo(expectedXxd));
+  }
+
+  @Test
+  public void resumableUpload() throws Exception {
+    byte[] expected = DataGenerator.base64Characters().genBytes(512 * 1024 + 45);
+    String expectedXxd = xxd(expected);
+
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    try (WriteChannel writer =
+        storage.writer(
+            info, BlobWriteOption.doesNotExist(), BlobWriteOption.extraHeaders(EXTRA_HEADERS))) {
+      writer.setChunkSize(256 * 1024);
+      writer.write(ByteBuffer.wrap(Arrays.copyOfRange(expected, 0, 256 * 1024)));
+      writer.write(ByteBuffer.wrap(Arrays.copyOfRange(expected, 256 * 1024, 512 * 1024)));
+      writer.write(ByteBuffer.wrap(Arrays.copyOfRange(expected, 512 * 1024, expected.length)));
+    }
+    IterableSubject subject = headers.assertRequestHeader(HEADER_NAME, FilteringPolicy.NO_FILTER);
+
+    byte[] actual = storage.readAllBytes(info.getBlobId());
+    String actualXxd = xxd(actual);
+    assertAll(
+        () -> subject.containsExactly(HEADER_VALUE, HEADER_VALUE, HEADER_VALUE, HEADER_VALUE),
+        () -> assertThat(actualXxd).isEqualTo(expectedXxd));
+  }
+
+  @Test
+  @CrossRun.Exclude(transports = Transport.GRPC)
+  public void batch() throws Exception {
+    BlobInfo info1 = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    BlobInfo info2 = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    BlobInfo info3 = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    storage.create(info1, BlobTargetOption.doesNotExist());
+    storage.create(info2, BlobTargetOption.doesNotExist());
+    storage.create(info3, BlobTargetOption.doesNotExist());
+
+    headers.clear();
+    OffsetDateTime now = Clock.systemUTC().instant().atOffset(ZoneOffset.UTC);
+
+    StorageBatch batch = storage.batch();
+    String batchHeaderKey = "x-batch-key";
+    StorageBatchResult<Blob> r1 =
+        batch.get(
+            info1.getBlobId(), BlobGetOption.extraHeaders(ImmutableMap.of(batchHeaderKey, "v1")));
+    StorageBatchResult<Blob> r2 =
+        batch.update(
+            info2.toBuilder().setCustomTimeOffsetDateTime(now).build(),
+            BlobTargetOption.extraHeaders(ImmutableMap.of(batchHeaderKey, "v2")));
+    StorageBatchResult<Boolean> r3 =
+        batch.delete(
+            info3.getBlobId(),
+            BlobSourceOption.extraHeaders(ImmutableMap.of(batchHeaderKey, "v3")));
+
+    batch.submit();
+    assertAll(
+        () -> assertThat(r1).isNotNull(),
+        () -> assertThat(r2.get().getCustomTimeOffsetDateTime()).isEqualTo(now),
+        () -> assertThat(r3.get()).isTrue(),
+        () -> {
+          IterableSubject subject = headers.assertRequestHeader(batchHeaderKey);
+          subject.containsExactly("v1", "v2", "v3");
+        });
+  }
+
+  @Test
+  public void rewrite() throws Exception {
+    BlobInfo info1 = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    BlobInfo info2 = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    BlobInfo gen1 = storage.create(info1, BlobTargetOption.doesNotExist());
+
+    CopyRequest copyRequest =
+        CopyRequest.newBuilder()
+            .setSource(gen1.getBlobId())
+            .setSourceOptions(
+                BlobSourceOption.extraHeaders(ImmutableMap.of("x-header-source", HEADER_VALUE)))
+            .setTarget(
+                info2,
+                BlobTargetOption.doesNotExist(),
+                BlobTargetOption.extraHeaders(ImmutableMap.of("x-header-target", HEADER_VALUE)))
+            .build();
+    headers.clear();
+    CopyWriter copyWriter = storage.copy(copyRequest);
+    copyWriter.getResult();
+    assertAll(
+        () -> {
+          IterableSubject subject = headers.assertRequestHeader("x-header-source");
+          subject.containsExactly(HEADER_VALUE);
+        },
+        () -> {
+          IterableSubject subject = headers.assertRequestHeader("x-header-target");
+          subject.containsExactly(HEADER_VALUE);
+        });
+  }
+
+  @Test
+  public void compose() throws Exception {
+    BlobInfo info1 = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    BlobInfo info2 = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    BlobInfo gen1 = storage.create(info1, BlobTargetOption.doesNotExist());
+
+    ComposeRequest composeRequest =
+        ComposeRequest.newBuilder()
+            .addSource(gen1.getBlobId().getName())
+            .setTarget(info2)
+            .setTargetOptions(
+                BlobTargetOption.doesNotExist(), BlobTargetOption.extraHeaders(EXTRA_HEADERS))
+            .build();
+    headers.clear();
+    Blob blob2 = storage.compose(composeRequest);
+    assertThat(blob2).isNotNull();
+    IterableSubject subject = headers.assertRequestHeader(HEADER_NAME, FilteringPolicy.NO_FILTER);
+    subject.containsExactly(HEADER_VALUE);
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java
index d4c25edb..52e29637 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java
@@ -50,6 +50,7 @@ import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicLong;
+import org.checkerframework.checker.nullness.qual.NonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.junit.Before;
 import org.junit.Rule;
@@ -99,7 +100,7 @@ public final class ITJsonResumableSessionPutTaskTest {
       JsonResumableSessionPutTask task =
           new JsonResumableSessionPutTask(
               httpClientContext,
-              uploadUrl,
+              jsonResumableWrite(uploadUrl),
               RewindableContent.empty(),
               HttpContentRange.of(ByteRangeSpec.explicitClosed(0L, 0L), 0));
 
@@ -147,7 +148,7 @@ public final class ITJsonResumableSessionPutTaskTest {
       JsonResumableSessionPutTask task =
           new JsonResumableSessionPutTask(
               httpClientContext,
-              uploadUrl,
+              jsonResumableWrite(uploadUrl),
               RewindableContent.empty(),
               HttpContentRange.of(ByteRangeSpec.explicitClosed(0L, 10L)));
 
@@ -228,7 +229,7 @@ public final class ITJsonResumableSessionPutTaskTest {
       JsonResumableSessionPutTask task =
           new JsonResumableSessionPutTask(
               httpClientContext,
-              uploadUrl,
+              jsonResumableWrite(uploadUrl),
               RewindableContent.of(tmpFile.getPath()),
               HttpContentRange.of(ByteRangeSpec.explicit(0L, _256KiBL)));
 
@@ -297,7 +298,7 @@ public final class ITJsonResumableSessionPutTaskTest {
       JsonResumableSessionPutTask task =
           new JsonResumableSessionPutTask(
               httpClientContext,
-              uploadUrl,
+              jsonResumableWrite(uploadUrl),
               RewindableContent.empty(),
               HttpContentRange.of(_256KiBL));
 
@@ -366,7 +367,7 @@ public final class ITJsonResumableSessionPutTaskTest {
       JsonResumableSessionPutTask task =
           new JsonResumableSessionPutTask(
               httpClientContext,
-              uploadUrl,
+              jsonResumableWrite(uploadUrl),
               RewindableContent.empty(),
               HttpContentRange.of(_512KiBL));
 
@@ -444,7 +445,7 @@ public final class ITJsonResumableSessionPutTaskTest {
       JsonResumableSessionPutTask task =
           new JsonResumableSessionPutTask(
               httpClientContext,
-              uploadUrl,
+              jsonResumableWrite(uploadUrl),
               RewindableContent.empty(),
               HttpContentRange.of(_256KiBL));
 
@@ -526,7 +527,7 @@ public final class ITJsonResumableSessionPutTaskTest {
       JsonResumableSessionPutTask task =
           new JsonResumableSessionPutTask(
               httpClientContext,
-              uploadUrl,
+              jsonResumableWrite(uploadUrl),
               RewindableContent.empty(),
               HttpContentRange.of(_512KiBL));
 
@@ -606,7 +607,7 @@ public final class ITJsonResumableSessionPutTaskTest {
       JsonResumableSessionPutTask task =
           new JsonResumableSessionPutTask(
               httpClientContext,
-              uploadUrl,
+              jsonResumableWrite(uploadUrl),
               RewindableContent.empty(),
               HttpContentRange.of(_128KiBL));
 
@@ -684,7 +685,7 @@ public final class ITJsonResumableSessionPutTaskTest {
       JsonResumableSessionPutTask task =
           new JsonResumableSessionPutTask(
               httpClientContext,
-              uploadUrl,
+              jsonResumableWrite(uploadUrl),
               RewindableContent.of(tmpFile.getPath()),
               HttpContentRange.of(ByteRangeSpec.explicit(_512KiBL, _768KiBL)));
 
@@ -716,7 +717,7 @@ public final class ITJsonResumableSessionPutTaskTest {
       JsonResumableSessionPutTask task =
           new JsonResumableSessionPutTask(
               httpClientContext,
-              uploadUrl,
+              jsonResumableWrite(uploadUrl),
               RewindableContent.of(tmpFile.getPath()),
               HttpContentRange.of(ByteRangeSpec.explicit(_512KiBL, _768KiBL)));
 
@@ -756,7 +757,10 @@ public final class ITJsonResumableSessionPutTaskTest {
 
       JsonResumableSessionPutTask task =
           new JsonResumableSessionPutTask(
-              httpClientContext, uploadUrl, RewindableContent.empty(), HttpContentRange.of(0));
+              httpClientContext,
+              jsonResumableWrite(uploadUrl),
+              RewindableContent.empty(),
+              HttpContentRange.of(0));
 
       StorageException se = assertThrows(StorageException.class, task::call);
       // the parse error happens while trying to read the success object, make sure we raise it as
@@ -791,7 +795,10 @@ public final class ITJsonResumableSessionPutTaskTest {
 
       JsonResumableSessionPutTask task =
           new JsonResumableSessionPutTask(
-              httpClientContext, uploadUrl, RewindableContent.empty(), HttpContentRange.of(0));
+              httpClientContext,
+              jsonResumableWrite(uploadUrl),
+              RewindableContent.empty(),
+              HttpContentRange.of(0));
 
       ResumableOperationResult<@Nullable StorageObject> operationResult = task.call();
       StorageObject call = operationResult.getObject();
@@ -857,4 +864,8 @@ public final class ITJsonResumableSessionPutTaskTest {
     assertThat(buf1.remaining()).isEqualTo(0);
     assertThat(buf2.position()).isEqualTo(13);
   }
+
+  static @NonNull JsonResumableWrite jsonResumableWrite(String uploadUrl) {
+    return JsonResumableWrite.of(new StorageObject(), ImmutableMap.of(), uploadUrl, 0);
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionQueryTaskTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionQueryTaskTest.java
index d11cc340..8e2fb189 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionQueryTaskTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionQueryTaskTest.java
@@ -17,6 +17,7 @@
 package com.google.cloud.storage;
 
 import static com.google.cloud.storage.ByteSizeConstants._256KiBL;
+import static com.google.cloud.storage.ITJsonResumableSessionPutTaskTest.jsonResumableWrite;
 import static com.google.common.truth.Truth.assertThat;
 import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE;
 import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpResponseStatus.OK;
@@ -84,7 +85,7 @@ public final class ITJsonResumableSessionQueryTaskTest {
       String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
 
       JsonResumableSessionQueryTask task =
-          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+          new JsonResumableSessionQueryTask(httpClientContext, jsonResumableWrite(uploadUrl));
 
       ResumableOperationResult<@Nullable StorageObject> result = task.call();
       StorageObject object = result.getObject();
@@ -107,7 +108,7 @@ public final class ITJsonResumableSessionQueryTaskTest {
       String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
 
       JsonResumableSessionQueryTask task =
-          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+          new JsonResumableSessionQueryTask(httpClientContext, jsonResumableWrite(uploadUrl));
 
       ResumableOperationResult<@Nullable StorageObject> result = task.call();
       StorageObject object = result.getObject();
@@ -134,7 +135,7 @@ public final class ITJsonResumableSessionQueryTaskTest {
       String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
 
       JsonResumableSessionQueryTask task =
-          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+          new JsonResumableSessionQueryTask(httpClientContext, jsonResumableWrite(uploadUrl));
 
       ResumableOperationResult<@Nullable StorageObject> result = task.call();
       assertThat(result.getPersistedSize()).isEqualTo(_256KiBL);
@@ -150,7 +151,7 @@ public final class ITJsonResumableSessionQueryTaskTest {
       String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
 
       JsonResumableSessionQueryTask task =
-          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+          new JsonResumableSessionQueryTask(httpClientContext, jsonResumableWrite(uploadUrl));
 
       ResumableOperationResult<@Nullable StorageObject> result = task.call();
       assertThat(result.getPersistedSize()).isEqualTo(0);
@@ -167,7 +168,7 @@ public final class ITJsonResumableSessionQueryTaskTest {
       String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
 
       JsonResumableSessionQueryTask task =
-          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+          new JsonResumableSessionQueryTask(httpClientContext, jsonResumableWrite(uploadUrl));
 
       StorageException se = assertThrows(StorageException.class, task::call);
       assertThat(se.getCode()).isEqualTo(0);
@@ -193,7 +194,7 @@ public final class ITJsonResumableSessionQueryTaskTest {
       String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
 
       JsonResumableSessionQueryTask task =
-          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+          new JsonResumableSessionQueryTask(httpClientContext, jsonResumableWrite(uploadUrl));
 
       StorageException se = assertThrows(StorageException.class, task::call);
       assertThat(se.getCode()).isEqualTo(0);
@@ -221,7 +222,7 @@ public final class ITJsonResumableSessionQueryTaskTest {
       String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
 
       JsonResumableSessionQueryTask task =
-          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+          new JsonResumableSessionQueryTask(httpClientContext, jsonResumableWrite(uploadUrl));
 
       StorageException se = assertThrows(StorageException.class, task::call);
       assertThat(se.getCode()).isEqualTo(0);
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java
index 2163c578..11069cba 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java
@@ -21,10 +21,12 @@ import static org.junit.Assert.assertThrows;
 
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.cloud.storage.Storage.PredefinedAcl;
+import com.google.cloud.storage.UnifiedOpts.Headers;
 import com.google.cloud.storage.UnifiedOpts.Mapper;
 import com.google.cloud.storage.UnifiedOpts.NoOpObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
+import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.hash.Hashing;
@@ -1303,6 +1305,102 @@ public final class UnifiedOptsGrpcTest {
     }
   }
 
+  @SuppressWarnings("unchecked")
+  public static final class HeadersOptTest {
+
+    @Test
+    public void duplicateHeaderValues_firstWins_grpc() {
+      Opts<Headers> o1 = Opts.from(UnifiedOpts.extraHeaders(ImmutableMap.of("k", "1")));
+      Opts<Headers> o2 = Opts.from(UnifiedOpts.extraHeaders(ImmutableMap.of("k", "2")));
+
+      Mapper<GrpcCallContext> mapper = o1.grpcMetadataMapper().andThen(o2.grpcMetadataMapper());
+
+      GrpcCallContext grpcCallContext = mapper.apply(GrpcCallContext.createDefault());
+      assertThat(grpcCallContext.getExtraHeaders())
+          .containsExactlyEntriesIn(ImmutableMap.of("k", ImmutableList.of("1")));
+    }
+
+    @Test
+    public void duplicateHeaderValues_firstWins_rpcOptions() {
+      Opts<Headers> o1 =
+          Opts.from(
+              UnifiedOpts.extraHeaders(ImmutableMap.of("k", "1", "a", "A")),
+              UnifiedOpts.extraHeaders(ImmutableMap.of("k", "2", "b", "B")));
+
+      ImmutableMap<StorageRpc.Option, ?> rpcOptions = o1.getRpcOptions();
+
+      ImmutableMap<String, String> extraHeaders =
+          (ImmutableMap<String, String>) rpcOptions.get(StorageRpc.Option.EXTRA_HEADERS);
+      assertThat(extraHeaders).isNotNull();
+      assertThat(extraHeaders)
+          .containsExactlyEntriesIn(ImmutableMap.of("k", "1", "a", "A", "b", "B"));
+    }
+
+    @Test
+    public void duplicateHeaderValues_firstWins_keyComparisonIsCaseInsensitive() {
+      Opts<Headers> o1 =
+          Opts.from(
+              UnifiedOpts.extraHeaders(ImmutableMap.of("K", "1", "a", "A")),
+              UnifiedOpts.extraHeaders(ImmutableMap.of("k", "2", "b", "B")));
+
+      ImmutableMap<StorageRpc.Option, ?> rpcOptions = o1.getRpcOptions();
+
+      ImmutableMap<String, String> extraHeaders =
+          (ImmutableMap<String, String>) rpcOptions.get(StorageRpc.Option.EXTRA_HEADERS);
+      assertThat(extraHeaders).isNotNull();
+      assertThat(extraHeaders)
+          .containsExactlyEntriesIn(ImmutableMap.of("k", "1", "a", "A", "b", "B"));
+    }
+
+    @Test
+    public void headersOnBlocklistResultInIllegalArgumentException() {
+      IllegalArgumentException expected =
+          assertThrows(
+              IllegalArgumentException.class,
+              () ->
+                  UnifiedOpts.extraHeaders(
+                      ImmutableMap.<String, String>builder()
+                          .put("Accept-Encoding", "a")
+                          .put("Cache-Control", "a")
+                          .put("Connection", "a")
+                          .put("Content-ID", "a")
+                          .put("Content-Length", "a")
+                          .put("Content-Range", "a")
+                          .put("Content-Transfer-Encoding", "a")
+                          .put("Content-Type", "a")
+                          .put("Date", "a")
+                          .put("ETag", "a")
+                          .put("If-Match", "a")
+                          .put("If-None-Match", "a")
+                          .put("Keep-Alive", "a")
+                          .put("Range", "a")
+                          .put("TE", "a")
+                          .put("Trailer", "a")
+                          .put("Transfer-Encoding", "a")
+                          .put("User-Agent", "a")
+                          .put("X-Goog-Api-Client", "a")
+                          .put("X-Goog-Content-Length-Range", "a")
+                          .put("X-Goog-Copy-Source-Encryption-Algorithm", "a")
+                          .put("X-Goog-Copy-Source-Encryption-Key", "a")
+                          .put("X-Goog-Copy-Source-Encryption-Key-Sha256", "a")
+                          .put("X-Goog-Encryption-Algorithm", "a")
+                          .put("X-Goog-Encryption-Key", "a")
+                          .put("X-Goog-Encryption-Key-Sha256", "a")
+                          .put("X-Goog-Meta-A", "a")
+                          .put("X-Goog-Request-Params", "a")
+                          .put("X-Goog-User-Project", "a")
+                          .put("X-HTTP-Method-Override", "a")
+                          .put("X-Upload-Content-Length", "a")
+                          .put("X-Upload-Content-Type", "a")
+                          .put("A", "a")
+                          .build()));
+      assertThat(expected)
+          .hasMessageThat()
+          .contains(
+              "[accept-encoding, cache-control, connection, content-id, content-length, content-range, content-transfer-encoding, content-type, date, etag, if-match, if-none-match, keep-alive, range, te, trailer, transfer-encoding, user-agent, x-goog-api-client, x-goog-content-length-range, x-goog-copy-source-encryption-algorithm, x-goog-copy-source-encryption-key, x-goog-copy-source-encryption-key-sha256, x-goog-encryption-algorithm, x-goog-encryption-key, x-goog-encryption-key-sha256, x-goog-meta-a, x-goog-request-params, x-goog-user-project, x-http-method-override, x-upload-content-length, x-upload-content-type]");
+    }
+  }
+
   private static CommonObjectRequestParams randEncryption() {
     byte[] bytes = new byte[32];
     new SecureRandom().nextBytes(bytes);
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/AssertRequestHeaders.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/AssertRequestHeaders.java
new file mode 100644
index 00000000..a9053ea8
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/AssertRequestHeaders.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2025 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 com.google.common.truth.IterableSubject;
+
+public interface AssertRequestHeaders {
+
+  void clear();
+
+  default IterableSubject assertRequestHeader(String headerName) {
+    return assertRequestHeader(headerName, FilteringPolicy.DISTINCT);
+  }
+
+  IterableSubject assertRequestHeader(String headerName, FilteringPolicy filteringPolicy);
+
+  enum FilteringPolicy {
+    DISTINCT,
+    NO_FILTER
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/GrpcRequestAuditing.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/GrpcRequestAuditing.java
index a674a079..464899bc 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/GrpcRequestAuditing.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/GrpcRequestAuditing.java
@@ -33,16 +33,19 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.function.Function;
+import java.util.stream.Stream;
+import org.checkerframework.checker.nullness.qual.NonNull;
 
-final class GrpcRequestAuditing implements ClientInterceptor {
+public final class GrpcRequestAuditing implements ClientInterceptor, AssertRequestHeaders {
 
   private final List<Metadata> requestHeaders;
 
-  GrpcRequestAuditing() {
+  public GrpcRequestAuditing() {
     requestHeaders = Collections.synchronizedList(new ArrayList<>());
   }
 
-  void clear() {
+  public void clear() {
     requestHeaders.clear();
   }
 
@@ -53,13 +56,31 @@ final class GrpcRequestAuditing implements ClientInterceptor {
     return next.newCall(method, withStreamTracerFactory);
   }
 
+  @Override
+  public IterableSubject assertRequestHeader(String headerName, FilteringPolicy filteringPolicy) {
+    Metadata.Key<String> key = Metadata.Key.of(headerName, Metadata.ASCII_STRING_MARSHALLER);
+    Function<Stream<String>, Stream<String>> filter;
+    switch (filteringPolicy) {
+      case DISTINCT:
+        filter = Stream::distinct;
+        break;
+      case NO_FILTER:
+        filter = Function.identity();
+        break;
+      default:
+        throw new IllegalStateException("Unhandled enum value: " + filteringPolicy);
+    }
+    return getIterableSubject(key, filter);
+  }
+
   public <T> IterableSubject assertRequestHeader(Metadata.Key<T> key) {
-    ImmutableList<Object> actual =
-        requestHeaders.stream()
-            .map(m -> m.get(key))
-            .filter(Objects::nonNull)
-            .distinct()
-            .collect(ImmutableList.toImmutableList());
+    return getIterableSubject(key, Stream::distinct);
+  }
+
+  private <T> @NonNull IterableSubject getIterableSubject(
+      Metadata.Key<T> key, Function<Stream<T>, Stream<T>> f) {
+    Stream<T> stream = requestHeaders.stream().map(m -> m.get(key)).filter(Objects::nonNull);
+    ImmutableList<Object> actual = f.apply(stream).collect(ImmutableList.toImmutableList());
     return assertWithMessage(String.format("Headers %s", key.name())).that(actual);
   }
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/RequestAuditing.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/RequestAuditing.java
index bbb1821c..a802c0de 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/RequestAuditing.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/RequestAuditing.java
@@ -38,12 +38,13 @@ import java.util.List;
 import java.util.Objects;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
-final class RequestAuditing extends HttpTransportOptions {
+public final class RequestAuditing extends HttpTransportOptions implements AssertRequestHeaders {
 
   private final List<HttpRequest> requests;
 
-  RequestAuditing() {
+  public RequestAuditing() {
     super(HttpTransportOptions.newBuilder());
     requests = Collections.synchronizedList(new ArrayList<>());
   }
@@ -61,7 +62,7 @@ final class RequestAuditing extends HttpTransportOptions {
     return ImmutableList.copyOf(requests);
   }
 
-  void clear() {
+  public void clear() {
     requests.clear();
   }
 
@@ -196,16 +197,28 @@ final class RequestAuditing extends HttpTransportOptions {
         .isEqualTo(ImmutableList.of(expectedValue));
   }
 
-  IterableSubject assertRequestHeader(String headerName) {
+  @Override
+  public IterableSubject assertRequestHeader(String headerName, FilteringPolicy filteringPolicy) {
+    Function<Stream<Object>, Stream<Object>> filter;
+    switch (filteringPolicy) {
+      case DISTINCT:
+        filter = Stream::distinct;
+        break;
+      case NO_FILTER:
+        filter = Function.identity();
+        break;
+      default:
+        throw new IllegalStateException("Unhandled enum value: " + filteringPolicy);
+    }
+
     ImmutableList<HttpRequest> requests = getRequests();
 
-    List<Object> actual =
+    Stream<Object> stream =
         requests.stream()
             .map(HttpRequest::getHeaders)
             .map(headers -> headers.get(headerName))
-            .filter(Objects::nonNull)
-            .distinct()
-            .collect(Collectors.toList());
+            .filter(Objects::nonNull);
+    List<Object> actual = filter.apply(stream).collect(Collectors.toList());
 
     return assertWithMessage(String.format("Headers %s", headerName)).that(actual);
   }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/StorageITRunner.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/StorageITRunner.java
index 814c5da6..d99df4b8 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/StorageITRunner.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/StorageITRunner.java
@@ -42,6 +42,8 @@ import org.junit.runners.Suite;
 import org.junit.runners.model.FrameworkField;
 import org.junit.runners.model.InitializationError;
 import org.junit.runners.model.TestClass;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Storage custom runner which will handle {@link CrossRun}, {@link SingleBackend}, {@link
@@ -53,6 +55,12 @@ import org.junit.runners.model.TestClass;
  * @see org.junit.runners.BlockJUnit4ClassRunner
  */
 public final class StorageITRunner extends Suite {
+  static {
+    org.slf4j.bridge.SLF4JBridgeHandler.removeHandlersForRootLogger();
+    org.slf4j.bridge.SLF4JBridgeHandler.install();
+  }
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(StorageITRunner.class);
 
   private final Lock childrenLock = new ReentrantLock();
   private volatile ImmutableList<Runner> filteredChildren = null;
@@ -67,6 +75,7 @@ public final class StorageITRunner extends Suite {
 
   @Override
   public void run(RunNotifier notifier) {
+    LOGGER.debug("run(notifier : {})", notifier);
     super.run(new RunNotifierUnion(notifier, Registry.getInstance()));
   }
 
diff --git a/google-cloud-storage/src/test/resources/logback.xml b/google-cloud-storage/src/test/resources/logback.xml
new file mode 100644
index 00000000..70cdedad
--- /dev/null
+++ b/google-cloud-storage/src/test/resources/logback.xml
@@ -0,0 +1,93 @@
+<!--
+  ~ Copyright 2025 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.
+  -->
+
+<configuration>
+  <contextListener class="ch.qos.logback.classic.jul.LevelChangePropagator">
+    <resetJUL>true</resetJUL>
+  </contextListener>
+  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder>
+      <pattern>%date %-5.5level [%-24.24thread] %-45.45logger{45} - %message%n</pattern>
+    </encoder>
+  </appender>
+  <appender name="NOEXCEPTION" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder>
+      <pattern>%date %-5.5level [%-24.24thread] %-45.45logger{45} - %message%nopex%n</pattern>
+    </encoder>
+  </appender>
+  <logger name="com.google.auth.oauth2.ComputeEngineCredentials" level="info"/>
+  <logger name="io.opencensus" level="info"/>
+  <logger name="io.grpc.ChannelLogger" level="info"/>
+  <!-- Context logs an exception when it's not able to find things, so suppress it's exception -->
+  <logger name="io.grpc.Context" level="info" additivity="false">
+    <appender-ref ref="NOEXCEPTION"/>
+  </logger>
+  <logger name="io.grpc.LoadBalancerRegistry" level="warn"/>
+  <logger name="io.grpc.NameResolverRegistry" level="warn"/>
+  <!--
+  GrpclbNameResolver logs an exception with several of it's messages that aren't
+  useful when you're only interested in the traffic
+  -->
+  <logger name="io.grpc.grpclb.GrpclbNameResolver" level="info" additivity="false">
+    <appender-ref ref="NOEXCEPTION"/>
+  </logger>
+  <logger name="io.grpc.ManagedChannelRegistry" level="info" additivity="false">
+    <appender-ref ref="NOEXCEPTION"/>
+  </logger>
+  <logger name="io.grpc.internal.ManagedChannelImplBuilder" level="info" additivity="false">
+    <appender-ref ref="NOEXCEPTION"/>
+  </logger>
+  <logger name="io.grpc.internal.AbstractManagedChannelImplBuilder" level="warn"/>
+  <logger name="io.grpc.internal.DnsNameResolver" level="warn"/>
+  <!--
+  netty initialization and event loop management appenders,
+  not necessarily what is desired when debugging requests
+  -->
+  <logger name="io.grpc.netty.shaded.io.grpc.netty" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty.buffer" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty.channel.DefaultChannelId" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty.channel.MultithreadEventLoopGroup" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty.handler.ssl" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty.util.NetUtil" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty.util.Recycler" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty.util.ResourceLeakDetector" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty.util.ResourceLeakDetectorFactory" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty.util.internal.CleanerJava6" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty.util.internal.InternalThreadLocalMap" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty.util.internal.NativeLibraryLoader" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty.util.internal.PlatformDependent" level="warn"/>
+  <logger name="io.grpc.netty.shaded.io.netty.util.internal.PlatformDependent0" level="warn"/>
+  <logger name="io.perfmark.PerfMark" level="warn"/>
+  <!--
+  jdk11 appender which includes lots of details about ssl cypher suites
+  -->
+  <logger name="jdk.event.security" level="warn"/>
+  <!-- http request logging -->
+  <logger name="sun.net.www.protocol.http.HttpURLConnection" level="warn"/>
+  <logger name="com.google.api.client.http.HttpTransport" level="warn"/>
+  <logger name="com.google.cloud.storage.it.GrpcPlainRequestLoggingInterceptor" level="warn"/>
+  <!-- grpc netty traffic logging -->
+  <!--<logger name="io.grpc.netty.shaded.io.netty.handler.ssl.SslHandler" level="debug"/>-->
+  <!--<logger name="io.grpc.netty.shaded.io.grpc.netty.NettyClientHandler" level="debug"/>-->
+
+  <logger name="com.google.cloud.storage" level="info"/>
+  <logger name="com.google.cloud.storage.it" level="warn"/>
+
+  <root level="info">
+    <appender-ref ref="STDOUT"/>
+  </root>
+</configuration>