Repository: googleapis/java-storage
Pull request #2072 (2023-06-16 00:27:34)

deps: update dependencies io.grpc:* to v1.56.0


Patch:
diff --git a/.github/.OwlBot.lock.yaml b/.github/.OwlBot.lock.yaml
index aadf54f6..73568a1e 100644
--- a/.github/.OwlBot.lock.yaml
+++ b/.github/.OwlBot.lock.yaml
@@ -13,4 +13,5 @@
 # limitations under the License.
 docker:
   image: gcr.io/cloud-devrel-public-resources/owlbot-java:latest
-  digest: sha256:ad9cabee4c022f1aab04a71332369e0c23841062124818a4490f73337f790337
+  digest: sha256:af2eda87a54601ae7b7b2be5055c17b43ac98a7805b586772db314de8a7d4a1d
+# created: 2023-06-16T02:10:09.149325782Z
diff --git a/.github/dependabot.yml b/.github/dependabot.yml
index c8f413b0..fde1ced4 100644
--- a/.github/dependabot.yml
+++ b/.github/dependabot.yml
@@ -5,10 +5,13 @@ updates:
     schedule:
       interval: "daily"
     # Disable version updates for Maven dependencies
-    open-pull-requests-limit: 0
+    # we use renovate-bot as well as shared-dependencies BOM to update maven dependencies.
+    ignore: "*"
   - package-ecosystem: "pip"
     directory: "/"
     schedule:
       interval: "daily"
     # Disable version updates for pip dependencies
-    open-pull-requests-limit: 0
\ No newline at end of file
+    # If a security vulnerability comes in, we will be notified about 
+    # it via template in the synthtool repository.
+    ignore: "*"
diff --git a/.github/release-please.yml b/.github/release-please.yml
index d71d8f4e..1280ad03 100644
--- a/.github/release-please.yml
+++ b/.github/release-please.yml
@@ -1,7 +1,8 @@
 bumpMinorPreMajor: true
 handleGHRelease: true
 releaseType: java-yoshi
-extraFiles: ["README.md"]
+extraFiles:
+  - README.md
 branches:
   - bumpMinorPreMajor: true
     handleGHRelease: true
@@ -31,3 +32,9 @@ branches:
     handleGHRelease: true
     releaseType: java-backport
     branch: 2.15.x
+  - bumpMinorPreMajor: true
+    handleGHRelease: true
+    releaseType: java-backport
+    extraFiles:
+      - README.md
+    branch: 2.22.x
diff --git a/.github/sync-repo-settings.yaml b/.github/sync-repo-settings.yaml
index b7d5dcb3..ec449b6a 100644
--- a/.github/sync-repo-settings.yaml
+++ b/.github/sync-repo-settings.yaml
@@ -129,6 +129,23 @@ branchProtectionRules:
       - cla/google
       - 'Kokoro - Test: Java GraalVM Native Image'
       - 'Kokoro - Test: Java 17 GraalVM Native Image'
