Repository: googleapis/java-storage
Pull request #2041 (2023-05-31 15:48:07)

docs(samples): add storage_grpc_quickstart samples

snippet-bot[bot]: <!-- probot comment [11299897]-->
Here is the summary of changes.
<details>
  <summary>You are about to add 1 region tag.</summary>

  - [samples/snippets/src/main/java/com/example/storage/QuickstartGrpcSample.java:19](https://github.com/googleapis/java-storage/blob/0d39e5c812c14846400c667d4e1d35e733f5ec68/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcSample.java#L19), tag `storage_quickstart_grpc`

</details>

---
This comment is generated by [snippet-bot](https://github.com/apps/snippet-bot).
If you find problems with this result, please file an issue at:
https://github.com/googleapis/repo-automation-bots/issues.
To update this comment, add `snippet-bot:force-run` label or use the checkbox below:
- [ ] Refresh this comment


Patch:
diff --git a/.github/.OwlBot.lock.yaml b/.github/.OwlBot.lock.yaml
index aadf54f6..47a70ea4 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:32851debfefed2b66038e0141f1b5c2103bb59ba80b7475adbc10ef7abab3de7
+# created: 2023-06-22T15:06:52.039318836Z
diff --git a/.github/dependabot.yml b/.github/dependabot.yml
index c8f413b0..203f9eac 100644
--- a/.github/dependabot.yml
+++ b/.github/dependabot.yml
@@ -5,10 +5,15 @@ 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:
+      - dependency-name: "*"
   - 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:
+      - dependency-name: "*"
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..898a8c57 100644
--- a/.github/sync-repo-settings.yaml
+++ b/.github/sync-repo-settings.yaml
@@ -8,8 +8,7 @@ branchProtectionRules:
     requiresCodeOwnerReviews: true
     requiresStrictStatusChecks: false
     requiredStatusCheckContexts:
-      - dependencies (8)
-      - dependencies (11)
+      - dependencies (17)
       - lint
       - clirr
       - units (8)
@@ -129,6 +128,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/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index e3bb26e3..65436cde 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -25,7 +25,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        java: [8, 11, 17]
+        java: [11, 17]
     steps:
     - uses: actions/checkout@v3
     - uses: actions/setup-java@v3
@@ -36,6 +36,28 @@ jobs:
     - run: .kokoro/build.sh
       env:
         JOB_TYPE: test
+  units-java8:
+    # Building using Java 17 and run the tests with Java 8 runtime
+    name: "units (8)"
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions/setup-java@v3
+        with:
+          java-version: 8
+          distribution: zulu
+      - name: "Set jvm system property environment variable for surefire plugin (unit tests)"
+        # Maven surefire plugin (unit tests) allows us to specify JVM to run the tests.
+        # https://maven.apache.org/surefire/maven-surefire-plugin/test-mojo.html#jvm
+        run: echo "SUREFIRE_JVM_OPT=-Djvm=${JAVA_HOME}/bin/java" >> $GITHUB_ENV
+        shell: bash
+      - uses: actions/setup-java@v3
+        with:
+          java-version: 17
+          distribution: zulu
+      - run: .kokoro/build.sh
+        env:
+          JOB_TYPE: test
   windows:
     runs-on: windows-latest
     steps:
@@ -54,7 +76,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        java: [8, 11, 17]
+        java: [17]
     steps:
     - uses: actions/checkout@v3
     - uses: actions/setup-java@v3
diff --git a/.kokoro/build.sh b/.kokoro/build.sh
index f105d675..49069482 100755
--- a/.kokoro/build.sh
+++ b/.kokoro/build.sh
@@ -47,7 +47,8 @@ set +e
 
 case ${JOB_TYPE} in
 test)
-    mvn test -B -ntp -Dclirr.skip=true -Denforcer.skip=true
+    echo "SUREFIRE_JVM_OPT: ${SUREFIRE_JVM_OPT}"
+    mvn test -B -ntp -Dclirr.skip=true -Denforcer.skip=true ${SUREFIRE_JVM_OPT}
     RETURN_CODE=$?
     ;;
 lint)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index cbe9b317..b4cf5918 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,19 @@
 # Changelog
 
+## [2.23.0](https://github.com/googleapis/java-storage/compare/v2.22.4...v2.23.0) (2023-06-21)
+
+
+### Features
+
+* Add new dedup utility method to Option classes ([#2063](https://github.com/googleapis/java-storage/issues/2063)) ([2ad196c](https://github.com/googleapis/java-storage/commit/2ad196c063e67c7efdac344792b67de3479d789d))
+
+
+### Dependencies
+
+* Update dependencies io.grpc:* to v1.56.0 ([#2072](https://github.com/googleapis/java-storage/issues/2072)) ([e10bde2](https://github.com/googleapis/java-storage/commit/e10bde26416bcf17401516e43949e12246f4831c))
+* Update dependency com.google.apis:google-api-services-storage to v1-rev20230617-2.0.0 ([#2077](https://github.com/googleapis/java-storage/issues/2077)) ([9f618cd](https://github.com/googleapis/java-storage/commit/9f618cddbeb471f7bd0f2332c70e501afbaccc36))
+* Update dependency org.graalvm.buildtools:native-maven-plugin to v0.9.23 ([#2074](https://github.com/googleapis/java-storage/issues/2074)) ([427f330](https://github.com/googleapis/java-storage/commit/427f330793a20b0c3da4cbe5e85984a0df508c79))
+
 ## [2.22.4](https://github.com/googleapis/java-storage/compare/v2.22.3...v2.22.4) (2023-06-07)
 
 
diff --git a/README.md b/README.md
index 1b1d586b..4079c42b 100644
--- a/README.md
+++ b/README.md
@@ -42,7 +42,7 @@ If you are using Maven without the BOM, add this to your dependencies:
 <dependency>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage</artifactId>
-  <version>2.22.4</version>
+  <version>2.23.0</version>
 </dependency>
 
 ```
@@ -57,13 +57,13 @@ implementation 'com.google.cloud:google-cloud-storage'
 If you are using Gradle without BOM, add this to your dependencies:
 
 ```Groovy
-implementation 'com.google.cloud:google-cloud-storage:2.22.4'
+implementation 'com.google.cloud:google-cloud-storage:2.23.0'
 ```
 
 If you are using SBT, add this to your dependencies:
 
 ```Scala
-libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.22.4"
+libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.23.0"
 ```
 <!-- {x-version-update-end} -->
 
@@ -224,6 +224,8 @@ Samples are in the [`samples/`](https://github.com/googleapis/java-storage/tree/
 | Configure Retries | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/ConfigureRetries.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/ConfigureRetries.java) |
 | Generate Signed Post Policy V4 | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/GenerateSignedPostPolicyV4.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/GenerateSignedPostPolicyV4.java) |
 | Get Service Account | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/GetServiceAccount.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/GetServiceAccount.java) |
+| Quickstart Grpc Dp Sample | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcDpSample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/QuickstartGrpcDpSample.java) |
+| Quickstart Grpc Sample | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcSample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/QuickstartGrpcSample.java) |
 | Quickstart Sample | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/QuickstartSample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/QuickstartSample.java) |
 | Add Bucket Default Owner | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/bucket/AddBucketDefaultOwner.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/bucket/AddBucketDefaultOwner.java) |
 | Add Bucket Iam Conditional Binding | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/bucket/AddBucketIamConditionalBinding.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/bucket/AddBucketIamConditionalBinding.java) |
@@ -426,7 +428,7 @@ Java is a registered trademark of Oracle and/or its affiliates.
 [kokoro-badge-link-5]: http://storage.googleapis.com/cloud-devrel-public/java/badges/java-storage/java11.html
 [stability-image]: https://img.shields.io/badge/stability-stable-green
 [maven-version-image]: https://img.shields.io/maven-central/v/com.google.cloud/google-cloud-storage.svg
-[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.22.4
+[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.23.0
 [authentication]: https://github.com/googleapis/google-cloud-java#authentication
 [auth-scopes]: https://developers.google.com/identity/protocols/oauth2/scopes
 [predefined-iam-roles]: https://cloud.google.com/iam/docs/understanding-roles#predefined_roles
diff --git a/gapic-google-cloud-storage-v2/pom.xml b/gapic-google-cloud-storage-v2/pom.xml
index e8aa83e0..863c8ff7 100644
--- a/gapic-google-cloud-storage-v2/pom.xml
+++ b/gapic-google-cloud-storage-v2/pom.xml
@@ -4,13 +4,13 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>gapic-google-cloud-storage-v2</artifactId>
-  <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
+  <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
   <name>gapic-google-cloud-storage-v2</name>
   <description>GRPC library for gapic-google-cloud-storage-v2</description>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/google-cloud-storage-bom/pom.xml b/google-cloud-storage-bom/pom.xml
index 356eca8e..19635d94 100644
--- a/google-cloud-storage-bom/pom.xml
+++ b/google-cloud-storage-bom/pom.xml
@@ -19,7 +19,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage-bom</artifactId>
-  <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   <packaging>pom</packaging>
   <parent>
     <groupId>com.google.cloud</groupId>
@@ -69,22 +69,22 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage</artifactId>
-        <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+        <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>gapic-google-cloud-storage-v2</artifactId>
-        <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
+        <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>grpc-google-cloud-storage-v2</artifactId>
-        <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+        <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-v2</artifactId>
-        <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
+        <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
       </dependency>
     </dependencies>
   </dependencyManagement>
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index aa16af16..9b73fd1e 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -2,7 +2,7 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
   <modelVersion>4.0.0</modelVersion>
   <artifactId>google-cloud-storage</artifactId>
-  <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   <packaging>jar</packaging>
   <name>Google Cloud Storage</name>
   <url>https://github.com/googleapis/java-storage</url>
@@ -12,11 +12,11 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <properties>
     <site.installationModule>google-cloud-storage</site.installationModule>
-    <pubsub-proto.version>1.105.14</pubsub-proto.version>
+    <pubsub-proto.version>1.105.15</pubsub-proto.version>
     <junit-platform.version>5.9.3</junit-platform.version>
   </properties>
   <dependencies>
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/GrpcStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
index 1b69e673..d623745a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
@@ -371,9 +371,6 @@ public final class GrpcStorageOptions extends StorageOptions
     /**
      * Option which signifies the client should attempt to connect to gcs via Direct Path.
      *
-     * <p>In order to use direct path, both this option must be true and the environment variable
-     * (not system property) {@code GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS} must be true.
-     *
      * <p><i>NOTE</i>There is no need to specify a new endpoint via {@link #setHost(String)} as the
      * underlying code will translate the normal {@code https://storage.googleapis.com:443} into the
      * proper Direct Path URI for you.
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/grpc-google-cloud-storage-v2/pom.xml b/grpc-google-cloud-storage-v2/pom.xml
index 1a6e1417..3844d4b1 100644
--- a/grpc-google-cloud-storage-v2/pom.xml
+++ b/grpc-google-cloud-storage-v2/pom.xml
@@ -4,13 +4,13 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>grpc-google-cloud-storage-v2</artifactId>
-  <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+  <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
   <name>grpc-google-cloud-storage-v2</name>
   <description>GRPC library for grpc-google-cloud-storage-v2</description>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/pom.xml b/pom.xml
index b7444791..20693897 100644
--- a/pom.xml
+++ b/pom.xml
@@ -4,7 +4,7 @@
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage-parent</artifactId>
   <packaging>pom</packaging>
-  <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   <name>Storage Parent</name>
   <url>https://github.com/googleapis/java-storage</url>
   <description>
@@ -54,33 +54,46 @@
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
     <github.global.server>github</github.global.server>
     <site.installationModule>google-cloud-storage-parent</site.installationModule>
-    <google.cloud.shared-dependencies.version>3.11.0</google.cloud.shared-dependencies.version>
+    <google.cloud.shared-dependencies.version>3.12.0</google.cloud.shared-dependencies.version>
   </properties>
 
   <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>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage</artifactId>
-        <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+        <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
       </dependency>
       <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>
         <artifactId>google-cloud-pubsub</artifactId>
-        <version>1.123.14</version>
+        <version>1.123.15</version>
         <scope>test</scope>
       </dependency>
       <dependency>
@@ -111,17 +124,17 @@
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-v2</artifactId>
-        <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
+        <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>grpc-google-cloud-storage-v2</artifactId>
-        <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+        <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>gapic-google-cloud-storage-v2</artifactId>
-        <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+        <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
       </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/proto-google-cloud-storage-v2/pom.xml b/proto-google-cloud-storage-v2/pom.xml
index dc5d1647..1c6a5e2d 100644
--- a/proto-google-cloud-storage-v2/pom.xml
+++ b/proto-google-cloud-storage-v2/pom.xml
@@ -4,13 +4,13 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>proto-google-cloud-storage-v2</artifactId>
-  <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
+  <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
   <name>proto-google-cloud-storage-v2</name>
   <description>PROTO library for proto-google-cloud-storage-v2</description>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/samples/install-without-bom/pom.xml b/samples/install-without-bom/pom.xml
index 6ad8b2bb..4dcfea2c 100644
--- a/samples/install-without-bom/pom.xml
+++ b/samples/install-without-bom/pom.xml
@@ -30,7 +30,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.22.4</version>
+      <version>2.23.0</version>
     </dependency>
     <!-- [END storage_install_without_bom] -->
 
@@ -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>
@@ -61,7 +61,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.123.14</version>
+      <version>1.123.15</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
diff --git a/samples/native-image-sample/pom.xml b/samples/native-image-sample/pom.xml
index af1943d5..cf1645b9 100644
--- a/samples/native-image-sample/pom.xml
+++ b/samples/native-image-sample/pom.xml
@@ -55,13 +55,13 @@
     <dependency>
       <groupId>com.google.truth</groupId>
       <artifactId>truth</artifactId>
-      <version>1.1.4</version>
+      <version>1.1.5</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.123.14</version>
+      <version>1.123.15</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
@@ -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..db00cbfb 100644
--- a/samples/snapshot/pom.xml
+++ b/samples/snapshot/pom.xml
@@ -28,7 +28,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+      <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
     </dependency>
 
     <dependency>
@@ -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>
@@ -52,7 +52,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.123.14</version>
+      <version>1.123.15</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml
index 95f581a7..bf5267cb 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>
@@ -72,7 +72,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.123.14</version>
+      <version>1.123.15</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
diff --git a/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcDpSample.java b/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcDpSample.java
new file mode 100644
index 00000000..a657300c
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcDpSample.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2023 Google Inc.
+ *
+ * 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.example.storage;
+
+// [START storage_grpc_quickstart_dp]
+// Imports the Google Cloud client library
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+
+public class QuickstartGrpcDpSample {
+  public static void main(String... args) throws Exception {
+
+    // Create an instance of options which will use the Google Cloud Storage gRPC API for all
+    // operations
+    StorageOptions options = StorageOptions.grpc().setAttemptDirectPath(true).build();
+
+    // Instantiates a client in a try-with-resource to automatically cleanup underlying resources
+    try (Storage storage = options.getService()) {
+      // The name for the new bucket
+      String bucketName = args[0]; // "my-new-bucket";
+
+      // Creates the new bucket using a request to the gRPC API via DirectPath
+      Bucket bucket = storage.create(BucketInfo.of(bucketName));
+
+      System.out.printf("Bucket %s created.%n", bucket.getName());
+    }
+  }
+}
+// [END storage_grpc_quickstart_dp]
diff --git a/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcSample.java b/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcSample.java
new file mode 100644
index 00000000..84c737ba
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcSample.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2023 Google Inc.
+ *
+ * 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.example.storage;
+
+// [START storage_grpc_quickstart]
+// Imports the Google Cloud client library
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+
+public class QuickstartGrpcSample {
+  public static void main(String... args) throws Exception {
+
+    // Create an instance of options which will use the Google Cloud Storage gRPC API for all
+    // operations
+    StorageOptions options = StorageOptions.grpc().build();
+
+    // Instantiates a client in a try-with-resource to automatically cleanup underlying resources
+    try (Storage storage = options.getService()) {
+      // The name for the new bucket
+      String bucketName = args[0]; // "my-new-bucket";
+
+      // Creates the new bucket using a request to the gRPC API
+      Bucket bucket = storage.create(BucketInfo.of(bucketName));
+
+      System.out.printf("Bucket %s created.%n", bucket.getName());
+    }
+  }
+}
+// [END storage_grpc_quickstart]
diff --git a/versions.txt b/versions.txt
index 84a23716..819d5aaf 100644
--- a/versions.txt
+++ b/versions.txt
@@ -1,7 +1,7 @@
 # Format:
 # module:released-version:current-version
 
-google-cloud-storage:2.22.4:2.22.5-SNAPSHOT
-gapic-google-cloud-storage-v2:2.22.4-alpha:2.22.5-alpha-SNAPSHOT
-grpc-google-cloud-storage-v2:2.22.4-alpha:2.22.5-alpha-SNAPSHOT
-proto-google-cloud-storage-v2:2.22.4-alpha:2.22.5-alpha-SNAPSHOT
+google-cloud-storage:2.23.0:2.23.1-SNAPSHOT
+gapic-google-cloud-storage-v2:2.23.0-alpha:2.23.1-alpha-SNAPSHOT
+grpc-google-cloud-storage-v2:2.23.0-alpha:2.23.1-alpha-SNAPSHOT
+proto-google-cloud-storage-v2:2.23.0-alpha:2.23.1-alpha-SNAPSHOT


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
diff --git a/samples/snippets/src/test/java/com/example/storage/QuickstartSampleIT.java b/samples/snippets/src/test/java/com/example/storage/QuickstartSampleIT.java
index 4c6a1e42..0fb5c058 100644
--- a/samples/snippets/src/test/java/com/example/storage/QuickstartSampleIT.java
+++ b/samples/snippets/src/test/java/com/example/storage/QuickstartSampleIT.java
@@ -38,14 +38,14 @@ public class QuickstartSampleIT {
 
   private String bucketName;
 
-  private static final void deleteBucket(String bucketName) {
+  private static void deleteBucket(String bucketName) {
     Storage storage = StorageOptions.getDefaultInstance().getService();
     storage.delete(bucketName);
   }
 
   @Before
   public void setUp() {
-    bucketName = "my-new-bucket-" + UUID.randomUUID().toString();
+    bucketName = "my-new-bucket-" + UUID.randomUUID();
   }
 
   @After
@@ -59,4 +59,18 @@ public class QuickstartSampleIT {
     String got = stdOutCaptureRule.getCapturedOutputAsUtf8String();
     assertThat(got).contains(String.format("Bucket %s created.", bucketName));
   }
+
+  @Test
+  public void testQuickstartGrpc() throws Exception {
+    QuickstartGrpcSample.main(bucketName);
+    String got = stdOutCaptureRule.getCapturedOutputAsUtf8String();
+    assertThat(got).contains(String.format("Bucket %s created.", bucketName));
+  }
+
+  @Test
+  public void testQuickstartGrpcDp() throws Exception {
+    QuickstartGrpcDpSample.main(bucketName);
+    String got = stdOutCaptureRule.getCapturedOutputAsUtf8String();
+    assertThat(got).contains(String.format("Bucket %s created.", bucketName));
+  }
 }