+  - pattern: 2.22.x
+    isAdminEnforced: true
+    requiredApprovingReviewCount: 1
+    requiresCodeOwnerReviews: true
+    requiresStrictStatusChecks: false
+    requiredStatusCheckContexts:
+      - dependencies (8)
+      - dependencies (11)
+      - lint
+      - clirr
+      - units (8)
+      - units (11)
+      - 'Kokoro - Test: Integration'
+      - cla/google
+      - OwlBot Post Processor
+      - 'Kokoro - Test: Java GraalVM Native Image'
+      - 'Kokoro - Test: Java 17 GraalVM Native Image'
 permissionRules:
   - team: yoshi-admins
     permission: admin
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java
index 15d3c489..4bba8aef 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java
@@ -40,6 +40,7 @@ import java.nio.file.Path;
 import java.security.Key;
 import java.time.OffsetDateTime;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -163,6 +164,40 @@ public class Blob extends BlobInfo {
       return new BlobSourceOption(UnifiedOpts.returnRawInputStream(shouldReturnRawInputStream));
     }
 
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static BlobSourceOption[] dedupe(BlobSourceOption... os) {
+      return Option.dedupe(BlobSourceOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobSourceOption[] dedupe(
+        Collection<BlobSourceOption> collection, BlobSourceOption... os) {
+      return Option.dedupe(BlobSourceOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobSourceOption[] dedupe(BlobSourceOption[] array, BlobSourceOption... os) {
+      return Option.dedupe(BlobSourceOption[]::new, array, os);
+    }
+
     static Storage.BlobSourceOption[] toSourceOptions(
         BlobInfo blobInfo, BlobSourceOption... options) {
       Storage.BlobSourceOption[] convertedOptions = new Storage.BlobSourceOption[options.length];
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Bucket.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Bucket.java
index af5c2cc2..c8827c1f 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Bucket.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Bucket.java
@@ -30,7 +30,6 @@ import com.google.cloud.storage.UnifiedOpts.BucketOptExtractor;
 import com.google.cloud.storage.UnifiedOpts.BucketSourceOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectOptExtractor;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
-import com.google.cloud.storage.UnifiedOpts.OptionShim;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
 import java.io.IOException;
@@ -41,6 +40,7 @@ import java.security.Key;
 import java.time.Duration;
 import java.time.OffsetDateTime;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -99,6 +99,41 @@ public class Bucket extends BucketInfo {
       return new BucketSourceOption(UnifiedOpts.userProject(userProject));
     }
 
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static BucketSourceOption[] dedupe(BucketSourceOption... os) {
+      return Option.dedupe(BucketSourceOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BucketSourceOption[] dedupe(
+        Collection<BucketSourceOption> collection, BucketSourceOption... os) {
+      return Option.dedupe(BucketSourceOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BucketSourceOption[] dedupe(
+        BucketSourceOption[] array, BucketSourceOption... os) {
+      return Option.dedupe(BucketSourceOption[]::new, array, os);
+    }
+
     static Storage.BucketSourceOption[] toSourceOptions(
         BucketInfo bucketInfo, BucketSourceOption... options) {
       Storage.BucketSourceOption[] convertedOptions =
@@ -237,6 +272,40 @@ public class Bucket extends BucketInfo {
       return new BlobTargetOption(UnifiedOpts.userProject(userProject));
     }
 
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static BlobTargetOption[] dedupe(BlobTargetOption... os) {
+      return Option.dedupe(BlobTargetOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobTargetOption[] dedupe(
+        Collection<BlobTargetOption> collection, BlobTargetOption... os) {
+      return Option.dedupe(BlobTargetOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobTargetOption[] dedupe(BlobTargetOption[] array, BlobTargetOption... os) {
+      return Option.dedupe(BlobTargetOption[]::new, array, os);
+    }
+
     static Storage.BlobTargetOption[] toTargetOptions(
         BlobInfo blobInfo, BlobTargetOption... options) {
       Storage.BlobTargetOption[] targetOptions = new Storage.BlobTargetOption[options.length];
@@ -255,7 +324,7 @@ public class Bucket extends BucketInfo {
   }
 
   /** Class for specifying blob write options when {@code Bucket} methods are used. */
-  public static class BlobWriteOption extends OptionShim<ObjectTargetOpt> implements Serializable {
+  public static class BlobWriteOption extends Option<ObjectTargetOpt> implements Serializable {
 
     private static final long serialVersionUID = 59762268190041584L;
 
@@ -366,6 +435,40 @@ public class Bucket extends BucketInfo {
       return new BlobWriteOption(UnifiedOpts.userProject(userProject));
     }
 
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static BlobWriteOption[] dedupe(BlobWriteOption... os) {
+      return Option.dedupe(BlobWriteOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobWriteOption[] dedupe(
+        Collection<BlobWriteOption> collection, BlobWriteOption... os) {
+      return Option.dedupe(BlobWriteOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobWriteOption[] dedupe(BlobWriteOption[] array, BlobWriteOption... os) {
+      return Option.dedupe(BlobWriteOption[]::new, array, os);
+    }
+
     static Storage.BlobWriteOption[] toWriteOptions(BlobInfo blobInfo, BlobWriteOption... options) {
       Storage.BlobWriteOption[] convertedOptions = new Storage.BlobWriteOption[options.length];
       for (int i = 0; i < options.length; i++) {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java
index d1e4e2f0..cae70d67 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java
@@ -16,6 +16,7 @@
 
 package com.google.cloud.storage;
 
+import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.RestorableState;
@@ -78,7 +79,9 @@ final class GapicCopyWriter extends CopyWriter {
           RewriteObjectRequest.newBuilder()
               .setRewriteToken(mostRecentResponse.getRewriteToken())
               .build();
-      mostRecentResponse = Retrying.run(options, alg, () -> callable.call(req), Decoder.identity());
+      GrpcCallContext retryContext = Retrying.newCallContext();
+      mostRecentResponse =
+          Retrying.run(options, alg, () -> callable.call(req, retryContext), Decoder.identity());
     }
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
index 329851da..a8725609 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
@@ -281,7 +281,8 @@ final class GapicWritableByteChannelSessionBuilder {
         return new UnbufferedWriteSession<>(
             requireNonNull(start, "start must be non null"),
             bindFunction(
-                    WriteFlushStrategy.fsyncEveryFlush(write, deps, alg), ResumableWrite::identity)
+                    WriteFlushStrategy.fsyncEveryFlush(write, deps, alg, Retrying::newCallContext),
+                    ResumableWrite::identity)
                 .andThen(StorageByteChannels.writable()::createSynchronized));
       }
     }
@@ -309,7 +310,8 @@ final class GapicWritableByteChannelSessionBuilder {
         return new BufferedWriteSession<>(
             requireNonNull(start, "start must be non null"),
             bindFunction(
-                    WriteFlushStrategy.fsyncEveryFlush(write, deps, alg), ResumableWrite::identity)
+                    WriteFlushStrategy.fsyncEveryFlush(write, deps, alg, Retrying::newCallContext),
+                    ResumableWrite::identity)
                 .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
                 .andThen(StorageByteChannels.writable()::createSynchronized));
       }
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 e8faef4d..a4164eb2 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
@@ -205,10 +205,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .setBucketId(bucketInfo.getName())
             .setParent("projects/_");
     CreateBucketRequest req = opts.createBucketsRequest().apply(builder).build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.createBucketCallable().call(req, grpcCallContext),
+        () -> storageClient.createBucketCallable().call(req, merge),
         syntaxDecoders.bucket);
   }
 
@@ -233,6 +234,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
     Hasher hasher = getHasherForRequest(req, Hasher.enabled());
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
@@ -240,8 +242,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
           UnbufferedWritableByteChannelSession<WriteObjectResponse> session =
               ResumableMedia.gapic()
                   .write()
-                  .byteChannel(
-                      storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
+                  .byteChannel(storageClient.writeObjectCallable().withDefaultCallContext(merge))
                   .setByteStringStrategy(ByteStringStrategy.noCopy())
                   .setHasher(hasher)
                   .direct()
@@ -286,23 +287,21 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
 
     Hasher hasher = getHasherForRequest(req, Hasher.enabled());
-    GapicWritableByteChannelSessionBuilder channelSessionBuilder =
-        ResumableMedia.gapic()
-            .write()
-            .byteChannel(
-                storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
-            .setHasher(hasher)
-            .setByteStringStrategy(ByteStringStrategy.noCopy());
 
     long size = Files.size(path);
     if (size < bufferSize) {
       // ignore the bufferSize argument if the file is smaller than it
+      GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
       return Retrying.run(
           getOptions(),
           retryAlgorithmManager.getFor(req),
           () -> {
             BufferedWritableByteChannelSession<WriteObjectResponse> session =
-                channelSessionBuilder
+                ResumableMedia.gapic()
+                    .write()
+                    .byteChannel(storageClient.writeObjectCallable().withDefaultCallContext(merge))
+                    .setHasher(hasher)
+                    .setByteStringStrategy(ByteStringStrategy.noCopy())
                     .direct()
                     .buffered(Buffers.allocate(size))
                     .setRequest(req)
@@ -320,7 +319,12 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     } else {
       ApiFuture<ResumableWrite> start = startResumableWrite(grpcCallContext, req);
       BufferedWritableByteChannelSession<WriteObjectResponse> session =
-          channelSessionBuilder
+          ResumableMedia.gapic()
+              .write()
+              .byteChannel(
+                  storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
+              .setHasher(hasher)
+              .setByteStringStrategy(ByteStringStrategy.noCopy())
               .resumable()
               .withRetryConfig(getOptions(), retryAlgorithmManager.idempotent())
               .buffered(Buffers.allocateAligned(bufferSize, _256KiB))
@@ -396,10 +400,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .setBucket(bucketNameCodec.encode(bucket.getName()));
     LockBucketRetentionPolicyRequest req =
         opts.lockBucketRetentionPolicyRequest().apply(builder).build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.lockBucketRetentionPolicyCallable().call(req, grpcCallContext),
+        () -> storageClient.lockBucketRetentionPolicyCallable().call(req, merge),
         syntaxDecoders.bucket);
   }
 
@@ -431,10 +436,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .apply(ListBucketsRequest.newBuilder())
             .build();
     try {
+      GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
       return Retrying.run(
           getOptions(),
           retryAlgorithmManager.getFor(request),
-          () -> storageClient.listBucketsPagedCallable().call(request, grpcCallContext),
+          () -> storageClient.listBucketsPagedCallable().call(request, merge),
           resp ->
               new TransformingPageDecorator<>(
                   resp.getPage(),
@@ -455,10 +461,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         ListObjectsRequest.newBuilder().setParent(bucketNameCodec.encode(bucket));
     ListObjectsRequest req = opts.listObjectsRequest().apply(builder).build();
     try {
+      GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
       return Retrying.run(
           getOptions(),
           retryAlgorithmManager.getFor(req),
-          () -> storageClient.listObjectsCallable().call(req, grpcCallContext),
+          () -> storageClient.listObjectsCallable().call(req, merge),
           resp -> new ListObjectsWithSyntheticDirectoriesPage(grpcCallContext, req, resp));
     } catch (Exception e) {
       throw StorageException.coalesce(e);
@@ -484,10 +491,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
                 .map(NamedField::getGrpcName)
                 .collect(ImmutableList.toImmutableList()));
     UpdateBucketRequest req = builder.build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.updateBucketCallable().call(req, grpcCallContext),
+        () -> storageClient.updateBucketCallable().call(req, merge),
         syntaxDecoders.bucket);
   }
 
@@ -510,10 +518,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
                 .map(NamedField::getGrpcName)
                 .collect(ImmutableList.toImmutableList()));
     UpdateObjectRequest req = builder.build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.updateObjectCallable().call(req, grpcCallContext),
+        () -> storageClient.updateObjectCallable().call(req, merge),
         syntaxDecoders.blob);
   }
 
@@ -531,10 +540,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         DeleteBucketRequest.newBuilder().setName(bucketNameCodec.encode(bucket));
     DeleteBucketRequest req = opts.deleteBucketsRequest().apply(builder).build();
     try {
+      GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
       Retrying.run(
           getOptions(),
           retryAlgorithmManager.getFor(req),
-          () -> storageClient.deleteBucketCallable().call(req, grpcCallContext),
+          () -> storageClient.deleteBucketCallable().call(req, merge),
           Decoder.identity());
       return true;
     } catch (StorageException e) {
@@ -558,13 +568,14 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .setObject(blob.getName());
     ifNonNull(blob.getGeneration(), builder::setGeneration);
     DeleteObjectRequest req = opts.deleteObjectsRequest().apply(builder).build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Boolean.TRUE.equals(
         Retrying.run(
             getOptions(),
             retryAlgorithmManager.getFor(req),
             () -> {
               try {
-                storageClient.deleteObjectCallable().call(req, grpcCallContext);
+                storageClient.deleteObjectCallable().call(req, merge);
                 return true;
               } catch (NotFoundException e) {
                 return false;
@@ -593,10 +604,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     final Object target = codecs.blobInfo().encode(composeRequest.getTarget());
     builder.setDestination(target);
     ComposeObjectRequest req = opts.composeObjectsRequest().apply(builder).build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.composeObjectCallable().call(req, grpcCallContext),
+        () -> storageClient.composeObjectCallable().call(req, merge),
         syntaxDecoders.blob);
   }
 
@@ -646,10 +658,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         srcOpts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     UnaryCallable<RewriteObjectRequest, RewriteResponse> callable =
         storageClient.rewriteObjectCallable().withDefaultCallContext(grpcCallContext);
+    GrpcCallContext retryContext = Retrying.newCallContext();
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> callable.call(req),
+        () -> callable.call(req, retryContext),
         (resp) -> new GapicCopyWriter(this, callable, retryAlgorithmManager.idempotent(), resp));
   }
 
@@ -687,7 +700,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob).prepend(defaultOpts);
     ReadObjectRequest request = getReadObjectRequest(blob, opts);
     Set<StatusCode.Code> codes = resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(request));
-    GrpcCallContext grpcCallContext = GrpcCallContext.createDefault().withRetryableCodes(codes);
+    GrpcCallContext grpcCallContext = Retrying.newCallContext().withRetryableCodes(codes);
     return new GrpcBlobReadChannel(
         storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext),
         request,
@@ -1221,10 +1234,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .apply(CreateHmacKeyRequest.newBuilder())
             .setServiceAccountEmail(serviceAccount.getEmail())
             .build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(request),
-        () -> storageClient.createHmacKeyCallable().call(request, grpcCallContext),
+        () -> storageClient.createHmacKeyCallable().call(request, merge),
         resp -> {
           ByteString secretKeyBytes = resp.getSecretKeyBytes();
           String b64SecretKey = BaseEncoding.base64().encode(secretKeyBytes.toByteArray());
@@ -1247,10 +1261,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .apply(ListHmacKeysRequest.newBuilder())
             .build();
     try {
+      GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
       return Retrying.run(
           getOptions(),
           retryAlgorithmManager.getFor(request),
-          () -> storageClient.listHmacKeysPagedCallable().call(request, grpcCallContext),
+          () -> storageClient.listHmacKeysPagedCallable().call(request, merge),
           resp ->
               new TransformingPageDecorator<>(
                   resp.getPage(),
@@ -1274,10 +1289,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .apply(GetHmacKeyRequest.newBuilder())
             .setAccessId(accessId)
             .build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(request),
-        () -> storageClient.getHmacKeyCallable().call(request, grpcCallContext),
+        () -> storageClient.getHmacKeyCallable().call(request, merge),
         codecs.hmacKeyMetadata());
   }
 
@@ -1291,11 +1307,12 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .setAccessId(hmacKeyMetadata.getAccessId())
             .setProject(projectNameCodec.encode(hmacKeyMetadata.getProjectId()))
             .build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
         () -> {
-          storageClient.deleteHmacKeyCallable().call(req, grpcCallContext);
+          storageClient.deleteHmacKeyCallable().call(req, merge);
           return null;
         },
         Decoder.identity());
@@ -1314,10 +1331,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         opts.updateHmacKeysRequest().apply(UpdateHmacKeyRequest.newBuilder()).setHmacKey(encode);
     UpdateHmacKeyRequest request =
         builder.setUpdateMask(FieldMask.newBuilder().addPaths("state").build()).build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(request),
-        () -> storageClient.updateHmacKeyCallable().call(request, grpcCallContext),
+        () -> storageClient.updateHmacKeyCallable().call(request, merge),
         codecs.hmacKeyMetadata());
   }
 
@@ -1329,10 +1347,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     GetIamPolicyRequest.Builder builder =
         GetIamPolicyRequest.newBuilder().setResource(bucketNameCodec.encode(bucket));
     GetIamPolicyRequest req = opts.getIamPolicyRequest().apply(builder).build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.getIamPolicyCallable().call(req, grpcCallContext),
+        () -> storageClient.getIamPolicyCallable().call(req, merge),
         codecs.policyCodec());
   }
 
@@ -1346,10 +1365,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .setResource(bucketNameCodec.encode(bucket))
             .setPolicy(codecs.policyCodec().encode(policy))
             .build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.setIamPolicyCallable().call(req, grpcCallContext),
+        () -> storageClient.setIamPolicyCallable().call(req, merge),
         codecs.policyCodec());
   }
 
@@ -1364,10 +1384,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .setResource(bucketNameCodec.encode(bucket))
             .addAllPermissions(permissions)
             .build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.testIamPermissionsCallable().call(req, grpcCallContext),
+        () -> storageClient.testIamPermissionsCallable().call(req, merge),
         resp -> {
           Set<String> heldPermissions = ImmutableSet.copyOf(resp.getPermissionsList());
           return permissions.stream()
@@ -1382,10 +1403,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         GetServiceAccountRequest.newBuilder()
             .setProject(projectNameCodec.encode(projectId))
             .build();
+    GrpcCallContext retryContext = Retrying.newCallContext();
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.getServiceAccountCallable().call(req),
+        () -> storageClient.getServiceAccountCallable().call(req, retryContext),
         codecs.serviceAccount());
   }
 
@@ -1397,10 +1419,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .setParent(bucketNameCodec.encode(bucket))
             .setNotificationConfig(encode)
             .build();
+    GrpcCallContext retryContext = Retrying.newCallContext();
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.createNotificationConfigCallable().call(req),
+        () -> storageClient.createNotificationConfigCallable().call(req, retryContext),
         syntaxDecoders.notificationConfig);
   }
 
@@ -1415,12 +1438,13 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     }
     GetNotificationConfigRequest req =
         GetNotificationConfigRequest.newBuilder().setName(name).build();
+    GrpcCallContext retryContext = Retrying.newCallContext();
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
         () -> {
           try {
-            return storageClient.getNotificationConfigCallable().call(req);
+            return storageClient.getNotificationConfigCallable().call(req, retryContext);
           } catch (NotFoundException e) {
             return null;
           }
@@ -1435,10 +1459,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .setParent(bucketNameCodec.encode(bucket))
             .build();
     ResultRetryAlgorithm<?> algorithm = retryAlgorithmManager.getFor(req);
+    GrpcCallContext retryContext = Retrying.newCallContext();
     return Retrying.run(
         getOptions(),
         algorithm,
-        () -> storageClient.listNotificationConfigsPagedCallable().call(req),
+        () -> storageClient.listNotificationConfigsPagedCallable().call(req, retryContext),
         resp -> {
           TransformingPageDecorator<
                   ListNotificationConfigsRequest,
@@ -1464,13 +1489,14 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     }
     DeleteNotificationConfigRequest req =
         DeleteNotificationConfigRequest.newBuilder().setName(name).build();
+    GrpcCallContext retryContext = Retrying.newCallContext();
     return Boolean.TRUE.equals(
         Retrying.run(
             getOptions(),
             retryAlgorithmManager.getFor(req),
             () -> {
               try {
-                storageClient.deleteNotificationConfigCallable().call(req);
+                storageClient.deleteNotificationConfigCallable().call(req, retryContext);
                 return true;
               } catch (NotFoundException e) {
                 return false;
@@ -1551,11 +1577,12 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
       ListObjectsRequest nextPageReq =
           req.toBuilder().setPageToken(resp.getNextPageToken()).build();
       try {
+        GrpcCallContext merge = Utils.merge(ctx, Retrying.newCallContext());
         ListObjectsResponse nextPageResp =
             Retrying.run(
                 GrpcStorageImpl.this.getOptions(),
                 retryAlgorithmManager.getFor(nextPageReq),
-                () -> storageClient.listObjectsCallable().call(nextPageReq, ctx),
+                () -> storageClient.listObjectsCallable().call(nextPageReq, merge),
                 Decoder.identity());
         return new ListObjectsWithSyntheticDirectoriesPage(ctx, nextPageReq, nextPageResp);
       } catch (Exception e) {
@@ -1649,6 +1676,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
                   page,
                   p -> p != null && p.hasNextPage(),
                   prev -> {
+                    // TODO: retry token header
                     // explicitly define this callable rather than using the method reference to
                     // prevent a javac 1.8 exception
                     // https://bugs.java.com/bugdatabase/view_bug.do?bug_id=8056984
@@ -1773,7 +1801,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts);
     Set<StatusCode.Code> codes =
         resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(readObjectRequest));
-    GrpcCallContext grpcCallContext = GrpcCallContext.createDefault().withRetryableCodes(codes);
+    GrpcCallContext grpcCallContext = Retrying.newCallContext().withRetryableCodes(codes);
     return ResumableMedia.gapic()
         .read()
         .byteChannel(storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext))
@@ -1787,12 +1815,13 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   ApiFuture<ResumableWrite> startResumableWrite(
       GrpcCallContext grpcCallContext, WriteObjectRequest req) {
     Set<StatusCode.Code> codes = resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(req));
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return ResumableMedia.gapic()
         .write()
         .resumableWrite(
             storageClient
                 .startResumableWriteCallable()
-                .withDefaultCallContext(grpcCallContext.withRetryableCodes(codes)),
+                .withDefaultCallContext(merge.withRetryableCodes(codes)),
             req);
   }
 
@@ -1818,10 +1847,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .setName(bucketNameCodec.encode(bucketName))
             .build();
 
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.getBucketCallable().call(req, grpcCallContext),
+        () -> storageClient.getBucketCallable().call(req, merge),
         Decoder.identity());
   }
 
@@ -1837,19 +1867,21 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .setName(bucketNameCodec.encode(bucketName))
             .build();
 
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.getBucketCallable().call(req, grpcCallContext),
+        () -> storageClient.getBucketCallable().call(req, merge),
         Decoder.identity());
   }
 
   private com.google.storage.v2.Bucket updateBucket(UpdateBucketRequest req) {
     GrpcCallContext grpcCallContext = GrpcCallContext.createDefault();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.updateBucketCallable().call(req, grpcCallContext),
+        () -> storageClient.updateBucketCallable().call(req, merge),
         Decoder.identity());
   }
 
@@ -1899,10 +1931,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .setObject(obj.getName())
             .build();
 
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.getObjectCallable().call(req, grpcCallContext),
+        () -> storageClient.getObjectCallable().call(req, merge),
         Decoder.identity());
   }
 
@@ -1926,10 +1959,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   private Object updateObject(UpdateObjectRequest req) {
     GrpcCallContext grpcCallContext = GrpcCallContext.createDefault();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.updateObjectCallable().call(req, grpcCallContext),
+        () -> storageClient.updateObjectCallable().call(req, merge),
         Decoder.identity());
   }
 
@@ -1957,12 +1991,13 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .setObject(blob.getName());
     ifNonNull(blob.getGeneration(), builder::setGeneration);
     GetObjectRequest req = opts.getObjectsRequest().apply(builder).build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
         () -> {
           try {
-            return storageClient.getObjectCallable().call(req, grpcCallContext);
+            return storageClient.getObjectCallable().call(req, merge);
           } catch (NotFoundException ignore) {
             return null;
           }
@@ -1978,10 +2013,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     GetBucketRequest.Builder builder =
         GetBucketRequest.newBuilder().setName(bucketNameCodec.encode(bucket));
     GetBucketRequest req = opts.getBucketsRequest().apply(builder).build();
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> storageClient.getBucketCallable().call(req, grpcCallContext),
+        () -> storageClient.getBucketCallable().call(req, merge),
         syntaxDecoders.bucket.andThen(opts.clearBucketFields()));
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Option.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Option.java
index 5fb7479c..e526486e 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Option.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Option.java
@@ -18,6 +18,11 @@ package com.google.cloud.storage;
 
 import com.google.cloud.storage.UnifiedOpts.Opt;
 import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /** Base class for Storage operation option. */
 @Deprecated
@@ -29,4 +34,32 @@ public abstract class Option<O extends Opt> extends UnifiedOpts.OptionShim<O>
   Option(O opt) {
     super(opt);
   }
+
+  @SafeVarargs
+  static <O extends Option<?>> O[] dedupe(IntFunction<O[]> gen, O... os) {
+    return dedupe(gen, Arrays.stream(os));
+  }
+
+  @SafeVarargs
+  static <O extends Option<?>> O[] dedupe(IntFunction<O[]> gen, Collection<O> collection, O... os) {
+    return dedupe(gen, Stream.of(collection.stream(), Arrays.stream(os)).flatMap(s -> s));
+  }
+
+  @SafeVarargs
+  static <O extends Option<?>> O[] dedupe(IntFunction<O[]> gen, O[] array, O... os) {
+    return dedupe(gen, Stream.of(Arrays.stream(array), Arrays.stream(os)).flatMap(s -> s));
+  }
+
+  /**
+   * All Options contain an {@link Opt}, {@code Opt}s are distinct classes allowing us to group
+   * based on those classes. Once grouped, we select the last element to provide last wins behavior.
+   *
+   * <p>Each of these helpers is an internal implementation detail, primarily due to the fact that
+   * generic arrays can not be instantiated in Java and requires a factory to be passed in.
+   */
+  private static <O extends Option<?>> O[] dedupe(IntFunction<O[]> gen, Stream<O> s) {
+    return s.collect(Collectors.groupingBy(o -> o.getOpt().getClass())).values().stream()
+        .map(l -> l.get(l.size() - 1))
+        .toArray(gen);
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java
index cfc961be..f625b58b 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java
@@ -20,14 +20,19 @@ import static com.google.cloud.RetryHelper.runWithRetries;
 
 import com.google.api.core.ApiClock;
 import com.google.api.core.NanoClock;
+import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.cloud.RetryHelper.RetryHelperException;
 import com.google.cloud.storage.Conversions.Decoder;
 import com.google.cloud.storage.spi.v1.HttpRpcContext;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import java.util.UUID;
 import java.util.concurrent.Callable;
 import java.util.function.Function;
+import org.checkerframework.checker.nullness.qual.NonNull;
 
 final class Retrying {
 
@@ -115,6 +120,14 @@ final class Retrying {
     }
   }
 
+  @NonNull
+  static GrpcCallContext newCallContext() {
+    return GrpcCallContext.createDefault()
+        .withExtraHeaders(
+            ImmutableMap.of(
+                "x-goog-gcs-idempotency-token", ImmutableList.of(UUID.randomUUID().toString())));
+  }
+
   static ResultRetryAlgorithm<?> neverRetry() {
     return new BasicResultRetryAlgorithm<Object>() {
       @Override
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 d1838985..63a0793c 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
@@ -45,7 +45,6 @@ import com.google.cloud.storage.UnifiedOpts.NamedField;
 import com.google.cloud.storage.UnifiedOpts.ObjectListOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
-import com.google.cloud.storage.UnifiedOpts.OptionShim;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
@@ -59,6 +58,7 @@ import java.net.URLConnection;
 import java.nio.file.Path;
 import java.security.Key;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedHashSet;
 import java.util.LinkedList;
@@ -358,6 +358,39 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
     public static BucketTargetOption projection(@NonNull String projection) {
       return new BucketTargetOption(UnifiedOpts.projection(projection));
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static BucketTargetOption[] dedupe(BucketTargetOption... os) {
+      return Option.dedupe(BucketTargetOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    public static BucketTargetOption[] dedupe(
+        Collection<BucketTargetOption> collection, BucketTargetOption... os) {
+      return Option.dedupe(BucketTargetOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    public static BucketTargetOption[] dedupe(
+        BucketTargetOption[] array, BucketTargetOption... os) {
+      return Option.dedupe(BucketTargetOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying bucket source options. */
@@ -400,6 +433,41 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
     public static BucketSourceOption requestedPolicyVersion(long version) {
       return new BucketSourceOption(UnifiedOpts.requestedPolicyVersion(version));
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static BucketSourceOption[] dedupe(BucketSourceOption... os) {
+      return Option.dedupe(BucketSourceOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BucketSourceOption[] dedupe(
+        Collection<BucketSourceOption> collection, BucketSourceOption... os) {
+      return Option.dedupe(BucketSourceOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BucketSourceOption[] dedupe(
+        BucketSourceOption[] array, BucketSourceOption... os) {
+      return Option.dedupe(BucketSourceOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying listHmacKeys options */
@@ -456,6 +524,41 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
     public static ListHmacKeysOption projectId(@NonNull String projectId) {
       return new ListHmacKeysOption(UnifiedOpts.projectId(projectId));
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static ListHmacKeysOption[] dedupe(ListHmacKeysOption... os) {
+      return Option.dedupe(ListHmacKeysOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static ListHmacKeysOption[] dedupe(
+        Collection<ListHmacKeysOption> collection, ListHmacKeysOption... os) {
+      return Option.dedupe(ListHmacKeysOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static ListHmacKeysOption[] dedupe(
+        ListHmacKeysOption[] array, ListHmacKeysOption... os) {
+      return Option.dedupe(ListHmacKeysOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying createHmacKey options */
@@ -482,6 +585,41 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
     public static CreateHmacKeyOption projectId(@NonNull String projectId) {
       return new CreateHmacKeyOption(UnifiedOpts.projectId(projectId));
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static CreateHmacKeyOption[] dedupe(CreateHmacKeyOption... os) {
+      return Option.dedupe(CreateHmacKeyOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static CreateHmacKeyOption[] dedupe(
+        Collection<CreateHmacKeyOption> collection, CreateHmacKeyOption... os) {
+      return Option.dedupe(CreateHmacKeyOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static CreateHmacKeyOption[] dedupe(
+        CreateHmacKeyOption[] array, CreateHmacKeyOption... os) {
+      return Option.dedupe(CreateHmacKeyOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying getHmacKey options */
@@ -507,6 +645,40 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
     public static GetHmacKeyOption projectId(@NonNull String projectId) {
       return new GetHmacKeyOption(UnifiedOpts.projectId(projectId));
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static GetHmacKeyOption[] dedupe(GetHmacKeyOption... os) {
+      return Option.dedupe(GetHmacKeyOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static GetHmacKeyOption[] dedupe(
+        Collection<GetHmacKeyOption> collection, GetHmacKeyOption... os) {
+      return Option.dedupe(GetHmacKeyOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static GetHmacKeyOption[] dedupe(GetHmacKeyOption[] array, GetHmacKeyOption... os) {
+      return Option.dedupe(GetHmacKeyOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying deleteHmacKey options */
@@ -523,6 +695,41 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
     public static DeleteHmacKeyOption userProject(@NonNull String userProject) {
       return new DeleteHmacKeyOption(UnifiedOpts.userProject(userProject));
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static DeleteHmacKeyOption[] dedupe(DeleteHmacKeyOption... os) {
+      return Option.dedupe(DeleteHmacKeyOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static DeleteHmacKeyOption[] dedupe(
+        Collection<DeleteHmacKeyOption> collection, DeleteHmacKeyOption... os) {
+      return Option.dedupe(DeleteHmacKeyOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static DeleteHmacKeyOption[] dedupe(
+        DeleteHmacKeyOption[] array, DeleteHmacKeyOption... os) {
+      return Option.dedupe(DeleteHmacKeyOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying updateHmacKey options */
@@ -539,6 +746,41 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
     public static UpdateHmacKeyOption userProject(@NonNull String userProject) {
       return new UpdateHmacKeyOption(UnifiedOpts.userProject(userProject));
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static UpdateHmacKeyOption[] dedupe(UpdateHmacKeyOption... os) {
+      return Option.dedupe(UpdateHmacKeyOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static UpdateHmacKeyOption[] dedupe(
+        Collection<UpdateHmacKeyOption> collection, UpdateHmacKeyOption... os) {
+      return Option.dedupe(UpdateHmacKeyOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static UpdateHmacKeyOption[] dedupe(
+        UpdateHmacKeyOption[] array, UpdateHmacKeyOption... os) {
+      return Option.dedupe(UpdateHmacKeyOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying bucket get options. */
@@ -593,6 +835,40 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
               .build();
       return new BucketGetOption(UnifiedOpts.fields(set));
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static BucketGetOption[] dedupe(BucketGetOption... os) {
+      return Option.dedupe(BucketGetOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BucketGetOption[] dedupe(
+        Collection<BucketGetOption> collection, BucketGetOption... os) {
+      return Option.dedupe(BucketGetOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BucketGetOption[] dedupe(BucketGetOption[] array, BucketGetOption... os) {
+      return Option.dedupe(BucketGetOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying blob target options. */
@@ -743,10 +1019,44 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
     public static BlobTargetOption kmsKeyName(@NonNull String kmsKeyName) {
       return new BlobTargetOption(UnifiedOpts.kmsKeyName(kmsKeyName));
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static BlobTargetOption[] dedupe(BlobTargetOption... os) {
+      return Option.dedupe(BlobTargetOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobTargetOption[] dedupe(
+        Collection<BlobTargetOption> collection, BlobTargetOption... os) {
+      return Option.dedupe(BlobTargetOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobTargetOption[] dedupe(BlobTargetOption[] array, BlobTargetOption... os) {
+      return Option.dedupe(BlobTargetOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying blob write options. */
-  class BlobWriteOption extends OptionShim<ObjectTargetOpt> implements Serializable {
+  class BlobWriteOption extends Option<ObjectTargetOpt> implements Serializable {
 
     private static final long serialVersionUID = 5536338021856320475L;
 
@@ -921,6 +1231,40 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
     public static BlobWriteOption detectContentType() {
       return new BlobWriteOption(UnifiedOpts.detectContentType());
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static BlobWriteOption[] dedupe(BlobWriteOption... os) {
+      return Option.dedupe(BlobWriteOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobWriteOption[] dedupe(
+        Collection<BlobWriteOption> collection, BlobWriteOption... os) {
+      return Option.dedupe(BlobWriteOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobWriteOption[] dedupe(BlobWriteOption[] array, BlobWriteOption... os) {
+      return Option.dedupe(BlobWriteOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying blob source options. */
@@ -1034,6 +1378,40 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
     public static BlobSourceOption shouldReturnRawInputStream(boolean shouldReturnRawInputStream) {
       return new BlobSourceOption(UnifiedOpts.returnRawInputStream(shouldReturnRawInputStream));
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static BlobSourceOption[] dedupe(BlobSourceOption... os) {
+      return Option.dedupe(BlobSourceOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobSourceOption[] dedupe(
+        Collection<BlobSourceOption> collection, BlobSourceOption... os) {
+      return Option.dedupe(BlobSourceOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobSourceOption[] dedupe(BlobSourceOption[] array, BlobSourceOption... os) {
+      return Option.dedupe(BlobSourceOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying blob get options. */
@@ -1161,6 +1539,40 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
     public static BlobGetOption shouldReturnRawInputStream(boolean shouldReturnRawInputStream) {
       return new BlobGetOption(UnifiedOpts.returnRawInputStream(shouldReturnRawInputStream));
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static BlobGetOption[] dedupe(BlobGetOption... os) {
+      return Option.dedupe(BlobGetOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobGetOption[] dedupe(
+        Collection<BlobGetOption> collection, BlobGetOption... os) {
+      return Option.dedupe(BlobGetOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobGetOption[] dedupe(BlobGetOption[] array, BlobGetOption... os) {
+      return Option.dedupe(BlobGetOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying bucket list options. */
@@ -1219,6 +1631,40 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
               .collect(ImmutableSet.toImmutableSet());
       return new BucketListOption(UnifiedOpts.fields(set));
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static BucketListOption[] dedupe(BucketListOption... os) {
+      return Option.dedupe(BucketListOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BucketListOption[] dedupe(
+        Collection<BucketListOption> collection, BucketListOption... os) {
+      return Option.dedupe(BucketListOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BucketListOption[] dedupe(BucketListOption[] array, BucketListOption... os) {
+      return Option.dedupe(BucketListOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying blob list options. */
@@ -1351,6 +1797,40 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
               .collect(ImmutableSet.toImmutableSet());
       return new BlobListOption(UnifiedOpts.fields(set));
     }
+
+    /**
+     * Deduplicate any options which are the same parameter. The value which comes last in {@code
+     * os} will be the value included in the return.
+     */
+    @BetaApi
+    public static BlobListOption[] dedupe(BlobListOption... os) {
+      return Option.dedupe(BlobListOption[]::new, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobListOption[] dedupe(
+        Collection<BlobListOption> collection, BlobListOption... os) {
+      return Option.dedupe(BlobListOption[]::new, collection, os);
+    }
+
+    /**
+     * Deduplicate any options which are the same parameter.
+     *
+     * <p>The value which comes last in {@code collection} and {@code os} will be the value included
+     * in the return. All options from {@code os} will override their counterparts in {@code
+     * collection}.
+     */
+    @BetaApi
+    public static BlobListOption[] dedupe(BlobListOption[] array, BlobListOption... os) {
+      return Option.dedupe(BlobListOption[]::new, array, os);
+    }
   }
 
   /** Class for specifying Post Policy V4 options. * */
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 81d7fbb9..67bafab8 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
@@ -20,6 +20,8 @@ import static java.util.Objects.requireNonNull;
 
 import com.google.api.client.util.DateTime;
 import com.google.api.core.InternalApi;
+import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.gax.rpc.ApiCallContext;
 import com.google.cloud.storage.Conversions.Codec;
 import com.google.cloud.storage.UnifiedOpts.NamedField;
 import com.google.common.annotations.VisibleForTesting;
@@ -289,4 +291,14 @@ final class Utils {
   private static String crc32cEncode(int from) {
     return BaseEncoding.base64().encode(Ints.toByteArray(from));
   }
+
+  /**
+   * Type preserving method for {@link GrpcCallContext#merge(ApiCallContext)}
+   *
+   * @see GrpcCallContext#merge(ApiCallContext)
+   */
+  @NonNull
+  static GrpcCallContext merge(@NonNull GrpcCallContext l, @NonNull GrpcCallContext r) {
+    return (GrpcCallContext) l.merge(r);
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
index 7deb0043..d5a205ed 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
@@ -31,6 +31,7 @@ import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Consumer;
 import java.util.function.LongConsumer;
+import java.util.function.Supplier;
 import org.checkerframework.checker.nullness.qual.NonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
@@ -57,12 +58,19 @@ final class WriteFlushStrategy {
   static FlusherFactory fsyncEveryFlush(
       ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
       RetryingDependencies deps,
-      ResultRetryAlgorithm<?> alg) {
+      ResultRetryAlgorithm<?> alg,
+      Supplier<GrpcCallContext> baseContextSupplier) {
     return (String bucketName,
         LongConsumer committedTotalBytesCallback,
         Consumer<WriteObjectResponse> onSuccessCallback) ->
         new FsyncEveryFlusher(
-            write, deps, alg, bucketName, committedTotalBytesCallback, onSuccessCallback);
+            write,
+            deps,
+            alg,
+            bucketName,
+            committedTotalBytesCallback,
+            onSuccessCallback,
+            baseContextSupplier);
   }
 
   /**
@@ -79,14 +87,14 @@ final class WriteFlushStrategy {
         new FsyncOnClose(write, bucketName, committedTotalBytesCallback, onSuccessCallback);
   }
 
-  private static GrpcCallContext contextWithBucketName(String bucketName) {
-    GrpcCallContext ret = GrpcCallContext.createDefault();
+  private static GrpcCallContext contextWithBucketName(
+      String bucketName, GrpcCallContext baseContext) {
     if (bucketName != null && !bucketName.isEmpty()) {
-      return ret.withExtraHeaders(
+      return baseContext.withExtraHeaders(
           ImmutableMap.of(
               "x-goog-request-params", ImmutableList.of(String.format("bucket=%s", bucketName))));
     }
-    return ret;
+    return baseContext;
   }
 
   /**
@@ -138,6 +146,7 @@ final class WriteFlushStrategy {
     private final String bucketName;
     private final LongConsumer sizeCallback;
     private final Consumer<WriteObjectResponse> completeCallback;
+    private final Supplier<GrpcCallContext> baseContextSupplier;
 
     private FsyncEveryFlusher(
         ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
@@ -145,13 +154,15 @@ final class WriteFlushStrategy {
         ResultRetryAlgorithm<?> alg,
         String bucketName,
         LongConsumer sizeCallback,
-        Consumer<WriteObjectResponse> completeCallback) {
+        Consumer<WriteObjectResponse> completeCallback,
+        Supplier<GrpcCallContext> baseContextSupplier) {
       this.write = write;
       this.deps = deps;
       this.alg = alg;
       this.bucketName = bucketName;
       this.sizeCallback = sizeCallback;
       this.completeCallback = completeCallback;
+      this.baseContextSupplier = baseContextSupplier;
     }
 
     public void flush(@NonNull List<WriteObjectRequest> segments) {
@@ -160,7 +171,8 @@ final class WriteFlushStrategy {
           alg,
           () -> {
             Observer observer = new Observer(sizeCallback, completeCallback);
-            GrpcCallContext internalContext = contextWithBucketName(bucketName);
+            GrpcCallContext internalContext =
+                contextWithBucketName(bucketName, baseContextSupplier.get());
             ApiStreamObserver<WriteObjectRequest> write =
                 this.write.withDefaultCallContext(internalContext).clientStreamingCall(observer);
 
@@ -230,7 +242,8 @@ final class WriteFlushStrategy {
       if (stream == null) {
         synchronized (this) {
           if (stream == null) {
-            GrpcCallContext internalContext = contextWithBucketName(bucketName);
+            GrpcCallContext internalContext =
+                contextWithBucketName(bucketName, GrpcCallContext.createDefault());
             stream =
                 this.write
                     .withDefaultCallContext(internalContext)
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 707ba457..18e95e19 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
@@ -182,6 +182,7 @@ public class HttpStorageRpc implements StorageRpc {
           newValue = invocationEntry;
         }
         headers.set("x-goog-api-client", newValue);
+        headers.set("x-goog-gcs-idempotency-token", invocationId);
       }
     }
   }
diff --git a/pom.xml b/pom.xml
index b7444791..f0003ff2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -59,12 +59,25 @@
 
   <dependencyManagement>
     <dependencies>
+      <dependency>
+        <groupId>io.grpc</groupId>
+        <artifactId>grpc-bom</artifactId>
+        <version>1.56.0</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-shared-dependencies</artifactId>
         <version>${google.cloud.shared-dependencies.version}</version>
         <type>pom</type>
         <scope>import</scope>
+        <exclusions>
+          <exclusion>
+            <groupId>io.grpc</groupId>
+            <artifactId>*</artifactId>
+          </exclusion>
+        </exclusions>
       </dependency>
 
       <dependency>
@@ -75,7 +88,7 @@
       <dependency>
         <groupId>com.google.apis</groupId>
         <artifactId>google-api-services-storage</artifactId>
-        <version>v1-rev20230301-2.0.0</version>
+        <version>v1-rev20230617-2.0.0</version>
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
@@ -164,7 +177,7 @@
       <dependency>
         <groupId>com.google.truth</groupId>
         <artifactId>truth</artifactId>
-        <version>1.1.4</version>
+        <version>1.1.5</version>
         <scope>test</scope>
         <exclusions>
           <exclusion>
diff --git a/samples/install-without-bom/pom.xml b/samples/install-without-bom/pom.xml
index 6ad8b2bb..f4c37b20 100644
--- a/samples/install-without-bom/pom.xml
+++ b/samples/install-without-bom/pom.xml
@@ -43,7 +43,7 @@
     <dependency>
       <groupId>com.google.truth</groupId>
       <artifactId>truth</artifactId>
-      <version>1.1.4</version>
+      <version>1.1.5</version>
       <scope>test</scope>
     </dependency>
       <dependency>
diff --git a/samples/native-image-sample/pom.xml b/samples/native-image-sample/pom.xml
index af1943d5..354c81d3 100644
--- a/samples/native-image-sample/pom.xml
+++ b/samples/native-image-sample/pom.xml
@@ -55,7 +55,7 @@
     <dependency>
       <groupId>com.google.truth</groupId>
       <artifactId>truth</artifactId>
-      <version>1.1.4</version>
+      <version>1.1.5</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -134,7 +134,7 @@
           <plugin>
             <groupId>org.graalvm.buildtools</groupId>
             <artifactId>native-maven-plugin</artifactId>
-            <version>0.9.22</version>
+            <version>0.9.23</version>
             <extensions>true</extensions>
             <configuration>
               <mainClass>com.example.storage.NativeImageStorageSample</mainClass>
diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml
index df4f7850..423332fc 100644
--- a/samples/snapshot/pom.xml
+++ b/samples/snapshot/pom.xml
@@ -40,7 +40,7 @@
     <dependency>
       <groupId>com.google.truth</groupId>
       <artifactId>truth</artifactId>
-      <version>1.1.4</version>
+      <version>1.1.5</version>
       <scope>test</scope>
     </dependency>
     <dependency>
diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml
index 95f581a7..df642f82 100644
--- a/samples/snippets/pom.xml
+++ b/samples/snippets/pom.xml
@@ -54,7 +54,7 @@
     <dependency>
       <groupId>com.google.truth</groupId>
       <artifactId>truth</artifactId>
-      <version>1.1.4</version>
+      <version>1.1.5</version>
       <scope>test</scope>
     </dependency>
     <dependency>


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java
index 6ee17f11..08d228c2 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java
@@ -191,7 +191,8 @@ public final class ITGapicUnbufferedWritableByteChannelTest {
               WriteFlushStrategy.fsyncEveryFlush(
                   sc.writeObjectCallable(),
                   RetryingDependencies.attemptOnce(),
-                  Retrying.neverRetry()));
+                  Retrying.neverRetry(),
+                  Retrying::newCallContext));
       ArrayList<String> debugMessages = new ArrayList<>();
       try {
         ImmutableList<ByteBuffer> buffers = TestUtils.subDivide(bytes, 10);
@@ -279,7 +280,8 @@ public final class ITGapicUnbufferedWritableByteChannelTest {
                     public boolean shouldRetry(Throwable t, Object ignore) {
                       return TestUtils.findThrowable(DataLossException.class, t) != null;
                     }
-                  }))) {
+                  },
+                  Retrying::newCallContext))) {
         writeCtx = c.getWriteCtx();
         ImmutableList<ByteBuffer> buffers = TestUtils.subDivide(bytes, 10);
         c.write(buffers.get(0));
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java
index 4c96d430..f4c037b1 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java
@@ -18,6 +18,7 @@ package com.google.cloud.storage;
 
 import static com.google.common.truth.Truth.assertThat;
 
+import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.rpc.ApiCallContext;
 import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.api.gax.rpc.ClientStreamingCallable;
@@ -47,7 +48,10 @@ public final class WriteFlushStrategyTest {
     doTest(
         write ->
             WriteFlushStrategy.fsyncEveryFlush(
-                write, RetryingDependencies.attemptOnce(), Retrying.neverRetry()),
+                write,
+                RetryingDependencies.attemptOnce(),
+                Retrying.neverRetry(),
+                GrpcCallContext::createDefault),
         "bucket-name",
         expectedHeaderNonNullNonEmpty);
   }
@@ -62,7 +66,10 @@ public final class WriteFlushStrategyTest {
     doTest(
         write ->
             WriteFlushStrategy.fsyncEveryFlush(
-                write, RetryingDependencies.attemptOnce(), Retrying.neverRetry()),
+                write,
+                RetryingDependencies.attemptOnce(),
+                Retrying.neverRetry(),
+                GrpcCallContext::createDefault),
         "",
         expectedHeaderNonNullEmpty);
   }
@@ -77,7 +84,10 @@ public final class WriteFlushStrategyTest {
     doTest(
         write ->
             WriteFlushStrategy.fsyncEveryFlush(
-                write, RetryingDependencies.attemptOnce(), Retrying.neverRetry()),
+                write,
+                RetryingDependencies.attemptOnce(),
+                Retrying.neverRetry(),
+                GrpcCallContext::createDefault),
         null,
         expectedHeaderNull);
   }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/DedupeOptionTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/DedupeOptionTest.java
new file mode 100644
index 00000000..b7b38df0
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/DedupeOptionTest.java
@@ -0,0 +1,163 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.it;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BucketTargetOption;
+import com.google.common.collect.ImmutableList;
+import org.junit.Test;
+
+public final class DedupeOptionTest {
+
+  @Test
+  public void dedupe_varargs() {
+    BucketTargetOption[] dedupe =
+        BucketTargetOption.dedupe(
+            BucketTargetOption.userProject("abc"),
+            BucketTargetOption.metagenerationMatch(),
+            BucketTargetOption.userProject("xyz"));
+
+    assertThat(dedupe)
+        .asList()
+        .containsExactly(
+            BucketTargetOption.metagenerationMatch(), BucketTargetOption.userProject("xyz"));
+  }
+
+  @Test
+  public void dedupe_collection_varargs() {
+    BucketTargetOption[] dedupe =
+        BucketTargetOption.dedupe(
+            ImmutableList.of(
+                BucketTargetOption.userProject("abc"), BucketTargetOption.metagenerationMatch()),
+            BucketTargetOption.userProject("xyz"));
+
+    assertThat(dedupe)
+        .asList()
+        .containsExactly(
+            BucketTargetOption.metagenerationMatch(), BucketTargetOption.userProject("xyz"));
+  }
+
+  @Test
+  public void dedupe_array_varargs() {
+    BucketTargetOption[] dedupe =
+        BucketTargetOption.dedupe(
+            new BucketTargetOption[] {
+              BucketTargetOption.userProject("abc"), BucketTargetOption.metagenerationMatch()
+            },
+            BucketTargetOption.userProject("xyz"));
+
+    assertThat(dedupe)
+        .asList()
+        .containsExactly(
+            BucketTargetOption.metagenerationMatch(), BucketTargetOption.userProject("xyz"));
+  }
+
+  @Test
+  public void allClasses_varargs() {
+    Storage.BlobGetOption.dedupe();
+    Storage.BlobListOption.dedupe();
+    Storage.BlobSourceOption.dedupe();
+    Storage.BlobTargetOption.dedupe();
+    Storage.BlobWriteOption.dedupe();
+    Storage.BucketGetOption.dedupe();
+    Storage.BucketListOption.dedupe();
+    Storage.BucketSourceOption.dedupe();
+    Storage.BucketTargetOption.dedupe();
+    Storage.CreateHmacKeyOption.dedupe();
+    Storage.DeleteHmacKeyOption.dedupe();
+    Storage.GetHmacKeyOption.dedupe();
+    Storage.ListHmacKeysOption.dedupe();
+    Storage.UpdateHmacKeyOption.dedupe();
+
+    Bucket.BlobTargetOption.dedupe();
+    Bucket.BlobWriteOption.dedupe();
+    Bucket.BucketSourceOption.dedupe();
+
+    Blob.BlobSourceOption.dedupe();
+  }
+
+  @Test
+  public void allClasses_collection_varargs() {
+    Storage.BlobGetOption.dedupe(ImmutableList.of());
+    Storage.BlobListOption.dedupe(ImmutableList.of());
+    Storage.BlobSourceOption.dedupe(ImmutableList.of());
+    Storage.BlobTargetOption.dedupe(ImmutableList.of());
+    Storage.BlobWriteOption.dedupe(ImmutableList.of());
+    Storage.BucketGetOption.dedupe(ImmutableList.of());
+    Storage.BucketListOption.dedupe(ImmutableList.of());
+    Storage.BucketSourceOption.dedupe(ImmutableList.of());
+    Storage.BucketTargetOption.dedupe(ImmutableList.of());
+    Storage.CreateHmacKeyOption.dedupe(ImmutableList.of());
+    Storage.DeleteHmacKeyOption.dedupe(ImmutableList.of());
+    Storage.GetHmacKeyOption.dedupe(ImmutableList.of());
+    Storage.ListHmacKeysOption.dedupe(ImmutableList.of());
+    Storage.UpdateHmacKeyOption.dedupe(ImmutableList.of());
+
+    Bucket.BlobTargetOption.dedupe(ImmutableList.of());
+    Bucket.BlobWriteOption.dedupe(ImmutableList.of());
+    Bucket.BucketSourceOption.dedupe(ImmutableList.of());
+
+    Blob.BlobSourceOption.dedupe(ImmutableList.of());
+  }
+
+  @Test
+  public void allClasses_array_varargs() {
+    String p = "proj";
+    Storage.BlobGetOption.dedupe(
+        new Storage.BlobGetOption[0], Storage.BlobGetOption.userProject(p));
+    Storage.BlobListOption.dedupe(
+        new Storage.BlobListOption[0], Storage.BlobListOption.userProject(p));
+    Storage.BlobSourceOption.dedupe(
+        new Storage.BlobSourceOption[0], Storage.BlobSourceOption.userProject(p));
+    Storage.BlobTargetOption.dedupe(
+        new Storage.BlobTargetOption[0], Storage.BlobTargetOption.userProject(p));
+    Storage.BlobWriteOption.dedupe(
+        new Storage.BlobWriteOption[0], Storage.BlobWriteOption.userProject(p));
+    Storage.BucketGetOption.dedupe(
+        new Storage.BucketGetOption[0], Storage.BucketGetOption.userProject(p));
+    Storage.BucketListOption.dedupe(
+        new Storage.BucketListOption[0], Storage.BucketListOption.userProject(p));
+    Storage.BucketSourceOption.dedupe(
+        new Storage.BucketSourceOption[0], Storage.BucketSourceOption.userProject(p));
+    Storage.BucketTargetOption.dedupe(
+        new Storage.BucketTargetOption[0], Storage.BucketTargetOption.userProject(p));
+    Storage.CreateHmacKeyOption.dedupe(
+        new Storage.CreateHmacKeyOption[0], Storage.CreateHmacKeyOption.userProject(p));
+    Storage.DeleteHmacKeyOption.dedupe(
+        new Storage.DeleteHmacKeyOption[0], Storage.DeleteHmacKeyOption.userProject(p));
+    Storage.GetHmacKeyOption.dedupe(
+        new Storage.GetHmacKeyOption[0], Storage.GetHmacKeyOption.userProject(p));
+    Storage.ListHmacKeysOption.dedupe(
+        new Storage.ListHmacKeysOption[0], Storage.ListHmacKeysOption.userProject(p));
+    Storage.UpdateHmacKeyOption.dedupe(
+        new Storage.UpdateHmacKeyOption[0], Storage.UpdateHmacKeyOption.userProject(p));
+
+    Bucket.BlobTargetOption.dedupe(
+        new Bucket.BlobTargetOption[0], Bucket.BlobTargetOption.userProject(p));
+    Bucket.BlobWriteOption.dedupe(
+        new Bucket.BlobWriteOption[0], Bucket.BlobWriteOption.userProject(p));
+    Bucket.BucketSourceOption.dedupe(
+        new Bucket.BucketSourceOption[0], Bucket.BucketSourceOption.userProject(p));
+
+    Blob.BlobSourceOption.dedupe(
+        new Blob.BlobSourceOption[0], Blob.BlobSourceOption.userProject(p));
+  }
+}
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
new file mode 100644
index 00000000..a674a079
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/GrpcRequestAuditing.java
@@ -0,0 +1,80 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.it;
+
+import static com.google.common.truth.Truth.assertWithMessage;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.truth.IterableSubject;
+import io.grpc.Attributes;
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
+import io.grpc.ClientStreamTracer;
+import io.grpc.ClientStreamTracer.StreamInfo;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+final class GrpcRequestAuditing implements ClientInterceptor {
+
+  private final List<Metadata> requestHeaders;
+
+  GrpcRequestAuditing() {
+    requestHeaders = Collections.synchronizedList(new ArrayList<>());
+  }
+
+  void clear() {
+    requestHeaders.clear();
+  }
+
+  @Override
+  public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+      MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+    CallOptions withStreamTracerFactory = callOptions.withStreamTracerFactory(new Factory());
+    return next.newCall(method, withStreamTracerFactory);
+  }
+
+  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 assertWithMessage(String.format("Headers %s", key.name())).that(actual);
+  }
+
+  private final class Factory extends ClientStreamTracer.Factory {
+    @Override
+    public ClientStreamTracer newClientStreamTracer(StreamInfo info, Metadata headers) {
+      return new Tracer();
+    }
+  }
+
+  private final class Tracer extends ClientStreamTracer {
+
+    @Override
+    public void streamCreated(Attributes transportAttrs, Metadata headers) {
+      requestHeaders.add(headers);
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcIdempotencyTokenTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcIdempotencyTokenTest.java
new file mode 100644
index 00000000..ec5d5425
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcIdempotencyTokenTest.java
@@ -0,0 +1,168 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.it;
+
+import static com.google.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.paging.Page;
+import com.google.cloud.WriteChannel;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.DataGenerator;
+import com.google.cloud.storage.Storage;
+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.StorageOptions;
+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.Inject;
+import com.google.cloud.storage.it.runner.annotations.SingleBackend;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.common.collect.ImmutableList;
+import com.google.common.truth.IterableSubject;
+import io.grpc.Metadata;
+import java.nio.ByteBuffer;
+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)
+@SingleBackend(Backend.PROD)
+public final class ITGrpcIdempotencyTokenTest {
+
+  private static final Metadata.Key<String> X_GOOG_GCS_IDEMPOTENCY_TOKEN =
+      Metadata.Key.of("x-goog-gcs-idempotency-token", Metadata.ASCII_STRING_MARSHALLER);
+  @Inject public BucketInfo bucket;
+  @Inject public Generator generator;
+
+  private Storage storage;
+  private GrpcRequestAuditing requestAuditing;
+
+  @Before
+  public void setUp() throws Exception {
+    requestAuditing = new GrpcRequestAuditing();
+    storage =
+        StorageOptions.grpc()
+            .setGrpcInterceptorProvider(() -> ImmutableList.of(requestAuditing))
+            .build()
+            .getService();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    if (storage != null) {
+      storage.close();
+    }
+  }
+
+  @Test
+  public void simpleUnary() throws Exception {
+    Bucket gen1 = storage.get(bucket.getName());
+
+    IterableSubject subject = requestAuditing.assertRequestHeader(X_GOOG_GCS_IDEMPOTENCY_TOKEN);
+    assertAll(() -> subject.hasSize(1));
+  }
+
+  @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());
+
+    requestAuditing.clear();
+    ImmutableList<String> expectedNamess = ImmutableList.of(blob1.getName(), blob2.getName());
+    Page<Blob> page =
+        storage.list(bucket.getName(), BlobListOption.prefix(baseName), BlobListOption.pageSize(1));
+
+    List<String> collect = page.streamAll().map(BlobInfo::getName).collect(Collectors.toList());
+    IterableSubject subject = requestAuditing.assertRequestHeader(X_GOOG_GCS_IDEMPOTENCY_TOKEN);
+    assertAll(
+        () -> assertThat(collect).hasSize(2),
+        () -> assertThat(collect).containsExactlyElementsIn(expectedNamess),
+        () -> subject.hasSize(2));
+  }
+
+  @Test
+  public void readObject() 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());
+
+    requestAuditing.clear();
+    byte[] actual = storage.readAllBytes(gen1.getBlobId(), BlobSourceOption.generationMatch());
+    IterableSubject subject = requestAuditing.assertRequestHeader(X_GOOG_GCS_IDEMPOTENCY_TOKEN);
+    String actualXxd = xxd(actual);
+
+    assertAll(() -> subject.hasSize(1), () -> 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();
+    requestAuditing.clear();
+    Blob gen1 = storage.create(info, expected, BlobTargetOption.doesNotExist());
+    IterableSubject subject = requestAuditing.assertRequestHeader(X_GOOG_GCS_IDEMPOTENCY_TOKEN);
+
+    byte[] actual = storage.readAllBytes(gen1.getBlobId(), BlobSourceOption.generationMatch());
+    String actualXxd = xxd(actual);
+
+    assertAll(() -> subject.hasSize(1), () -> 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())) {
+      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 = requestAuditing.assertRequestHeader(X_GOOG_GCS_IDEMPOTENCY_TOKEN);
+
+    byte[] actual = storage.readAllBytes(info.getBlobId());
+    String actualXxd = xxd(actual);
+
+    // We expect 4 distinct requests:
+    //  1. start resumable session
+    //  2. PUT first 256KiB
+    //  3. PUT second 256KiB
+    //  4. Finalize session and put final 45B
+    assertAll(() -> subject.hasSize(4), () -> assertThat(actualXxd).isEqualTo(expectedXxd));
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITHttpIdempotencyTokenTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITHttpIdempotencyTokenTest.java
new file mode 100644
index 00000000..7030c522
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITHttpIdempotencyTokenTest.java
@@ -0,0 +1,182 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.it;
+
+import static com.google.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.paging.Page;
+import com.google.cloud.WriteChannel;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.DataGenerator;
+import com.google.cloud.storage.Storage;
+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.BucketListOption;
+import com.google.cloud.storage.StorageOptions;
+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.Inject;
+import com.google.cloud.storage.it.runner.annotations.SingleBackend;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.common.collect.ImmutableList;
+import com.google.common.truth.IterableSubject;
+import java.nio.ByteBuffer;
+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)
+@SingleBackend(Backend.PROD)
+public final class ITHttpIdempotencyTokenTest {
+
+  private static final String X_GOOG_GCS_IDEMPOTENCY_TOKEN = "x-goog-gcs-idempotency-token";
+  @Inject public BucketInfo bucket;
+  @Inject public Generator generator;
+
+  private Storage storage;
+  private RequestAuditing requestAuditing;
+
+  @Before
+  public void setUp() throws Exception {
+    requestAuditing = new RequestAuditing();
+    storage = StorageOptions.http().setTransportOptions(requestAuditing).build().getService();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    if (storage != null) {
+      storage.close();
+    }
+  }
+
+  @Test
+  public void simpleUnary() throws Exception {
+    Bucket gen1 = storage.get(bucket.getName());
+
+    IterableSubject subject = requestAuditing.assertRequestHeader(X_GOOG_GCS_IDEMPOTENCY_TOKEN);
+    assertAll(() -> subject.hasSize(1));
+  }
+
+  @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());
+
+    requestAuditing.clear();
+    ImmutableList<String> expectedNamess = ImmutableList.of(blob1.getName(), blob2.getName());
+    Page<Blob> page =
+        storage.list(bucket.getName(), BlobListOption.prefix(baseName), BlobListOption.pageSize(1));
+
+    List<String> collect = page.streamAll().map(BlobInfo::getName).collect(Collectors.toList());
+    IterableSubject subject = requestAuditing.assertRequestHeader(X_GOOG_GCS_IDEMPOTENCY_TOKEN);
+    assertAll(
+        () -> assertThat(collect).hasSize(2),
+        () -> assertThat(collect).containsExactlyElementsIn(expectedNamess),
+        () -> subject.hasSize(2));
+  }
+
+  @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()) {
+      requestAuditing.clear();
+      Page<Bucket> page =
+          storage.list(BucketListOption.prefix(baseName), BucketListOption.pageSize(1));
+
+      List<Bucket> collect = page.streamAll().collect(Collectors.toList());
+      IterableSubject subject = requestAuditing.assertRequestHeader(X_GOOG_GCS_IDEMPOTENCY_TOKEN);
+      assertAll(() -> assertThat(collect).hasSize(2), () -> subject.hasSize(2));
+    }
+  }
+
+  @Test
+  public void readObject() 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());
+
+    requestAuditing.clear();
+    byte[] actual = storage.readAllBytes(gen1.getBlobId(), BlobSourceOption.generationMatch());
+    IterableSubject subject = requestAuditing.assertRequestHeader(X_GOOG_GCS_IDEMPOTENCY_TOKEN);
+    String actualXxd = xxd(actual);
+
+    assertAll(() -> subject.hasSize(1), () -> 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();
+    requestAuditing.clear();
+    Blob gen1 = storage.create(info, expected, BlobTargetOption.doesNotExist());
+    IterableSubject subject = requestAuditing.assertRequestHeader(X_GOOG_GCS_IDEMPOTENCY_TOKEN);
+
+    byte[] actual = storage.readAllBytes(gen1.getBlobId(), BlobSourceOption.generationMatch());
+    String actualXxd = xxd(actual);
+
+    assertAll(() -> subject.hasSize(1), () -> 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())) {
+      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 = requestAuditing.assertRequestHeader(X_GOOG_GCS_IDEMPOTENCY_TOKEN);
+
+    byte[] actual = storage.readAllBytes(info.getBlobId());
+    String actualXxd = xxd(actual);
+
+    // We expect 4 distinct requests:
+    //  1. start resumable session
+    //  2. PUT first 256KiB
+    //  3. PUT second 256KiB
+    //  4. Finalize session and put final 45B
+    assertAll(() -> subject.hasSize(4), () -> assertThat(actualXxd).isEqualTo(expectedXxd));
+  }
+}
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 1087ad8b..bbb1821c 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
@@ -31,6 +31,7 @@ import com.google.cloud.ServiceOptions;
 import com.google.cloud.http.HttpTransportOptions;
 import com.google.cloud.storage.it.CSEKSupport.EncryptionKeyTuple;
 import com.google.common.collect.ImmutableList;
+import com.google.common.truth.IterableSubject;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -194,4 +195,18 @@ final class RequestAuditing extends HttpTransportOptions {
         .that(collect)
         .isEqualTo(ImmutableList.of(expectedValue));
   }
+
+  IterableSubject assertRequestHeader(String headerName) {
+    ImmutableList<HttpRequest> requests = getRequests();
+
+    List<Object> actual =
+        requests.stream()
+            .map(HttpRequest::getHeaders)
+            .map(headers -> headers.get(headerName))
+            .filter(Objects::nonNull)
+            .distinct()
+            .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/SneakyException.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/SneakyException.java
index 52790c3d..aa6d0059 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/SneakyException.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/SneakyException.java
@@ -23,7 +23,7 @@ package com.google.cloud.storage.it.runner;
  * <p>This class provides some utility methods to sneakily (not method declared) throw exceptions
  * and later unwrap any sneakily wrapped exception if it's needed.
  */
-final class SneakyException extends RuntimeException {
+public final class SneakyException extends RuntimeException {
 
   public SneakyException(Throwable cause) {
     super(cause);
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/Generator.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/Generator.java
index 383053fa..e056a341 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/Generator.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/Generator.java
@@ -45,7 +45,9 @@ public final class Generator implements ManagedLifecycle {
       throw new IllegalStateException("No actively running test in registry.");
     }
     AtomicInteger counter = counters.computeIfAbsent(currentTest, (d) -> new AtomicInteger(1));
-    return String.format("%s-%04d", currentTest.getMethodName(), counter.getAndIncrement());
+    return String.format(
+        "%s.%s-%04d",
+        currentTest.getClassName(), currentTest.getMethodName(), counter.getAndIncrement());
   }
 
   @Override
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/TestBench.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/TestBench.java
index 41c4f497..03a53f4e 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/TestBench.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/TestBench.java
@@ -30,9 +30,13 @@ import com.google.api.core.NanoClock;
 import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.cloud.RetryHelper.RetryHelperException;
+import com.google.cloud.Tuple;
 import com.google.cloud.conformance.storage.v1.InstructionList;
 import com.google.cloud.conformance.storage.v1.Method;
+import com.google.cloud.storage.it.runner.SneakyException;
+import com.google.common.base.Charsets;
 import com.google.common.collect.ImmutableList;
+import com.google.common.io.CharStreams;
 import com.google.gson.Gson;
 import com.google.gson.JsonArray;
 import com.google.gson.JsonElement;
@@ -41,15 +45,20 @@ import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileReader;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
 import java.net.SocketException;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import org.threeten.bp.Duration;
 
 /**
@@ -411,9 +420,47 @@ public final class TestBench implements ManagedLifecycle {
   static final class Builder {
     private static final String DEFAULT_BASE_URI = "http://localhost:9000";
     private static final String DEFAULT_GRPC_BASE_URI = "http://localhost:9005";
-    private static final String DEFAULT_IMAGE_NAME =
-        "gcr.io/cloud-devrel-public-resources/storage-testbench";
-    private static final String DEFAULT_IMAGE_TAG = "v0.35.0";
+    private static final String DEFAULT_IMAGE_NAME;
+    private static final String DEFAULT_IMAGE_TAG;
+
+    static {
+      ClassLoader cl = Thread.currentThread().getContextClassLoader();
+      Tuple<String, String> nameAndTag =
+          SneakyException.unwrap(
+                  () -> {
+                    InputStream dockerfileText =
+                        cl.getResourceAsStream(
+                            "com/google/cloud/storage/it/runner/registry/Dockerfile");
+                    //noinspection UnstableApiUsage
+                    return Optional.ofNullable(dockerfileText)
+                        .map(is -> new InputStreamReader(is, Charsets.UTF_8))
+                        .flatMap(
+                            reader ->
+                                SneakyException.sneaky(
+                                    () ->
+                                        CharStreams.readLines(reader).stream()
+                                            .filter(line -> !line.startsWith("#"))
+                                            .filter(line -> line.startsWith("FROM"))
+                                            .findFirst()
+                                            .flatMap(
+                                                from -> {
+                                                  Pattern pattern =
+                                                      Pattern.compile("FROM (.*?):(.*)$");
+                                                  Matcher matcher = pattern.matcher(from);
+                                                  if (matcher.matches()) {
+                                                    return Optional.of(
+                                                        Tuple.of(
+                                                            matcher.group(1), matcher.group(2)));
+                                                  } else {
+                                                    return Optional.empty();
+                                                  }
+                                                })));
+                  })
+              .orElse(Tuple.of(null, null));
+      DEFAULT_IMAGE_NAME = nameAndTag.x();
+      DEFAULT_IMAGE_TAG = nameAndTag.y();
+    }
+
     private static final String DEFAULT_CONTAINER_NAME = "default";
 
     private boolean ignorePullError;
@@ -483,8 +530,8 @@ public final class TestBench implements ManagedLifecycle {
           ignorePullError,
           baseUri,
           gRPCBaseUri,
-          dockerImageName,
-          dockerImageTag,
+          requireNonNull(dockerImageName, "dockerImageName must be non null"),
+          requireNonNull(dockerImageTag, "dockerImageTag must be non null"),
           String.format("storage-testbench_%s", containerName));
     }
   }
diff --git a/google-cloud-storage/src/test/resources/com/google/cloud/storage/it/runner/registry/Dockerfile b/google-cloud-storage/src/test/resources/com/google/cloud/storage/it/runner/registry/Dockerfile
new file mode 100644
index 00000000..03141400
--- /dev/null
+++ b/google-cloud-storage/src/test/resources/com/google/cloud/storage/it/runner/registry/Dockerfile
@@ -0,0 +1,1 @@
+FROM gcr.io/cloud-devrel-public-resources/storage-testbench:v0.35.0