Repository: googleapis/java-storage
Pull request #2554 (2024-05-17 23:03:52)

test: add mockito tests
Re-adds most of the coverage that was lost when all the EasyMock tests were deleted, but as Mockito tests


Patch:
diff --git a/.github/.OwlBot.lock.yaml b/.github/.OwlBot.lock.yaml
index 6983bb26..5db36a5f 100644
--- a/.github/.OwlBot.lock.yaml
+++ b/.github/.OwlBot.lock.yaml
@@ -13,5 +13,5 @@
 # limitations under the License.
 docker:
   image: gcr.io/cloud-devrel-public-resources/owlbot-java:latest
-  digest: sha256:25b384ee1674eda3984ec41c15b514a63bbeb5eda4d57c73c7e6f5adef2fd2f1
-# created: 2024-04-05T19:12:34.133475268Z
+  digest: sha256:68ba5f5164a4b55529d358bb262feaa000536a0c62980727dd05a91bbb47ea5e
+# created: 2024-05-09T16:31:37.168667071Z
diff --git a/.github/workflows/approve-readme.yaml b/.github/workflows/approve-readme.yaml
index f5fc7d51..59f00b8e 100644
--- a/.github/workflows/approve-readme.yaml
+++ b/.github/workflows/approve-readme.yaml
@@ -21,7 +21,7 @@ jobs:
     runs-on: ubuntu-latest
     if: github.repository_owner == 'googleapis' && github.head_ref == 'autosynth-readme'
     steps:
-      - uses: actions/github-script@v6
+      - uses: actions/github-script@v7
         with:
           github-token: ${{secrets.YOSHI_APPROVER_TOKEN}}
           script: |
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index ae66b197..b91fa381 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -27,8 +27,8 @@ jobs:
       matrix:
         java: [11, 17, 21]
     steps:
-    - uses: actions/checkout@v3
-    - uses: actions/setup-java@v3
+    - uses: actions/checkout@v4
+    - uses: actions/setup-java@v4
       with:
         distribution: temurin
         java-version: ${{matrix.java}}
@@ -41,8 +41,8 @@ jobs:
     name: "units (8)"
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
-      - uses: actions/setup-java@v3
+      - uses: actions/checkout@v4
+      - uses: actions/setup-java@v4
         with:
           java-version: 8
           distribution: temurin
@@ -51,7 +51,7 @@ jobs:
         # 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
+      - uses: actions/setup-java@v4
         with:
           java-version: 17
           distribution: temurin
@@ -63,8 +63,8 @@ jobs:
     steps:
     - name: Support longpaths
       run: git config --system core.longpaths true
-    - uses: actions/checkout@v3
-    - uses: actions/setup-java@v3
+    - uses: actions/checkout@v4
+    - uses: actions/setup-java@v4
       with:
         distribution: temurin
         java-version: 8
@@ -78,8 +78,8 @@ jobs:
       matrix:
         java: [17]
     steps:
-    - uses: actions/checkout@v3
-    - uses: actions/setup-java@v3
+    - uses: actions/checkout@v4
+    - uses: actions/setup-java@v4
       with:
         distribution: temurin
         java-version: ${{matrix.java}}
@@ -88,8 +88,8 @@ jobs:
   javadoc:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v3
-    - uses: actions/setup-java@v3
+    - uses: actions/checkout@v4
+    - uses: actions/setup-java@v4
       with:
         distribution: temurin
         java-version: 17
@@ -100,8 +100,8 @@ jobs:
   lint:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v3
-    - uses: actions/setup-java@v3
+    - uses: actions/checkout@v4
+    - uses: actions/setup-java@v4
       with:
         distribution: temurin
         java-version: 11
@@ -112,8 +112,8 @@ jobs:
   clirr:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v3
-    - uses: actions/setup-java@v3
+    - uses: actions/checkout@v4
+    - uses: actions/setup-java@v4
       with:
         distribution: temurin
         java-version: 8
diff --git a/.github/workflows/renovate_config_check.yaml b/.github/workflows/renovate_config_check.yaml
index 87d8eb2b..7c5ec786 100644
--- a/.github/workflows/renovate_config_check.yaml
+++ b/.github/workflows/renovate_config_check.yaml
@@ -14,7 +14,7 @@ jobs:
       uses: actions/checkout@v4
 
     - name: Set up Node.js
-      uses: actions/setup-node@v3
+      uses: actions/setup-node@v4
       with:
         node-version: '20'
 
diff --git a/.github/workflows/samples.yaml b/.github/workflows/samples.yaml
index 10d252d7..03b29395 100644
--- a/.github/workflows/samples.yaml
+++ b/.github/workflows/samples.yaml
@@ -20,8 +20,8 @@ jobs:
   checkstyle:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
-      - uses: actions/setup-java@v3
+      - uses: actions/checkout@v4
+      - uses: actions/setup-java@v4
         with:
           distribution: temurin
           java-version: 8
diff --git a/.github/workflows/unmanaged_dependency_check.yaml b/.github/workflows/unmanaged_dependency_check.yaml
index f15d6521..50b31a20 100644
--- a/.github/workflows/unmanaged_dependency_check.yaml
+++ b/.github/workflows/unmanaged_dependency_check.yaml
@@ -17,6 +17,6 @@ jobs:
           # repository
           .kokoro/build.sh
       - name: Unmanaged dependency check
-        uses: googleapis/sdk-platform-java/java-shared-dependencies/unmanaged-dependency-check@google-cloud-shared-dependencies/v3.30.0
+        uses: googleapis/sdk-platform-java/java-shared-dependencies/unmanaged-dependency-check@google-cloud-shared-dependencies/v3.30.1
         with:
           bom-path: google-cloud-storage-bom/pom.xml
diff --git a/.kokoro/presubmit/graalvm-native-17.cfg b/.kokoro/presubmit/graalvm-native-17.cfg
index 8bcc468e..c283c89e 100644
--- a/.kokoro/presubmit/graalvm-native-17.cfg
+++ b/.kokoro/presubmit/graalvm-native-17.cfg
@@ -3,7 +3,7 @@
 # Configure the docker image for kokoro-trampoline.
 env_vars: {
   key: "TRAMPOLINE_IMAGE"
-  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_b:3.30.0"
+  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_b:3.30.1"
 }
 
 env_vars: {
diff --git a/.kokoro/presubmit/graalvm-native.cfg b/.kokoro/presubmit/graalvm-native.cfg
index 65d8c499..c997432f 100644
--- a/.kokoro/presubmit/graalvm-native.cfg
+++ b/.kokoro/presubmit/graalvm-native.cfg
@@ -3,7 +3,7 @@
 # Configure the docker image for kokoro-trampoline.
 env_vars: {
   key: "TRAMPOLINE_IMAGE"
-  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_a:3.30.0"
+  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_a:3.30.1"
 }
 
 env_vars: {
diff --git a/.readme-partials.yaml b/.readme-partials.yaml
index 3858d893..2f8c71d3 100644
--- a/.readme-partials.yaml
+++ b/.readme-partials.yaml
@@ -32,19 +32,19 @@ custom_content: |
   <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.38.1-beta-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+      <version>2.39.0-beta</version><!-- {x-version-update:google-cloud-storage-control:current} -->
   </dependency>
   ```
   
   If you are using Gradle 5.x or later, add this to your dependencies:
   ```Groovy
-  implementation platform('com.google.cloud:libraries-bom:2.38.0')
+  implementation platform('com.google.cloud:libraries-bom:2.39.0')
   implementation 'com.google.cloud:google-cloud-storage-control'
   ```
   If you are using Gradle without BOM, add this to your dependencies:
   
   ```Groovy
-  implementation 'com.google.cloud:google-cloud-storage-control:2.38.1-beta-SNAPSHOT' <!-- {x-version-update:google-cloud-storage-control:current} -->
+  implementation 'com.google.cloud:google-cloud-storage-control:2.39.0-beta' <!-- {x-version-update:google-cloud-storage-control:current} -->
   ```
   
   #### Creating an authorized service object
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a0634b2f..4d94b38a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,23 @@
 # Changelog
 
+## [2.39.0](https://github.com/googleapis/java-storage/compare/v2.38.0...v2.39.0) (2024-05-22)
+
+
+### Features
+
+* Plumb PartNamingStrategy for Parallel Composite Uploads in Transfer Manager ([#2547](https://github.com/googleapis/java-storage/issues/2547)) ([79d721d](https://github.com/googleapis/java-storage/commit/79d721dc44d1c5f4b2df3697f62b958a3b9b999c))
+
+
+### Bug Fixes
+
+* Update GapicUnbufferedChunkedResumableWritableByteChannel to be tolerant of non-quantum writes ([#2537](https://github.com/googleapis/java-storage/issues/2537)) ([1701fde](https://github.com/googleapis/java-storage/commit/1701fde04a80aa8ce97c5e762158406024456782))
+
+
+### Dependencies
+
+* Update dependency com.google.cloud:sdk-platform-java-config to v3.30.1 ([#2550](https://github.com/googleapis/java-storage/issues/2550)) ([e9807ec](https://github.com/googleapis/java-storage/commit/e9807ecda46615c8d1108bcea7a25ecba8154b6c))
+* Update dependency org.graalvm.buildtools:native-maven-plugin to v0.10.2 ([#2552](https://github.com/googleapis/java-storage/issues/2552)) ([a207829](https://github.com/googleapis/java-storage/commit/a207829040c176909141de100e4273f5aa32e78f))
+
 ## [2.38.0](https://github.com/googleapis/java-storage/compare/v2.37.0...v2.38.0) (2024-05-09)
 
 
diff --git a/README.md b/README.md
index fbd25481..1a1f2b9d 100644
--- a/README.md
+++ b/README.md
@@ -19,7 +19,7 @@ If you are using Maven with [BOM][libraries-bom], add this to your pom.xml file:
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>libraries-bom</artifactId>
-      <version>26.38.0</version>
+      <version>26.39.0</version>
       <type>pom</type>
       <scope>import</scope>
     </dependency>
@@ -46,12 +46,12 @@ 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.37.0</version>
+  <version>2.38.0</version>
 </dependency>
 <dependency>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage-control</artifactId>
-  <version>2.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.39.0</version><!-- {x-version-update:google-cloud-storage:current} -->
 </dependency>
 
 ```
@@ -141,19 +141,19 @@ If you are using Maven without the BOM, add this to your dependencies:
 <dependency>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-control</artifactId>
-    <version>2.38.1-beta-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+    <version>2.39.0-beta</version><!-- {x-version-update:google-cloud-storage-control:current} -->
 </dependency>
 ```
 
 If you are using Gradle 5.x or later, add this to your dependencies:
 ```Groovy
-implementation platform('com.google.cloud:libraries-bom:2.38.0')
+implementation platform('com.google.cloud:libraries-bom:2.39.0')
 implementation 'com.google.cloud:google-cloud-storage-control'
 ```
 If you are using Gradle without BOM, add this to your dependencies:
 
 ```Groovy
-implementation 'com.google.cloud:google-cloud-storage-control:2.38.1-beta-SNAPSHOT' <!-- {x-version-update:google-cloud-storage-control:current} -->
+implementation 'com.google.cloud:google-cloud-storage-control:2.39.0-beta' <!-- {x-version-update:google-cloud-storage-control:current} -->
 ```
 
 #### Creating an authorized service object
diff --git a/gapic-google-cloud-storage-v2/pom.xml b/gapic-google-cloud-storage-v2/pom.xml
index 605cdd06..fcc442cd 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.38.1-alpha-SNAPSHOT</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
+  <version>2.39.0-alpha</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.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.39.0</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 c2e8f527..0c56765c 100644
--- a/google-cloud-storage-bom/pom.xml
+++ b/google-cloud-storage-bom/pom.xml
@@ -19,12 +19,12 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage-bom</artifactId>
-  <version>2.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.39.0</version><!-- {x-version-update:google-cloud-storage:current} -->
   <packaging>pom</packaging>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>sdk-platform-java-config</artifactId>
-    <version>3.30.0</version>
+    <version>3.30.1</version>
     <relativePath/>
   </parent>
 
@@ -69,37 +69,37 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage</artifactId>
-        <version>2.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+        <version>2.39.0</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.38.1-alpha-SNAPSHOT</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
+        <version>2.39.0-alpha</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.38.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+        <version>2.39.0-alpha</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.38.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
+        <version>2.39.0-alpha</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage-control</artifactId>
-        <version>2.38.1-beta-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+        <version>2.39.0-beta</version><!-- {x-version-update:google-cloud-storage-control:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>grpc-google-cloud-storage-control-v2</artifactId>
-        <version>2.38.1-beta-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
+        <version>2.39.0-beta</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-control-v2</artifactId>
-        <version>2.38.1-beta-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
+        <version>2.39.0-beta</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
       </dependency>
     </dependencies>
   </dependencyManagement>
diff --git a/google-cloud-storage-control/pom.xml b/google-cloud-storage-control/pom.xml
index c131e66b..f1f24952 100644
--- a/google-cloud-storage-control/pom.xml
+++ b/google-cloud-storage-control/pom.xml
@@ -5,13 +5,13 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-control</artifactId>
-    <version>2.38.1-beta-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+    <version>2.39.0-beta</version><!-- {x-version-update:google-cloud-storage-control:current} -->
     <name>google-cloud-storage-control</name>
     <description>GRPC library for google-cloud-storage-control</description>
     <parent>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage-parent</artifactId>
-        <version>2.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+        <version>2.39.0</version><!-- {x-version-update:google-cloud-storage:current} -->
     </parent>
 
     <dependencies>
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index 830b4f18..213bebcd 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.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.39.0</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.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.39.0</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <properties>
     <site.installationModule>google-cloud-storage</site.installationModule>
-    <pubsub-proto.version>1.111.3</pubsub-proto.version>
+    <pubsub-proto.version>1.111.5</pubsub-proto.version>
     <junit-platform.version>5.10.2</junit-platform.version>
   </properties>
   <dependencies>
@@ -181,13 +181,13 @@
     <dependency>
       <groupId>com.google.api.grpc</groupId>
       <artifactId>proto-google-cloud-kms-v1</artifactId>
-      <version>0.136.0</version>
+      <version>0.137.0</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-kms</artifactId>
-      <version>2.45.0</version>
+      <version>2.46.0</version>
       <scope>test</scope>
     </dependency>
     <dependency>
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ChunkSegmenter.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ChunkSegmenter.java
index 50e3d7e6..774c1e0a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ChunkSegmenter.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ChunkSegmenter.java
@@ -18,7 +18,10 @@ package com.google.cloud.storage;
 
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.math.IntMath;
 import com.google.protobuf.ByteString;
+import java.math.RoundingMode;
 import java.nio.ByteBuffer;
 import java.util.ArrayDeque;
 import java.util.Deque;
@@ -40,6 +43,13 @@ final class ChunkSegmenter {
 
   @VisibleForTesting
   ChunkSegmenter(Hasher hasher, ByteStringStrategy bss, int maxSegmentSize, int blockSize) {
+    int mod = maxSegmentSize % blockSize;
+    Preconditions.checkArgument(
+        mod == 0,
+        "maxSegmentSize % blockSize == 0 (%s % %s == %s)",
+        maxSegmentSize,
+        blockSize,
+        mod);
     this.hasher = hasher;
     this.bss = bss;
     this.maxSegmentSize = maxSegmentSize;
@@ -79,32 +89,92 @@ final class ChunkSegmenter {
   }
 
   ChunkSegment[] segmentBuffers(ByteBuffer[] bbs, int offset, int length) {
+    return segmentBuffers(bbs, offset, length, true);
+  }
+
+  ChunkSegment[] segmentBuffers(
+      ByteBuffer[] bbs, int offset, int length, boolean allowUnalignedBlocks) {
+    // turn this into a single branch, rather than multiple that would need to be checked each
+    // element of the iteration
+    if (allowUnalignedBlocks) {
+      return segmentWithUnaligned(bbs, offset, length);
+    } else {
+      return segmentWithoutUnaligned(bbs, offset, length);
+    }
+  }
+
+  private ChunkSegment[] segmentWithUnaligned(ByteBuffer[] bbs, int offset, int length) {
+    Deque<ChunkSegment> data = new ArrayDeque<>();
+
+    for (int i = offset; i < length; i++) {
+      ByteBuffer buffer = bbs[i];
+      int remaining;
+      while ((remaining = buffer.remaining()) > 0) {
+        consumeBytes(data, remaining, buffer);
+      }
+    }
+
+    return data.toArray(new ChunkSegment[0]);
+  }
+
+  private ChunkSegment[] segmentWithoutUnaligned(ByteBuffer[] bbs, int offset, int length) {
     Deque<ChunkSegment> data = new ArrayDeque<>();
 
+    final long totalRemaining = Buffers.totalRemaining(bbs, offset, length);
+    long consumedSoFar = 0;
+
+    int currentBlockPending = blockSize;
+
     for (int i = offset; i < length; i++) {
       ByteBuffer buffer = bbs[i];
       int remaining;
       while ((remaining = buffer.remaining()) > 0) {
-        // either no chunk or most recent chunk is full, start a new one
-        ChunkSegment peekLast = data.peekLast();
-        if (peekLast == null || peekLast.b.size() == maxSegmentSize) {
-          int limit = Math.min(remaining, maxSegmentSize);
-          ChunkSegment datum = newSegment(buffer, limit);
-          data.addLast(datum);
+        long overallRemaining = totalRemaining - consumedSoFar;
+        if (overallRemaining < blockSize && currentBlockPending == blockSize) {
+          break;
+        }
+
+        int numBytesConsumable;
+        if (remaining >= blockSize) {
+          int blockCount = IntMath.divide(remaining, blockSize, RoundingMode.DOWN);
+          numBytesConsumable = blockCount * blockSize;
+        } else if (currentBlockPending < blockSize) {
+          numBytesConsumable = currentBlockPending;
+          currentBlockPending = blockSize;
         } else {
-          ChunkSegment chunkSoFar = data.pollLast();
-          //noinspection ConstantConditions -- covered by peekLast check above
-          int limit = Math.min(remaining, maxSegmentSize - chunkSoFar.b.size());
-          ChunkSegment datum = newSegment(buffer, limit);
-          ChunkSegment plus = chunkSoFar.concat(datum);
-          data.addLast(plus);
+          numBytesConsumable = remaining;
+          currentBlockPending = currentBlockPending - remaining;
+        }
+        if (numBytesConsumable <= 0) {
+          continue;
         }
+
+        consumedSoFar += consumeBytes(data, numBytesConsumable, buffer);
       }
     }
 
     return data.toArray(new ChunkSegment[0]);
   }
 
+  private long consumeBytes(Deque<ChunkSegment> data, int numBytesConsumable, ByteBuffer buffer) {
+    // either no chunk or most recent chunk is full, start a new one
+    ChunkSegment peekLast = data.peekLast();
+    if (peekLast == null || peekLast.b.size() == maxSegmentSize) {
+      int limit = Math.min(numBytesConsumable, maxSegmentSize);
+      ChunkSegment datum = newSegment(buffer, limit);
+      data.addLast(datum);
+      return limit;
+    } else {
+      ChunkSegment chunkSoFar = data.pollLast();
+      //noinspection ConstantConditions -- covered by peekLast check above
+      int limit = Math.min(numBytesConsumable, maxSegmentSize - chunkSoFar.b.size());
+      ChunkSegment datum = newSegment(buffer, limit);
+      ChunkSegment plus = chunkSoFar.concat(datum);
+      data.addLast(plus);
+      return limit;
+    }
+  }
+
   private ChunkSegment newSegment(ByteBuffer buffer, int limit) {
     final ByteBuffer slice = buffer.slice();
     slice.limit(limit);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java
index c3c3cd09..6acac1e6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java
@@ -121,7 +121,10 @@ final class GapicUnbufferedChunkedResumableWritableByteChannel
 
     long begin = writeCtx.getConfirmedBytes().get();
     RewindableContent content = RewindableContent.of(srcs, srcsOffset, srcsLength);
-    ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength);
+    ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength, finalize);
+    if (data.length == 0) {
+      return 0;
+    }
 
     List<WriteObjectRequest> messages = new ArrayList<>();
 
diff --git a/grpc-google-cloud-storage-control-v2/pom.xml b/grpc-google-cloud-storage-control-v2/pom.xml
index f7abf20d..dd513cc3 100644
--- a/grpc-google-cloud-storage-control-v2/pom.xml
+++ b/grpc-google-cloud-storage-control-v2/pom.xml
@@ -4,13 +4,13 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>grpc-google-cloud-storage-control-v2</artifactId>
-  <version>2.38.1-beta-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
+  <version>2.39.0-beta</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
   <name>grpc-google-cloud-storage-control-v2</name>
   <description>GRPC library for google-cloud-storage</description>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.39.0</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/grpc-google-cloud-storage-v2/pom.xml b/grpc-google-cloud-storage-v2/pom.xml
index 509018e2..4f9b97ee 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.38.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+  <version>2.39.0-alpha</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.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.39.0</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/pom.xml b/pom.xml
index 86da4c26..10bb5f3f 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.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.39.0</version><!-- {x-version-update:google-cloud-storage:current} -->
   <name>Storage Parent</name>
   <url>https://github.com/googleapis/java-storage</url>
   <description>
@@ -14,7 +14,7 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>sdk-platform-java-config</artifactId>
-    <version>3.30.0</version>
+    <version>3.30.1</version>
   </parent>
 
   <developers>
@@ -76,7 +76,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage</artifactId>
-        <version>2.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+        <version>2.39.0</version><!-- {x-version-update:google-cloud-storage:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.apis</groupId>
@@ -86,7 +86,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-pubsub</artifactId>
-        <version>1.129.3</version>
+        <version>1.129.5</version>
         <scope>test</scope>
       </dependency>
       <dependency>
@@ -98,32 +98,32 @@
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-v2</artifactId>
-        <version>2.38.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
+        <version>2.39.0-alpha</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.38.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+        <version>2.39.0-alpha</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.38.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+        <version>2.39.0-alpha</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>grpc-google-cloud-storage-control-v2</artifactId>
-        <version>2.38.1-beta-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
+        <version>2.39.0-beta</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-control-v2</artifactId>
-        <version>2.38.1-beta-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
+        <version>2.39.0-beta</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage-control</artifactId>
-        <version>2.38.1-beta-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+        <version>2.39.0-beta</version><!-- {x-version-update:google-cloud-storage-control:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
diff --git a/proto-google-cloud-storage-control-v2/pom.xml b/proto-google-cloud-storage-control-v2/pom.xml
index c20dc636..fa454bb1 100644
--- a/proto-google-cloud-storage-control-v2/pom.xml
+++ b/proto-google-cloud-storage-control-v2/pom.xml
@@ -4,13 +4,13 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>proto-google-cloud-storage-control-v2</artifactId>
-  <version>2.38.1-beta-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
+  <version>2.39.0-beta</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
   <name>proto-google-cloud-storage-control-v2</name>
   <description>Proto library for  proto-google-cloud-storage-control-v2</description>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.39.0</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/proto-google-cloud-storage-v2/pom.xml b/proto-google-cloud-storage-v2/pom.xml
index 9b0942c2..dfc85cdc 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.38.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
+  <version>2.39.0-alpha</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.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.39.0</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 fe5516e8..d4893b17 100644
--- a/samples/install-without-bom/pom.xml
+++ b/samples/install-without-bom/pom.xml
@@ -30,12 +30,12 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.37.0</version>
+      <version>2.38.0</version>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+      <version>2.39.0</version><!-- {x-version-update:google-cloud-storage:current} -->
     </dependency>
     <!-- [END storage_install_without_bom] -->
 
@@ -66,13 +66,13 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.129.3</version>
+      <version>1.129.5</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.37.0-alpha</version>
+      <version>2.38.0-beta</version>
       <scope>compile</scope>
     </dependency>
   </dependencies>
@@ -83,7 +83,7 @@
       <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>build-helper-maven-plugin</artifactId>
-        <version>3.5.0</version>
+        <version>3.6.0</version>
         <executions>
           <execution>
             <id>add-snippets-source</id>
diff --git a/samples/native-image-sample/pom.xml b/samples/native-image-sample/pom.xml
index 1c2ba958..9229c8e1 100644
--- a/samples/native-image-sample/pom.xml
+++ b/samples/native-image-sample/pom.xml
@@ -29,7 +29,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>26.38.0</version>
+        <version>26.39.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
@@ -61,13 +61,13 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.129.3</version>
+      <version>1.129.5</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.38.1-beta-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+      <version>2.39.0-beta</version><!-- {x-version-update:google-cloud-storage-control:current} -->
       <scope>compile</scope>
     </dependency>
   </dependencies>
@@ -140,7 +140,7 @@
           <plugin>
             <groupId>org.graalvm.buildtools</groupId>
             <artifactId>native-maven-plugin</artifactId>
-            <version>0.10.1</version>
+            <version>0.10.2</version>
             <extensions>true</extensions>
             <configuration>
               <mainClass>com.example.storage.NativeImageStorageSample</mainClass>
diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml
index 31ad6932..2012436e 100644
--- a/samples/snapshot/pom.xml
+++ b/samples/snapshot/pom.xml
@@ -28,12 +28,12 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+      <version>2.39.0</version><!-- {x-version-update:google-cloud-storage:current} -->
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.38.1-beta-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+      <version>2.39.0-beta</version><!-- {x-version-update:google-cloud-storage-control:current} -->
       <scope>compile</scope>
     </dependency>
 
@@ -58,7 +58,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.129.3</version>
+      <version>1.129.5</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
@@ -69,7 +69,7 @@
       <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>build-helper-maven-plugin</artifactId>
-        <version>3.5.0</version>
+        <version>3.6.0</version>
         <executions>
           <execution>
             <id>add-snippets-source</id>
diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml
index d27887a4..7be5d0f9 100644
--- a/samples/snippets/pom.xml
+++ b/samples/snippets/pom.xml
@@ -31,7 +31,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>26.38.0</version>
+        <version>26.39.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
@@ -76,7 +76,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.129.3</version>
+      <version>1.129.5</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
diff --git a/storage-shared-benchmarking/pom.xml b/storage-shared-benchmarking/pom.xml
index 2314abf3..ba49f60b 100644
--- a/storage-shared-benchmarking/pom.xml
+++ b/storage-shared-benchmarking/pom.xml
@@ -10,7 +10,7 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.39.0</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
 
   <properties>
@@ -31,7 +31,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.38.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+      <version>2.39.0</version><!-- {x-version-update:google-cloud-storage:current} -->
       <classifier>tests</classifier>
     </dependency>
     <dependency>
diff --git a/versions.txt b/versions.txt
index bb209224..46958f51 100644
--- a/versions.txt
+++ b/versions.txt
@@ -1,10 +1,10 @@
 # Format:
 # module:released-version:current-version
 
-google-cloud-storage:2.38.0:2.38.1-SNAPSHOT
-gapic-google-cloud-storage-v2:2.38.0-alpha:2.38.1-alpha-SNAPSHOT
-grpc-google-cloud-storage-v2:2.38.0-alpha:2.38.1-alpha-SNAPSHOT
-proto-google-cloud-storage-v2:2.38.0-alpha:2.38.1-alpha-SNAPSHOT
-google-cloud-storage-control:2.38.0-beta:2.38.1-beta-SNAPSHOT
-proto-google-cloud-storage-control-v2:2.38.0-beta:2.38.1-beta-SNAPSHOT
-grpc-google-cloud-storage-control-v2:2.38.0-beta:2.38.1-beta-SNAPSHOT
+google-cloud-storage:2.39.0:2.39.0
+gapic-google-cloud-storage-v2:2.39.0-alpha:2.39.0-alpha
+grpc-google-cloud-storage-v2:2.39.0-alpha:2.39.0-alpha
+proto-google-cloud-storage-v2:2.39.0-alpha:2.39.0-alpha
+google-cloud-storage-control:2.39.0-beta:2.39.0-beta
+proto-google-cloud-storage-control-v2:2.39.0-beta:2.39.0-beta
+grpc-google-cloud-storage-control-v2:2.39.0-beta:2.39.0-beta


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobTest.java
new file mode 100644
index 00000000..f312837b
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobTest.java
@@ -0,0 +1,604 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.api.core.ApiClock;
+import com.google.api.gax.retrying.RetrySettings;
+import com.google.cloud.ReadChannel;
+import com.google.cloud.storage.Acl.Project;
+import com.google.cloud.storage.Acl.Project.ProjectRole;
+import com.google.cloud.storage.Acl.Role;
+import com.google.cloud.storage.Acl.User;
+import com.google.cloud.storage.Blob.BlobSourceOption;
+import com.google.cloud.storage.BlobInfo.BuilderImpl;
+import com.google.cloud.storage.Storage.CopyRequest;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.io.BaseEncoding;
+import java.io.File;
+import java.net.URL;
+import java.nio.file.Path;
+import java.security.Key;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import javax.crypto.spec.SecretKeySpec;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+public class BlobTest {
+
+  private static final Acl ACL = Acl.of(User.ofAllAuthenticatedUsers(), Role.OWNER);
+  private static final Acl OTHER_ACL = Acl.of(new Project(ProjectRole.OWNERS, "p"), Role.READER);
+  private static final List<Acl> ACLS = ImmutableList.of(ACL, OTHER_ACL);
+  private static final Integer COMPONENT_COUNT = 2;
+  private static final String CONTENT_TYPE = "text/html";
+  private static final String CACHE_CONTROL = "cache";
+  private static final String CONTENT_DISPOSITION = "content-disposition";
+  private static final String CONTENT_ENCODING = "UTF-8";
+  private static final String CONTENT_LANGUAGE = "En";
+  private static final String CRC32 = "FF00";
+  private static final String CRC32_HEX_STRING = "145d34";
+  private static final Long DELETE_TIME = System.currentTimeMillis();
+  private static final String ETAG = "0xFF00";
+  private static final Long GENERATION = 1L;
+  private static final String GENERATED_ID = "B/N:1";
+  private static final String MD5 = "FF00";
+  private static final String MD5_HEX_STRING = "145d34";
+  private static final String MEDIA_LINK = "http://media/b/n";
+  private static final Map<String, String> METADATA = ImmutableMap.of("n1", "v1", "n2", "v2");
+  private static final Long META_GENERATION = 10L;
+  private static final User OWNER = new User("user@gmail.com");
+  private static final String SELF_LINK = "http://storage/b/n";
+  private static final Long SIZE = 1024L;
+  private static final Long UPDATE_TIME = DELETE_TIME - 1L;
+  private static final Long CREATE_TIME = UPDATE_TIME - 1L;
+  private static final Long CUSTOM_TIME = CREATE_TIME - 1L;
+  private static final StorageClass STORAGE_CLASS = StorageClass.COLDLINE;
+  private static final Long TIME_STORAGE_CLASS_UPDATED = CREATE_TIME;
+  private static final String ENCRYPTION_ALGORITHM = "AES256";
+  private static final String KEY_SHA256 = "keySha";
+  private static final BlobInfo.CustomerEncryption CUSTOMER_ENCRYPTION =
+      new BlobInfo.CustomerEncryption(ENCRYPTION_ALGORITHM, KEY_SHA256);
+  private static final String KMS_KEY_NAME =
+      "projects/p/locations/kr-loc/keyRings/kr/cryptoKeys/key";
+  private static final Boolean EVENT_BASED_HOLD = true;
+  private static final Boolean TEMPORARY_HOLD = true;
+  private static final Long RETENTION_EXPIRATION_TIME = 10L;
+  private static final BlobInfo FULL_BLOB_INFO =
+      BlobInfo.newBuilder("b", "n", GENERATION)
+          .setAcl(ACLS)
+          .setComponentCount(COMPONENT_COUNT)
+          .setContentType(CONTENT_TYPE)
+          .setCacheControl(CACHE_CONTROL)
+          .setContentDisposition(CONTENT_DISPOSITION)
+          .setContentEncoding(CONTENT_ENCODING)
+          .setContentLanguage(CONTENT_LANGUAGE)
+          .setCrc32c(CRC32)
+          .setDeleteTime(DELETE_TIME)
+          .setEtag(ETAG)
+          .setGeneratedId(GENERATED_ID)
+          .setMd5(MD5)
+          .setMediaLink(MEDIA_LINK)
+          .setMetadata(METADATA)
+          .setMetageneration(META_GENERATION)
+          .setOwner(OWNER)
+          .setSelfLink(SELF_LINK)
+          .setSize(SIZE)
+          .setUpdateTime(UPDATE_TIME)
+          .setCreateTime(CREATE_TIME)
+          .setCustomTime(CUSTOM_TIME)
+          .setStorageClass(STORAGE_CLASS)
+          .setTimeStorageClassUpdated(TIME_STORAGE_CLASS_UPDATED)
+          .setCustomerEncryption(CUSTOMER_ENCRYPTION)
+          .setKmsKeyName(KMS_KEY_NAME)
+          .setEventBasedHold(EVENT_BASED_HOLD)
+          .setTemporaryHold(TEMPORARY_HOLD)
+          .setRetentionExpirationTime(RETENTION_EXPIRATION_TIME)
+          .build();
+  private static final BlobInfo BLOB_INFO =
+      BlobInfo.newBuilder("b", "n", 12345678L).setMetageneration(42L).build();
+  private static final BlobInfo BLOB_INFO_NO_GENERATION =
+      BlobInfo.newBuilder(BLOB_INFO.getBucket(), BLOB_INFO.getName())
+          .setMetageneration(42L)
+          .build();
+  private static final BlobInfo DIRECTORY_INFO =
+      BlobInfo.newBuilder("b", "n/").setSize(0L).setIsDirectory(true).build();
+  private static final String BASE64_KEY = "JVzfVl8NLD9FjedFuStegjRfES5ll5zc59CIXw572OA=";
+  private static final Key KEY =
+      new SecretKeySpec(BaseEncoding.base64().decode(BASE64_KEY), "AES256");
+
+  // This retrying setting is used by test testDownloadWithRetries. This unit test is setup
+  // to write one byte and then throw retryable exception, it then writes another bytes on
+  // second call succeeds.
+  private static final RetrySettings RETRY_SETTINGS =
+      RetrySettings.newBuilder().setMaxAttempts(2).build();
+  private static final ApiClock API_CLOCK =
+      new ApiClock() {
+        @Override
+        public long nanoTime() {
+          return 42_000_000_000L;
+        }
+
+        @Override
+        public long millisTime() {
+          return 42_000L;
+        }
+      };
+
+  private Storage storage;
+  private Blob blob;
+  private Blob expectedBlob;
+  private Storage serviceMockReturnsOptions = Mockito.mock(Storage.class);
+  private HttpStorageOptions mockOptions = Mockito.mock(HttpStorageOptions.class);
+  private final HttpRetryAlgorithmManager retryAlgorithmManager =
+      HttpStorageOptions.getDefaultInstance().getRetryAlgorithmManager();
+
+  @Before
+  public void setUp() {
+    storage = Mockito.mock(Storage.class);
+  }
+
+  private void initializeExpectedBlob() {
+    when(serviceMockReturnsOptions.getOptions()).thenReturn(mockOptions);
+    when(mockOptions.getRetryAlgorithmManager()).thenReturn(retryAlgorithmManager);
+    expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(BLOB_INFO));
+  }
+
+  private void initializeBlob() {
+    blob = new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO));
+  }
+
+  @Test
+  public void testExists_True() throws Exception {
+    initializeExpectedBlob();
+    Storage.BlobGetOption[] expectedOptions = {Storage.BlobGetOption.fields()};
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.get(expectedBlob.getBlobId(), expectedOptions)).thenReturn(expectedBlob);
+    initializeBlob();
+    assertTrue(blob.exists());
+
+    verify(storage).getOptions();
+    verify(storage).get(expectedBlob.getBlobId(), expectedOptions);
+  }
+
+  @Test
+  public void testExists_False() throws Exception {
+    Storage.BlobGetOption[] expectedOptions = {Storage.BlobGetOption.fields()};
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.get(BLOB_INFO.getBlobId(), expectedOptions)).thenReturn(null);
+    initializeBlob();
+    assertFalse(blob.exists());
+
+    verify(storage).getOptions();
+    verify(storage).get(BLOB_INFO.getBlobId(), expectedOptions);
+  }
+
+  @Test
+  public void testContent() throws Exception {
+    initializeExpectedBlob();
+    byte[] content = {1, 2};
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.readAllBytes(BLOB_INFO.getBlobId())).thenReturn(content);
+    initializeBlob();
+    assertArrayEquals(content, blob.getContent());
+
+    verify(storage).getOptions();
+    verify(storage).readAllBytes(BLOB_INFO.getBlobId());
+  }
+
+  @Test
+  public void testContentWithDecryptionKey() throws Exception {
+    initializeExpectedBlob();
+    byte[] content = {1, 2};
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.readAllBytes(
+            BLOB_INFO.getBlobId(), Storage.BlobSourceOption.decryptionKey(BASE64_KEY)))
+        .thenReturn(content);
+    initializeBlob();
+    assertArrayEquals(content, blob.getContent(BlobSourceOption.decryptionKey(BASE64_KEY)));
+    assertArrayEquals(content, blob.getContent(BlobSourceOption.decryptionKey(KEY)));
+
+    verify(storage).getOptions();
+    verify(storage, times(2))
+        .readAllBytes(BLOB_INFO.getBlobId(), Storage.BlobSourceOption.decryptionKey(BASE64_KEY));
+  }
+
+  @Test
+  public void testReload() throws Exception {
+    initializeExpectedBlob();
+    Blob expectedReloadedBlob = expectedBlob.toBuilder().setCacheControl("c").build();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.get(BLOB_INFO_NO_GENERATION.getBlobId(), new Storage.BlobGetOption[0]))
+        .thenReturn(expectedReloadedBlob);
+    initializeBlob();
+    Blob updatedBlob = blob.reload();
+    assertEquals(expectedReloadedBlob, updatedBlob);
+
+    verify(storage).getOptions();
+    verify(storage).get(BLOB_INFO_NO_GENERATION.getBlobId(), new Storage.BlobGetOption[0]);
+  }
+
+  @Test
+  public void testReloadNull() throws Exception {
+    initializeExpectedBlob();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.get(BLOB_INFO_NO_GENERATION.getBlobId(), new Storage.BlobGetOption[0]))
+        .thenReturn(null);
+    initializeBlob();
+    Blob reloadedBlob = blob.reload();
+    assertNull(reloadedBlob);
+
+    verify(storage).getOptions();
+    verify(storage).get(BLOB_INFO_NO_GENERATION.getBlobId(), new Storage.BlobGetOption[0]);
+  }
+
+  @Test
+  public void testReloadWithOptions() throws Exception {
+    initializeExpectedBlob();
+    Blob expectedReloadedBlob = expectedBlob.toBuilder().setCacheControl("c").build();
+    Storage.BlobGetOption[] options = {Storage.BlobGetOption.metagenerationMatch(42L)};
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.get(BLOB_INFO_NO_GENERATION.getBlobId(), options))
+        .thenReturn(expectedReloadedBlob);
+    initializeBlob();
+    Blob updatedBlob = blob.reload(BlobSourceOption.metagenerationMatch());
+    assertEquals(expectedReloadedBlob, updatedBlob);
+
+    verify(storage).getOptions();
+    verify(storage).get(BLOB_INFO_NO_GENERATION.getBlobId(), options);
+  }
+
+  @Test
+  public void testUpdate() throws Exception {
+    initializeExpectedBlob();
+    Blob expectedUpdatedBlob = expectedBlob.toBuilder().setCacheControl("c").build();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.update(eq(expectedUpdatedBlob), new Storage.BlobTargetOption[0]))
+        .thenReturn(expectedUpdatedBlob);
+    initializeBlob();
+    Blob updatedBlob = new Blob(storage, new BlobInfo.BuilderImpl(expectedUpdatedBlob));
+    Blob actualUpdatedBlob = updatedBlob.update();
+    assertEquals(expectedUpdatedBlob, actualUpdatedBlob);
+
+    verify(storage, times(2)).getOptions();
+    verify(storage).update(eq(expectedUpdatedBlob), new Storage.BlobTargetOption[0]);
+  }
+
+  @Test
+  public void testDelete() throws Exception {
+    initializeExpectedBlob();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.delete(BLOB_INFO.getBlobId(), new Storage.BlobSourceOption[0])).thenReturn(true);
+    initializeBlob();
+    assertTrue(blob.delete());
+
+    verify(storage).getOptions();
+    verify(storage).delete(BLOB_INFO.getBlobId(), new Storage.BlobSourceOption[0]);
+  }
+
+  @Test
+  public void testCopyToBucket() throws Exception {
+    initializeExpectedBlob();
+    BlobInfo target = BlobInfo.newBuilder(BlobId.of("bt", "n")).build();
+    CopyWriter copyWriter = Mockito.mock(CopyWriter.class);
+    ArgumentCaptor<CopyRequest> capturedCopyRequest = ArgumentCaptor.forClass(CopyRequest.class);
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.copy(capturedCopyRequest.capture())).thenReturn(copyWriter);
+    initializeBlob();
+    CopyWriter returnedCopyWriter = blob.copyTo("bt");
+    assertEquals(copyWriter, returnedCopyWriter);
+    assertEquals(BLOB_INFO_NO_GENERATION.getBlobId(), capturedCopyRequest.getValue().getSource());
+    assertEquals(target, capturedCopyRequest.getValue().getTarget());
+    assertFalse(capturedCopyRequest.getValue().overrideInfo());
+    assertTrue(capturedCopyRequest.getValue().getSourceOptions().isEmpty());
+    assertTrue(capturedCopyRequest.getValue().getTargetOptions().isEmpty());
+
+    verify(storage).getOptions();
+    verify(storage).copy(capturedCopyRequest.capture());
+  }
+
+  @Test
+  public void testCopyTo() throws Exception {
+    initializeExpectedBlob();
+    BlobInfo target = BlobInfo.newBuilder(BlobId.of("bt", "nt")).build();
+    CopyWriter copyWriter = Mockito.mock(CopyWriter.class);
+    ArgumentCaptor<CopyRequest> capturedCopyRequest = ArgumentCaptor.forClass(CopyRequest.class);
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.copy(capturedCopyRequest.capture())).thenReturn(copyWriter);
+    initializeBlob();
+    CopyWriter returnedCopyWriter = blob.copyTo("bt", "nt");
+    assertEquals(copyWriter, returnedCopyWriter);
+    assertEquals(BLOB_INFO_NO_GENERATION.getBlobId(), capturedCopyRequest.getValue().getSource());
+    assertEquals(target, capturedCopyRequest.getValue().getTarget());
+    assertFalse(capturedCopyRequest.getValue().overrideInfo());
+    assertTrue(capturedCopyRequest.getValue().getSourceOptions().isEmpty());
+    assertTrue(capturedCopyRequest.getValue().getTargetOptions().isEmpty());
+
+    verify(storage).getOptions();
+    verify(storage).copy(capturedCopyRequest.capture());
+  }
+
+  @Test
+  public void testCopyToBlobId() throws Exception {
+    initializeExpectedBlob();
+    BlobInfo target = BlobInfo.newBuilder(BlobId.of("bt", "nt")).build();
+    BlobId targetId = BlobId.of("bt", "nt");
+    CopyWriter copyWriter = Mockito.mock(CopyWriter.class);
+    ArgumentCaptor<CopyRequest> capturedCopyRequest = ArgumentCaptor.forClass(CopyRequest.class);
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.copy(capturedCopyRequest.capture())).thenReturn(copyWriter);
+    initializeBlob();
+    CopyWriter returnedCopyWriter = blob.copyTo(targetId);
+    assertEquals(copyWriter, returnedCopyWriter);
+    assertEquals(BLOB_INFO_NO_GENERATION.getBlobId(), capturedCopyRequest.getValue().getSource());
+    assertEquals(target, capturedCopyRequest.getValue().getTarget());
+    assertFalse(capturedCopyRequest.getValue().overrideInfo());
+    assertTrue(capturedCopyRequest.getValue().getSourceOptions().isEmpty());
+    assertTrue(capturedCopyRequest.getValue().getTargetOptions().isEmpty());
+
+    verify(storage).getOptions();
+    verify(storage).copy(capturedCopyRequest.capture());
+  }
+
+  @Test
+  public void testReader() throws Exception {
+    initializeExpectedBlob();
+    ReadChannel channel = Mockito.mock(ReadChannel.class);
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.reader(BLOB_INFO.getBlobId())).thenReturn(channel);
+    initializeBlob();
+    assertSame(channel, blob.reader());
+
+    verify(storage).getOptions();
+    verify(storage).reader(BLOB_INFO.getBlobId());
+  }
+
+  @Test
+  public void testReaderWithDecryptionKey() throws Exception {
+    initializeExpectedBlob();
+    ReadChannel channel = Mockito.mock(ReadChannel.class);
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.reader(BLOB_INFO.getBlobId(), Storage.BlobSourceOption.decryptionKey(BASE64_KEY)))
+        .thenReturn(channel);
+    initializeBlob();
+    assertSame(channel, blob.reader(BlobSourceOption.decryptionKey(BASE64_KEY)));
+    assertSame(channel, blob.reader(BlobSourceOption.decryptionKey(KEY)));
+
+    verify(storage).getOptions();
+    verify(storage, times(2))
+        .reader(BLOB_INFO.getBlobId(), Storage.BlobSourceOption.decryptionKey(BASE64_KEY));
+  }
+
+  @Test
+  public void testSignUrl() throws Exception {
+    initializeExpectedBlob();
+    URL url = new URL("http://localhost:123/bla");
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.signUrl(expectedBlob, 100, TimeUnit.SECONDS)).thenReturn(url);
+    initializeBlob();
+    assertEquals(url, blob.signUrl(100, TimeUnit.SECONDS));
+
+    verify(storage).getOptions();
+    verify(storage).signUrl(expectedBlob, 100, TimeUnit.SECONDS);
+  }
+
+  @Test
+  public void testGetAcl() throws Exception {
+    initializeExpectedBlob();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.getAcl(BLOB_INFO.getBlobId(), User.ofAllAuthenticatedUsers())).thenReturn(ACL);
+    initializeBlob();
+    assertEquals(ACL, blob.getAcl(User.ofAllAuthenticatedUsers()));
+
+    verify(storage).getOptions();
+    verify(storage).getAcl(BLOB_INFO.getBlobId(), User.ofAllAuthenticatedUsers());
+  }
+
+  @Test
+  public void testDeleteAcl() throws Exception {
+    initializeExpectedBlob();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.deleteAcl(BLOB_INFO.getBlobId(), User.ofAllAuthenticatedUsers())).thenReturn(true);
+    initializeBlob();
+    assertTrue(blob.deleteAcl(User.ofAllAuthenticatedUsers()));
+
+    verify(storage).getOptions();
+    verify(storage).deleteAcl(BLOB_INFO.getBlobId(), User.ofAllAuthenticatedUsers());
+  }
+
+  @Test
+  public void testCreateAcl() throws Exception {
+    initializeExpectedBlob();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
+    when(storage.createAcl(BLOB_INFO.getBlobId(), ACL)).thenReturn(returnedAcl);
+    initializeBlob();
+    assertEquals(returnedAcl, blob.createAcl(ACL));
+
+    verify(storage).getOptions();
+    verify(storage).createAcl(BLOB_INFO.getBlobId(), ACL);
+  }
+
+  @Test
+  public void testUpdateAcl() throws Exception {
+    initializeExpectedBlob();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
+    when(storage.updateAcl(BLOB_INFO.getBlobId(), ACL)).thenReturn(returnedAcl);
+    initializeBlob();
+    assertEquals(returnedAcl, blob.updateAcl(ACL));
+
+    verify(storage).getOptions();
+    verify(storage).updateAcl(BLOB_INFO.getBlobId(), ACL);
+  }
+
+  @Test
+  public void testListAcls() throws Exception {
+    initializeExpectedBlob();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.listAcls(BLOB_INFO.getBlobId())).thenReturn(ACLS);
+    initializeBlob();
+    assertEquals(ACLS, blob.listAcls());
+
+    verify(storage).getOptions();
+    verify(storage).listAcls(BLOB_INFO.getBlobId());
+  }
+
+  @Test
+  public void testToBuilder() {
+    when(storage.getOptions()).thenReturn(mockOptions);
+    Blob fullBlob = new Blob(storage, new BlobInfo.BuilderImpl(FULL_BLOB_INFO));
+    assertEquals(fullBlob, fullBlob.toBuilder().build());
+    Blob simpleBlob = new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO));
+    assertEquals(simpleBlob, simpleBlob.toBuilder().build());
+    Blob directory = new Blob(storage, new BlobInfo.BuilderImpl(DIRECTORY_INFO));
+    assertEquals(directory, directory.toBuilder().build());
+  }
+
+  @Test
+  public void testBuilder() {
+    initializeExpectedBlob();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    Blob.Builder builder = new Blob.Builder(new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO)));
+    Blob blob =
+        builder
+            .setAcl(ACLS)
+            .setComponentCount(COMPONENT_COUNT)
+            .setContentType(CONTENT_TYPE)
+            .setCacheControl(CACHE_CONTROL)
+            .setContentDisposition(CONTENT_DISPOSITION)
+            .setContentEncoding(CONTENT_ENCODING)
+            .setContentLanguage(CONTENT_LANGUAGE)
+            .setCrc32c(CRC32)
+            .setCreateTime(CREATE_TIME)
+            .setCustomTime(CUSTOM_TIME)
+            .setStorageClass(STORAGE_CLASS)
+            .setTimeStorageClassUpdated(TIME_STORAGE_CLASS_UPDATED)
+            .setCustomerEncryption(CUSTOMER_ENCRYPTION)
+            .setKmsKeyName(KMS_KEY_NAME)
+            .setEventBasedHold(EVENT_BASED_HOLD)
+            .setTemporaryHold(TEMPORARY_HOLD)
+            .setRetentionExpirationTime(RETENTION_EXPIRATION_TIME)
+            .setDeleteTime(DELETE_TIME)
+            .setEtag(ETAG)
+            .setGeneratedId(GENERATED_ID)
+            .setMd5(MD5)
+            .setMediaLink(MEDIA_LINK)
+            .setMetadata(METADATA)
+            .setMetageneration(META_GENERATION)
+            .setOwner(OWNER)
+            .setSelfLink(SELF_LINK)
+            .setSize(SIZE)
+            .setUpdateTime(UPDATE_TIME)
+            .build();
+    assertEquals("b", blob.getBucket());
+    assertEquals("n", blob.getName());
+    assertEquals(ACLS, blob.getAcl());
+    assertEquals(COMPONENT_COUNT, blob.getComponentCount());
+    assertEquals(CONTENT_TYPE, blob.getContentType());
+    assertEquals(CACHE_CONTROL, blob.getCacheControl());
+    assertEquals(CONTENT_DISPOSITION, blob.getContentDisposition());
+    assertEquals(CONTENT_ENCODING, blob.getContentEncoding());
+    assertEquals(CONTENT_LANGUAGE, blob.getContentLanguage());
+    assertEquals(CRC32, blob.getCrc32c());
+    assertEquals(CRC32_HEX_STRING, blob.getCrc32cToHexString());
+    assertEquals(CREATE_TIME, blob.getCreateTime());
+    assertEquals(CUSTOM_TIME, blob.getCustomTime());
+    assertEquals(STORAGE_CLASS, blob.getStorageClass());
+    assertEquals(TIME_STORAGE_CLASS_UPDATED, blob.getTimeStorageClassUpdated());
+    assertEquals(CUSTOMER_ENCRYPTION, blob.getCustomerEncryption());
+    assertEquals(KMS_KEY_NAME, blob.getKmsKeyName());
+    assertEquals(EVENT_BASED_HOLD, blob.getEventBasedHold());
+    assertEquals(TEMPORARY_HOLD, blob.getTemporaryHold());
+    assertEquals(RETENTION_EXPIRATION_TIME, blob.getRetentionExpirationTime());
+    assertEquals(DELETE_TIME, blob.getDeleteTime());
+    assertEquals(ETAG, blob.getEtag());
+    assertEquals(GENERATED_ID, blob.getGeneratedId());
+    assertEquals(MD5, blob.getMd5());
+    assertEquals(MD5_HEX_STRING, blob.getMd5ToHexString());
+    assertEquals(MEDIA_LINK, blob.getMediaLink());
+    assertEquals(METADATA, blob.getMetadata());
+    assertEquals(META_GENERATION, blob.getMetageneration());
+    assertEquals(OWNER, blob.getOwner());
+    assertEquals(SELF_LINK, blob.getSelfLink());
+    assertEquals(SIZE, blob.getSize());
+    assertEquals(UPDATE_TIME, blob.getUpdateTime());
+    assertEquals(storage.getOptions(), blob.getStorage().getOptions());
+    assertFalse(blob.isDirectory());
+    builder = new Blob.Builder(new Blob(storage, new BlobInfo.BuilderImpl(DIRECTORY_INFO)));
+    blob = builder.setBlobId(BlobId.of("b", "n/")).setIsDirectory(true).setSize(0L).build();
+    assertEquals("b", blob.getBucket());
+    assertEquals("n/", blob.getName());
+    assertNull(blob.getAcl());
+    assertNull(blob.getComponentCount());
+    assertNull(blob.getContentType());
+    assertNull(blob.getCacheControl());
+    assertNull(blob.getContentDisposition());
+    assertNull(blob.getContentEncoding());
+    assertNull(blob.getContentLanguage());
+    assertNull(blob.getCrc32c());
+    assertNull(blob.getCrc32cToHexString());
+    assertNull(blob.getCreateTime());
+    assertNull(blob.getStorageClass());
+    assertNull(blob.getTimeStorageClassUpdated());
+    assertNull(blob.getCustomerEncryption());
+    assertNull(blob.getKmsKeyName());
+    assertNull(blob.getEventBasedHold());
+    assertNull(blob.getTemporaryHold());
+    assertNull(blob.getRetentionExpirationTime());
+    assertNull(blob.getDeleteTime());
+    assertNull(blob.getEtag());
+    assertNull(blob.getGeneratedId());
+    assertNull(blob.getMd5());
+    assertNull(blob.getMd5ToHexString());
+    assertNull(blob.getMediaLink());
+    assertNull(blob.getMetadata());
+    assertNull(blob.getMetageneration());
+    assertNull(blob.getOwner());
+    assertNull(blob.getSelfLink());
+    assertEquals(0L, (long) blob.getSize());
+    assertNull(blob.getUpdateTime());
+    assertNull(blob.getCustomTime());
+    assertTrue(blob.isDirectory());
+  }
+
+  @Test
+  public void testDownloadTo() throws Exception {
+    File file = File.createTempFile("blob", ".tmp");
+    Path path = file.toPath();
+
+    Storage s = Mockito.mock(Storage.class);
+    Blob blob = new Blob(s, new BuilderImpl(BlobInfo.newBuilder("buck", "obj").build()));
+
+    Mockito.doNothing().when(s).downloadTo(blob.getBlobId(), path);
+    blob.downloadTo(path);
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java
new file mode 100644
index 00000000..5600cd6e
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java
@@ -0,0 +1,916 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static com.google.cloud.storage.Acl.Role.WRITER;
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.api.gax.paging.Page;
+import com.google.cloud.PageImpl;
+import com.google.cloud.storage.Acl.Project;
+import com.google.cloud.storage.Acl.Project.ProjectRole;
+import com.google.cloud.storage.Acl.Role;
+import com.google.cloud.storage.Acl.User;
+import com.google.cloud.storage.BucketInfo.AgeDeleteRule;
+import com.google.cloud.storage.BucketInfo.DeleteRule;
+import com.google.cloud.storage.BucketInfo.LifecycleRule;
+import com.google.cloud.storage.BucketInfo.LifecycleRule.LifecycleAction;
+import com.google.cloud.storage.BucketInfo.LifecycleRule.LifecycleCondition;
+import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.common.base.Function;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import com.google.common.hash.Hashing;
+import com.google.common.io.BaseEncoding;
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.security.Key;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import javax.crypto.spec.SecretKeySpec;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class BucketTest {
+
+  private static final Acl ACL = Acl.of(User.ofAllAuthenticatedUsers(), Role.OWNER);
+  private static final Acl OTHER_ACL = Acl.of(new Project(ProjectRole.OWNERS, "p"), Role.READER);
+  private static final List<Acl> ACLS = ImmutableList.of(ACL, OTHER_ACL);
+  private static final String ETAG = "0xFF00";
+  private static final String GENERATED_ID = "B/N:1";
+  private static final Long META_GENERATION = 10L;
+  private static final User OWNER = new User("user@gmail.com");
+  private static final String SELF_LINK = "http://storage/b/n";
+  private static final Long CREATE_TIME = System.currentTimeMillis();
+  private static final Long UPDATE_TIME = CREATE_TIME - 1L;
+  private static final List<Cors> CORS = Collections.singletonList(Cors.newBuilder().build());
+  private static final List<Acl> DEFAULT_ACL =
+      Collections.singletonList(Acl.of(User.ofAllAuthenticatedUsers(), WRITER));
+
+  @SuppressWarnings({"unchecked", "deprecation"})
+  private static final List<? extends DeleteRule> DELETE_RULES =
+      Collections.singletonList(new AgeDeleteRule(5));
+
+  private static final List<? extends BucketInfo.LifecycleRule> LIFECYCLE_RULES =
+      Collections.singletonList(
+          new LifecycleRule(
+              LifecycleAction.newDeleteAction(),
+              LifecycleCondition.newBuilder().setAge(5).build()));
+  private static final String INDEX_PAGE = "index.html";
+  private static final String NOT_FOUND_PAGE = "error.html";
+  private static final String LOCATION = "ASIA";
+  private static final StorageClass STORAGE_CLASS = StorageClass.STANDARD;
+  private static final String DEFAULT_KMS_KEY_NAME =
+      "projects/p/locations/kr-loc/keyRings/kr/cryptoKeys/key";
+  private static final Boolean VERSIONING_ENABLED = true;
+  private static final Map<String, String> BUCKET_LABELS = ImmutableMap.of("label1", "value1");
+  private static final Boolean REQUESTER_PAYS = true;
+  private static final String USER_PROJECT = "test-project";
+  private static final Boolean DEFAULT_EVENT_BASED_HOLD = true;
+  private static final Long RETENTION_EFFECTIVE_TIME = 10L;
+  private static final Long RETENTION_PERIOD = 10L;
+  private static final Boolean RETENTION_POLICY_IS_LOCKED = false;
+  private static final List<String> LOCATION_TYPES =
+      ImmutableList.of("multi-region", "region", "dual-region");
+  private static final String LOCATION_TYPE = "multi-region";
+
+  @SuppressWarnings({"unchecked", "deprecation"})
+  private static final BucketInfo FULL_BUCKET_INFO =
+      BucketInfo.newBuilder("b")
+          .setAcl(ACLS)
+          .setEtag(ETAG)
+          .setGeneratedId(GENERATED_ID)
+          .setMetageneration(META_GENERATION)
+          .setOwner(OWNER)
+          .setSelfLink(SELF_LINK)
+          .setCors(CORS)
+          .setCreateTime(CREATE_TIME)
+          .setUpdateTime(UPDATE_TIME)
+          .setDefaultAcl(DEFAULT_ACL)
+          .setDeleteRules(DELETE_RULES)
+          .setLifecycleRules(LIFECYCLE_RULES)
+          .setIndexPage(INDEX_PAGE)
+          .setNotFoundPage(NOT_FOUND_PAGE)
+          .setLocation(LOCATION)
+          .setStorageClass(STORAGE_CLASS)
+          .setVersioningEnabled(VERSIONING_ENABLED)
+          .setLabels(BUCKET_LABELS)
+          .setRequesterPays(REQUESTER_PAYS)
+          .setDefaultKmsKeyName(DEFAULT_KMS_KEY_NAME)
+          .setDefaultEventBasedHold(DEFAULT_EVENT_BASED_HOLD)
+          .setRetentionEffectiveTime(RETENTION_EFFECTIVE_TIME)
+          .setRetentionPeriod(RETENTION_PERIOD)
+          .setRetentionPolicyIsLocked(RETENTION_POLICY_IS_LOCKED)
+          .build();
+
+  private static final BucketInfo BUCKET_INFO =
+      BucketInfo.newBuilder("b").setMetageneration(42L).build();
+  private static final String CONTENT_TYPE = "text/plain";
+  private static final String BASE64_KEY = "JVzfVl8NLD9FjedFuStegjRfES5ll5zc59CIXw572OA=";
+  private static final Key KEY =
+      new SecretKeySpec(BaseEncoding.base64().decode(BASE64_KEY), "AES256");
+  private final HttpRetryAlgorithmManager retryAlgorithmManager =
+      HttpStorageOptions.getDefaultInstance().getRetryAlgorithmManager();
+
+  private Storage storage;
+  private Storage serviceMockReturnsOptions = Mockito.mock(Storage.class);
+  private HttpStorageOptions mockOptions = Mockito.mock(HttpStorageOptions.class);
+  private Bucket bucket;
+  private Bucket expectedBucket;
+  private List<Blob> blobResults;
+
+  @Before
+  public void setUp() {
+    storage = Mockito.mock(Storage.class);
+  }
+
+  private void initializeExpectedBucket() {
+    when(serviceMockReturnsOptions.getOptions()).thenReturn(mockOptions);
+    when(mockOptions.getRetryAlgorithmManager()).thenReturn(retryAlgorithmManager);
+    expectedBucket = new Bucket(serviceMockReturnsOptions, new BucketInfo.BuilderImpl(BUCKET_INFO));
+    blobResults =
+        ImmutableList.of(
+            new Blob(
+                serviceMockReturnsOptions,
+                new BlobInfo.BuilderImpl(BlobInfo.newBuilder("b", "n1").build())),
+            new Blob(
+                serviceMockReturnsOptions,
+                new BlobInfo.BuilderImpl(BlobInfo.newBuilder("b", "n2").build())),
+            new Blob(
+                serviceMockReturnsOptions,
+                new BlobInfo.BuilderImpl(BlobInfo.newBuilder("b", "n3").build())));
+  }
+
+  private void initializeBucket() {
+    bucket = new Bucket(storage, new BucketInfo.BuilderImpl(BUCKET_INFO));
+  }
+
+  @Test
+  public void testExists_True() throws Exception {
+    initializeExpectedBucket();
+    Storage.BucketGetOption[] expectedOptions = {Storage.BucketGetOption.fields()};
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.get(BUCKET_INFO.getName(), expectedOptions)).thenReturn(expectedBucket);
+    initializeBucket();
+    assertTrue(bucket.exists());
+
+    verify(storage).getOptions();
+    verify(storage).get(BUCKET_INFO.getName(), expectedOptions);
+  }
+
+  @Test
+  public void testExists_False() throws Exception {
+    initializeExpectedBucket();
+    Storage.BucketGetOption[] expectedOptions = {Storage.BucketGetOption.fields()};
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.get(BUCKET_INFO.getName(), expectedOptions)).thenReturn(null);
+    initializeBucket();
+    assertFalse(bucket.exists());
+
+    verify(storage).getOptions();
+    verify(storage).get(BUCKET_INFO.getName(), expectedOptions);
+  }
+
+  @Test
+  public void testReload() throws Exception {
+    initializeExpectedBucket();
+    BucketInfo updatedInfo = BUCKET_INFO.toBuilder().setNotFoundPage("p").build();
+    Bucket expectedUpdatedBucket =
+        new Bucket(serviceMockReturnsOptions, new BucketInfo.BuilderImpl(updatedInfo));
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.get(updatedInfo.getName())).thenReturn(expectedUpdatedBucket);
+    initializeBucket();
+    Bucket updatedBucket = bucket.reload();
+    assertEquals(expectedUpdatedBucket, updatedBucket);
+
+    verify(storage).getOptions();
+    verify(storage).get(updatedInfo.getName());
+  }
+
+  @Test
+  public void testReloadNull() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.get(BUCKET_INFO.getName())).thenReturn(null);
+    initializeBucket();
+    assertNull(bucket.reload());
+
+    verify(storage).getOptions();
+    verify(storage).get(BUCKET_INFO.getName());
+  }
+
+  @Test
+  public void testReloadWithOptions() throws Exception {
+    initializeExpectedBucket();
+    BucketInfo updatedInfo = BUCKET_INFO.toBuilder().setNotFoundPage("p").build();
+    Bucket expectedUpdatedBucket =
+        new Bucket(serviceMockReturnsOptions, new BucketInfo.BuilderImpl(updatedInfo));
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.get(updatedInfo.getName(), Storage.BucketGetOption.metagenerationMatch(42L)))
+        .thenReturn(expectedUpdatedBucket);
+    initializeBucket();
+    Bucket updatedBucket = bucket.reload(Bucket.BucketSourceOption.metagenerationMatch());
+    assertEquals(expectedUpdatedBucket, updatedBucket);
+
+    verify(storage).getOptions();
+    verify(storage).get(updatedInfo.getName(), Storage.BucketGetOption.metagenerationMatch(42L));
+  }
+
+  @Test
+  public void testUpdate() throws Exception {
+    initializeExpectedBucket();
+    Bucket expectedUpdatedBucket = expectedBucket.toBuilder().setNotFoundPage("p").build();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.update(expectedUpdatedBucket)).thenReturn(expectedUpdatedBucket);
+    initializeBucket();
+    Bucket updatedBucket = new Bucket(storage, new BucketInfo.BuilderImpl(expectedUpdatedBucket));
+    Bucket actualUpdatedBucket = updatedBucket.update();
+    assertEquals(expectedUpdatedBucket, actualUpdatedBucket);
+
+    verify(storage, times(2)).getOptions();
+    verify(storage).update(expectedUpdatedBucket);
+  }
+
+  @Test
+  public void testDelete() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.delete(BUCKET_INFO.getName())).thenReturn(true);
+    initializeBucket();
+    assertTrue(bucket.delete());
+
+    verify(storage).getOptions();
+    verify(storage).delete(BUCKET_INFO.getName());
+  }
+
+  @Test
+  public void testList() throws Exception {
+    initializeExpectedBucket();
+    PageImpl<Blob> expectedBlobPage = new PageImpl<>(null, "c", blobResults);
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.list(BUCKET_INFO.getName())).thenReturn(expectedBlobPage);
+    initializeBucket();
+    Page<Blob> blobPage = bucket.list();
+    Iterator<Blob> blobInfoIterator = blobPage.getValues().iterator();
+    Iterator<Blob> blobIterator = blobPage.getValues().iterator();
+    while (blobInfoIterator.hasNext() && blobIterator.hasNext()) {
+      assertEquals(blobInfoIterator.next(), blobIterator.next());
+    }
+    assertFalse(blobInfoIterator.hasNext());
+    assertFalse(blobIterator.hasNext());
+    assertEquals(expectedBlobPage.getNextPageToken(), blobPage.getNextPageToken());
+
+    verify(storage).getOptions();
+    verify(storage).list(BUCKET_INFO.getName());
+  }
+
+  @Test
+  public void testGet() throws Exception {
+    initializeExpectedBucket();
+    Blob expectedBlob =
+        new Blob(
+            serviceMockReturnsOptions,
+            new BlobInfo.BuilderImpl(BlobInfo.newBuilder("b", "n").build()));
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.get(BlobId.of(expectedBucket.getName(), "n"), new Storage.BlobGetOption[0]))
+        .thenReturn(expectedBlob);
+    initializeBucket();
+    Blob blob = bucket.get("n");
+    assertEquals(expectedBlob, blob);
+
+    verify(storage).getOptions();
+    verify(storage).get(BlobId.of(expectedBucket.getName(), "n"), new Storage.BlobGetOption[0]);
+  }
+
+  @Test
+  public void testGetAllArray() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    List<BlobId> blobIds =
+        Lists.transform(
+            blobResults,
+            new Function<Blob, BlobId>() {
+              @Override
+              public BlobId apply(Blob blob) {
+                return blob.getBlobId();
+              }
+            });
+    when(storage.get(blobIds)).thenReturn(blobResults);
+    initializeBucket();
+    assertEquals(blobResults, bucket.get("n1", "n2", "n3"));
+
+    verify(storage).getOptions();
+    verify(storage).get(blobIds);
+  }
+
+  @Test
+  public void testGetAllIterable() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    List<BlobId> blobIds =
+        Lists.transform(
+            blobResults,
+            new Function<Blob, BlobId>() {
+              @Override
+              public BlobId apply(Blob blob) {
+                return blob.getBlobId();
+              }
+            });
+    when(storage.get(blobIds)).thenReturn(blobResults);
+    initializeBucket();
+    assertEquals(blobResults, bucket.get(ImmutableList.of("n1", "n2", "n3")));
+
+    verify(storage).getOptions();
+    verify(storage).get(blobIds);
+  }
+
+  @Test
+  public void testCreate() throws Exception {
+    initializeExpectedBucket();
+    BlobInfo info = BlobInfo.newBuilder("b", "n").setContentType(CONTENT_TYPE).build();
+    Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
+    byte[] content = {0xD, 0xE, 0xA, 0xD};
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.create(info, content)).thenReturn(expectedBlob);
+    initializeBucket();
+    Blob blob = bucket.create("n", content, CONTENT_TYPE);
+    assertEquals(expectedBlob, blob);
+
+    verify(storage).getOptions();
+    verify(storage).create(info, content);
+  }
+
+  @Test
+  public void testCreateNoContentType() throws Exception {
+    initializeExpectedBucket();
+    BlobInfo info = BlobInfo.newBuilder("b", "n").build();
+    Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
+    byte[] content = {0xD, 0xE, 0xA, 0xD};
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.create(info, content)).thenReturn(expectedBlob);
+    initializeBucket();
+    Blob blob = bucket.create("n", content);
+    assertEquals(expectedBlob, blob);
+
+    verify(storage).getOptions();
+    verify(storage).create(info, content);
+  }
+
+  @Test
+  public void testCreateWithOptions() throws Exception {
+    initializeExpectedBucket();
+    BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
+    Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
+    byte[] content = {0xD, 0xE, 0xA, 0xD};
+    Storage.PredefinedAcl acl = Storage.PredefinedAcl.ALL_AUTHENTICATED_USERS;
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.create(
+            info,
+            content,
+            new BlobTargetOption(UnifiedOpts.generationMatch(42L)),
+            new BlobTargetOption(UnifiedOpts.metagenerationMatch(24L)),
+            Storage.BlobTargetOption.predefinedAcl(acl),
+            Storage.BlobTargetOption.encryptionKey(BASE64_KEY),
+            Storage.BlobTargetOption.userProject(USER_PROJECT)))
+        .thenReturn(expectedBlob);
+    initializeBucket();
+    Blob blob =
+        bucket.create(
+            "n",
+            content,
+            CONTENT_TYPE,
+            Bucket.BlobTargetOption.generationMatch(42L),
+            Bucket.BlobTargetOption.metagenerationMatch(24L),
+            Bucket.BlobTargetOption.predefinedAcl(acl),
+            Bucket.BlobTargetOption.encryptionKey(BASE64_KEY),
+            Bucket.BlobTargetOption.userProject(USER_PROJECT));
+    assertEquals(expectedBlob, blob);
+
+    verify(storage).getOptions();
+    verify(storage)
+        .create(
+            info,
+            content,
+            new BlobTargetOption(UnifiedOpts.generationMatch(42L)),
+            new BlobTargetOption(UnifiedOpts.metagenerationMatch(24L)),
+            Storage.BlobTargetOption.predefinedAcl(acl),
+            Storage.BlobTargetOption.encryptionKey(BASE64_KEY),
+            Storage.BlobTargetOption.userProject(USER_PROJECT));
+  }
+
+  @Test
+  public void testCreateWithEncryptionKey() throws Exception {
+    initializeExpectedBucket();
+    BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
+    Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
+    byte[] content = {0xD, 0xE, 0xA, 0xD};
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.create(info, content, Storage.BlobTargetOption.encryptionKey(KEY)))
+        .thenReturn(expectedBlob);
+    initializeBucket();
+    Blob blob =
+        bucket.create("n", content, CONTENT_TYPE, Bucket.BlobTargetOption.encryptionKey(KEY));
+    assertEquals(expectedBlob, blob);
+
+    verify(storage).getOptions();
+    verify(storage).create(info, content, Storage.BlobTargetOption.encryptionKey(KEY));
+  }
+
+  @Test
+  public void testCreateWithKmsKeyName() throws Exception {
+    initializeExpectedBucket();
+    BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
+    Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
+    byte[] content = {0xD, 0xE, 0xA, 0xD};
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.create(info, content, Storage.BlobTargetOption.kmsKeyName(DEFAULT_KMS_KEY_NAME)))
+        .thenReturn(expectedBlob);
+    initializeBucket();
+    Blob blob =
+        bucket.create(
+            "n", content, CONTENT_TYPE, Bucket.BlobTargetOption.kmsKeyName(DEFAULT_KMS_KEY_NAME));
+    assertEquals(expectedBlob, blob);
+
+    verify(storage).getOptions();
+    verify(storage)
+        .create(info, content, Storage.BlobTargetOption.kmsKeyName(DEFAULT_KMS_KEY_NAME));
+  }
+
+  @Test
+  public void testCreateNotExists() throws Exception {
+    initializeExpectedBucket();
+    BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
+    Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
+    byte[] content = {0xD, 0xE, 0xA, 0xD};
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.create(info, content, new BlobTargetOption(UnifiedOpts.doesNotExist())))
+        .thenReturn(expectedBlob);
+    initializeBucket();
+    Blob blob = bucket.create("n", content, CONTENT_TYPE, Bucket.BlobTargetOption.doesNotExist());
+    assertEquals(expectedBlob, blob);
+
+    verify(storage).getOptions();
+    verify(storage).create(info, content, new BlobTargetOption(UnifiedOpts.doesNotExist()));
+  }
+
+  @Test
+  @SuppressWarnings({"unchecked", "deprecation"})
+  public void testCreateFromStream() throws Exception {
+    initializeExpectedBucket();
+    BlobInfo info = BlobInfo.newBuilder("b", "n").setContentType(CONTENT_TYPE).build();
+    Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
+    byte[] content = {0xD, 0xE, 0xA, 0xD};
+    InputStream streamContent = new ByteArrayInputStream(content);
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.create(info, streamContent)).thenReturn(expectedBlob);
+    initializeBucket();
+    Blob blob = bucket.create("n", streamContent, CONTENT_TYPE);
+    assertEquals(expectedBlob, blob);
+
+    verify(storage).getOptions();
+    verify(storage).create(info, streamContent);
+  }
+
+  @Test
+  @SuppressWarnings({"unchecked", "deprecation"})
+  public void testCreateFromStreamNoContentType() throws Exception {
+    initializeExpectedBucket();
+    BlobInfo info = BlobInfo.newBuilder("b", "n").build();
+    Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
+    byte[] content = {0xD, 0xE, 0xA, 0xD};
+    InputStream streamContent = new ByteArrayInputStream(content);
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.create(info, streamContent)).thenReturn(expectedBlob);
+    initializeBucket();
+    Blob blob = bucket.create("n", streamContent);
+    assertEquals(expectedBlob, blob);
+
+    verify(storage).getOptions();
+    verify(storage).create(info, streamContent);
+  }
+
+  @Test
+  @SuppressWarnings({"unchecked", "deprecation"})
+  public void testCreateFromStreamWithOptions() throws Exception {
+    initializeExpectedBucket();
+    BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
+    Blob expectedBlob = info.asBlob(serviceMockReturnsOptions);
+    byte[] content = {0xD, 0xE, 0xA, 0xD};
+    String crc32c = Utils.crc32cCodec.encode(Hashing.crc32c().hashBytes(content).asInt());
+    Storage.PredefinedAcl acl = Storage.PredefinedAcl.ALL_AUTHENTICATED_USERS;
+    InputStream streamContent = new ByteArrayInputStream(content);
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.create(
+            info,
+            streamContent,
+            new BlobWriteOption(UnifiedOpts.generationMatch(42L)),
+            new BlobWriteOption(UnifiedOpts.metagenerationMatch(24L)),
+            Storage.BlobWriteOption.predefinedAcl(acl),
+            new BlobWriteOption(UnifiedOpts.crc32cMatch(crc32c)),
+            new BlobWriteOption(UnifiedOpts.md5Match("md5")),
+            Storage.BlobWriteOption.encryptionKey(BASE64_KEY),
+            Storage.BlobWriteOption.userProject(USER_PROJECT)))
+        .thenReturn(expectedBlob);
+    initializeBucket();
+    Blob blob =
+        bucket.create(
+            "n",
+            streamContent,
+            CONTENT_TYPE,
+            Bucket.BlobWriteOption.generationMatch(42L),
+            Bucket.BlobWriteOption.metagenerationMatch(24L),
+            Bucket.BlobWriteOption.predefinedAcl(acl),
+            Bucket.BlobWriteOption.crc32cMatch(crc32c),
+            Bucket.BlobWriteOption.md5Match("md5"),
+            Bucket.BlobWriteOption.encryptionKey(BASE64_KEY),
+            Bucket.BlobWriteOption.userProject(USER_PROJECT));
+    assertEquals(expectedBlob, blob);
+
+    verify(storage).getOptions();
+    verify(storage)
+        .create(
+            info,
+            streamContent,
+            new BlobWriteOption(UnifiedOpts.generationMatch(42L)),
+            new BlobWriteOption(UnifiedOpts.metagenerationMatch(24L)),
+            Storage.BlobWriteOption.predefinedAcl(acl),
+            new BlobWriteOption(UnifiedOpts.crc32cMatch(crc32c)),
+            new BlobWriteOption(UnifiedOpts.md5Match("md5")),
+            Storage.BlobWriteOption.encryptionKey(BASE64_KEY),
+            Storage.BlobWriteOption.userProject(USER_PROJECT));
+  }
+
+  @Test
+  @SuppressWarnings({"unchecked", "deprecation"})
+  public void testCreateFromStreamWithEncryptionKey() throws Exception {
+    initializeExpectedBucket();
+    BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
+    Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
+    byte[] content = {0xD, 0xE, 0xA, 0xD};
+    InputStream streamContent = new ByteArrayInputStream(content);
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.create(info, streamContent, Storage.BlobWriteOption.encryptionKey(KEY)))
+        .thenReturn(expectedBlob);
+    initializeBucket();
+    Blob blob =
+        bucket.create("n", streamContent, CONTENT_TYPE, Bucket.BlobWriteOption.encryptionKey(KEY));
+    assertEquals(expectedBlob, blob);
+
+    verify(storage).getOptions();
+    verify(storage).create(info, streamContent, Storage.BlobWriteOption.encryptionKey(KEY));
+  }
+
+  @Test
+  @SuppressWarnings({"unchecked", "deprecation"})
+  public void testCreateFromStreamNotExists() throws Exception {
+    initializeExpectedBucket();
+    BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
+    Blob expectedBlob = info.asBlob(serviceMockReturnsOptions);
+    byte[] content = {0xD, 0xE, 0xA, 0xD};
+    InputStream streamContent = new ByteArrayInputStream(content);
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.create(info, streamContent, new BlobWriteOption(UnifiedOpts.doesNotExist())))
+        .thenReturn(expectedBlob);
+    initializeBucket();
+    Blob blob =
+        bucket.create("n", streamContent, CONTENT_TYPE, Bucket.BlobWriteOption.doesNotExist());
+    assertEquals(expectedBlob, blob);
+
+    verify(storage).getOptions();
+    verify(storage).create(info, streamContent, new BlobWriteOption(UnifiedOpts.doesNotExist()));
+  }
+
+  @Test
+  public void testGetAcl() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.getAcl(BUCKET_INFO.getName(), User.ofAllAuthenticatedUsers())).thenReturn(ACL);
+    initializeBucket();
+    assertEquals(ACL, bucket.getAcl(User.ofAllAuthenticatedUsers()));
+
+    verify(storage).getOptions();
+    verify(storage).getAcl(BUCKET_INFO.getName(), User.ofAllAuthenticatedUsers());
+  }
+
+  @Test
+  public void testDeleteAcl() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.deleteAcl(BUCKET_INFO.getName(), User.ofAllAuthenticatedUsers())).thenReturn(true);
+    initializeBucket();
+    assertTrue(bucket.deleteAcl(User.ofAllAuthenticatedUsers()));
+
+    verify(storage).getOptions();
+    verify(storage).deleteAcl(BUCKET_INFO.getName(), User.ofAllAuthenticatedUsers());
+  }
+
+  @Test
+  public void testCreateAcl() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
+    when(storage.createAcl(BUCKET_INFO.getName(), ACL)).thenReturn(returnedAcl);
+    initializeBucket();
+    assertEquals(returnedAcl, bucket.createAcl(ACL));
+
+    verify(storage).getOptions();
+    verify(storage).createAcl(BUCKET_INFO.getName(), ACL);
+  }
+
+  @Test
+  public void testUpdateAcl() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
+    when(storage.updateAcl(BUCKET_INFO.getName(), ACL)).thenReturn(returnedAcl);
+    initializeBucket();
+    assertEquals(returnedAcl, bucket.updateAcl(ACL));
+    verify(storage).getOptions();
+    verify(storage).updateAcl(BUCKET_INFO.getName(), ACL);
+  }
+
+  @Test
+  public void testListAcls() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.listAcls(BUCKET_INFO.getName())).thenReturn(ACLS);
+    initializeBucket();
+    assertEquals(ACLS, bucket.listAcls());
+
+    verify(storage).getOptions();
+    verify(storage).listAcls(BUCKET_INFO.getName());
+  }
+
+  @Test
+  public void testGetDefaultAcl() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.getDefaultAcl(BUCKET_INFO.getName(), User.ofAllAuthenticatedUsers()))
+        .thenReturn(ACL);
+    initializeBucket();
+    assertEquals(ACL, bucket.getDefaultAcl(User.ofAllAuthenticatedUsers()));
+
+    verify(storage).getOptions();
+    verify(storage).getDefaultAcl(BUCKET_INFO.getName(), User.ofAllAuthenticatedUsers());
+  }
+
+  @Test
+  public void testDeleteDefaultAcl() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.deleteDefaultAcl(BUCKET_INFO.getName(), User.ofAllAuthenticatedUsers()))
+        .thenReturn(true);
+    initializeBucket();
+    assertTrue(bucket.deleteDefaultAcl(User.ofAllAuthenticatedUsers()));
+
+    verify(storage).getOptions();
+    verify(storage).deleteDefaultAcl(BUCKET_INFO.getName(), User.ofAllAuthenticatedUsers());
+  }
+
+  @Test
+  public void testCreateDefaultAcl() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
+    when(storage.createDefaultAcl(BUCKET_INFO.getName(), ACL)).thenReturn(returnedAcl);
+    initializeBucket();
+    assertEquals(returnedAcl, bucket.createDefaultAcl(ACL));
+
+    verify(storage).getOptions();
+    verify(storage).createDefaultAcl(BUCKET_INFO.getName(), ACL);
+  }
+
+  @Test
+  public void testUpdateDefaultAcl() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
+    when(storage.updateDefaultAcl(BUCKET_INFO.getName(), ACL)).thenReturn(returnedAcl);
+    initializeBucket();
+    assertEquals(returnedAcl, bucket.updateDefaultAcl(ACL));
+
+    verify(storage).getOptions();
+    verify(storage).updateDefaultAcl(BUCKET_INFO.getName(), ACL);
+  }
+
+  @Test
+  public void testListDefaultAcls() throws Exception {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.listDefaultAcls(BUCKET_INFO.getName())).thenReturn(ACLS);
+    initializeBucket();
+    assertEquals(ACLS, bucket.listDefaultAcls());
+
+    verify(storage).getOptions();
+    verify(storage).listDefaultAcls(BUCKET_INFO.getName());
+  }
+
+  @Test
+  public void testLockRetention() throws Exception {
+    initializeExpectedBucket();
+    Bucket expectedRetentionLockedBucket =
+        expectedBucket
+            .toBuilder()
+            .setRetentionPeriod(RETENTION_PERIOD)
+            .setRetentionPolicyIsLocked(true)
+            .build();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.lockRetentionPolicy(
+            expectedRetentionLockedBucket,
+            Storage.BucketTargetOption.metagenerationMatch(),
+            Storage.BucketTargetOption.userProject(USER_PROJECT)))
+        .thenReturn(expectedRetentionLockedBucket);
+    initializeBucket();
+    Bucket lockedRetentionPolicyBucket =
+        new Bucket(storage, new BucketInfo.BuilderImpl(expectedRetentionLockedBucket));
+    Bucket actualRetentionLockedBucket =
+        lockedRetentionPolicyBucket.lockRetentionPolicy(
+            Storage.BucketTargetOption.metagenerationMatch(),
+            Storage.BucketTargetOption.userProject(USER_PROJECT));
+    assertEquals(expectedRetentionLockedBucket, actualRetentionLockedBucket);
+
+    verify(storage, times(2)).getOptions();
+    verify(storage)
+        .lockRetentionPolicy(
+            expectedRetentionLockedBucket,
+            Storage.BucketTargetOption.metagenerationMatch(),
+            Storage.BucketTargetOption.userProject(USER_PROJECT));
+  }
+
+  @Test
+  @SuppressWarnings({"unchecked", "deprecation"})
+  public void testToBuilder() {
+    when(storage.getOptions()).thenReturn(mockOptions);
+    Bucket fullBucket = new Bucket(storage, new BucketInfo.BuilderImpl(FULL_BUCKET_INFO));
+    assertEquals(fullBucket, fullBucket.toBuilder().build());
+    Bucket simpleBlob = new Bucket(storage, new BucketInfo.BuilderImpl(BUCKET_INFO));
+    assertEquals(simpleBlob, simpleBlob.toBuilder().build());
+  }
+
+  @Test
+  @SuppressWarnings({"unchecked", "deprecation"})
+  public void testBuilder() {
+    initializeExpectedBucket();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    Bucket.Builder builder =
+        new Bucket.Builder(new Bucket(storage, new BucketInfo.BuilderImpl(BUCKET_INFO)));
+    Bucket bucket =
+        builder
+            .setAcl(ACLS)
+            .setEtag(ETAG)
+            .setGeneratedId(GENERATED_ID)
+            .setMetageneration(META_GENERATION)
+            .setOwner(OWNER)
+            .setSelfLink(SELF_LINK)
+            .setCors(CORS)
+            .setCreateTime(CREATE_TIME)
+            .setUpdateTime(UPDATE_TIME)
+            .setDefaultAcl(DEFAULT_ACL)
+            .setDeleteRules(DELETE_RULES)
+            .setLifecycleRules(LIFECYCLE_RULES)
+            .setIndexPage(INDEX_PAGE)
+            .setNotFoundPage(NOT_FOUND_PAGE)
+            .setLocation(LOCATION)
+            .setLocationType(LOCATION_TYPE)
+            .setStorageClass(STORAGE_CLASS)
+            .setVersioningEnabled(VERSIONING_ENABLED)
+            .setLabels(BUCKET_LABELS)
+            .setRequesterPays(REQUESTER_PAYS)
+            .setDefaultKmsKeyName(DEFAULT_KMS_KEY_NAME)
+            .setDefaultEventBasedHold(DEFAULT_EVENT_BASED_HOLD)
+            .setRetentionEffectiveTime(RETENTION_EFFECTIVE_TIME)
+            .setRetentionPeriod(RETENTION_PERIOD)
+            .setRetentionPolicyIsLocked(RETENTION_POLICY_IS_LOCKED)
+            .build();
+    assertEquals("b", bucket.getName());
+    assertEquals(ACLS, bucket.getAcl());
+    assertEquals(ETAG, bucket.getEtag());
+    assertEquals(GENERATED_ID, bucket.getGeneratedId());
+    assertEquals(META_GENERATION, bucket.getMetageneration());
+    assertEquals(OWNER, bucket.getOwner());
+    assertEquals(SELF_LINK, bucket.getSelfLink());
+    assertEquals(CREATE_TIME, bucket.getCreateTime());
+    assertEquals(UPDATE_TIME, bucket.getUpdateTime());
+    assertEquals(CORS, bucket.getCors());
+    assertEquals(DEFAULT_ACL, bucket.getDefaultAcl());
+    assertEquals(DELETE_RULES, bucket.getDeleteRules());
+    assertEquals(LIFECYCLE_RULES, bucket.getLifecycleRules());
+    assertEquals(INDEX_PAGE, bucket.getIndexPage());
+    assertEquals(NOT_FOUND_PAGE, bucket.getNotFoundPage());
+    assertEquals(LOCATION, bucket.getLocation());
+    assertEquals(STORAGE_CLASS, bucket.getStorageClass());
+    assertEquals(VERSIONING_ENABLED, bucket.versioningEnabled());
+    assertEquals(BUCKET_LABELS, bucket.getLabels());
+    assertEquals(REQUESTER_PAYS, bucket.requesterPays());
+    assertEquals(DEFAULT_KMS_KEY_NAME, bucket.getDefaultKmsKeyName());
+    assertEquals(DEFAULT_EVENT_BASED_HOLD, bucket.getDefaultEventBasedHold());
+    assertEquals(RETENTION_EFFECTIVE_TIME, bucket.getRetentionEffectiveTime());
+    assertEquals(RETENTION_PERIOD, bucket.getRetentionPeriod());
+    assertEquals(RETENTION_POLICY_IS_LOCKED, bucket.retentionPolicyIsLocked());
+    assertEquals(storage.getOptions(), bucket.getStorage().getOptions());
+    assertTrue(LOCATION_TYPES.contains(LOCATION_TYPE));
+  }
+
+  @Test
+  public void testDeleteLifecycleRules() {
+    initializeExpectedBucket();
+    Bucket bucket =
+        new Bucket(serviceMockReturnsOptions, new BucketInfo.BuilderImpl(FULL_BUCKET_INFO));
+    assertThat(bucket.getLifecycleRules()).hasSize(1);
+    Bucket expectedUpdatedBucket = bucket.toBuilder().deleteLifecycleRules().build();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.update(expectedUpdatedBucket)).thenReturn(expectedUpdatedBucket);
+    initializeBucket();
+    Bucket updatedBucket = new Bucket(storage, new BucketInfo.BuilderImpl(expectedUpdatedBucket));
+    Bucket actualUpdatedBucket = updatedBucket.update();
+    assertThat(actualUpdatedBucket.getLifecycleRules()).hasSize(0);
+
+    verify(storage, times(2)).getOptions();
+    verify(storage).update(expectedUpdatedBucket);
+  }
+
+  @Test
+  public void testUpdateBucketLogging() {
+    initializeExpectedBucket();
+    BucketInfo.Logging logging =
+        BucketInfo.Logging.newBuilder()
+            .setLogBucket("logs-bucket")
+            .setLogObjectPrefix("test-logs")
+            .build();
+    BucketInfo bucketInfo = BucketInfo.newBuilder("b").setLogging(logging).build();
+    Bucket bucket = new Bucket(serviceMockReturnsOptions, new BucketInfo.BuilderImpl(bucketInfo));
+    assertThat(bucket.getLogging().getLogBucket()).isEqualTo("logs-bucket");
+    assertThat(bucket.getLogging().getLogObjectPrefix()).isEqualTo("test-logs");
+    Bucket expectedUpdatedBucket = bucket.toBuilder().setLogging(null).build();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.update(expectedUpdatedBucket)).thenReturn(expectedUpdatedBucket);
+    initializeBucket();
+    Bucket updatedBucket = new Bucket(storage, new BucketInfo.BuilderImpl(expectedUpdatedBucket));
+    Bucket actualUpdatedBucket = updatedBucket.update();
+    assertThat(actualUpdatedBucket.getLogging().getLogBucket()).isNull();
+    assertThat(actualUpdatedBucket.getLogging().getLogObjectPrefix()).isNull();
+
+    verify(storage, times(2)).getOptions();
+    verify(storage).update(expectedUpdatedBucket);
+  }
+
+  @Test
+  public void testRemoveBucketCORS() {
+    initializeExpectedBucket();
+    List<Cors.Origin> origins = ImmutableList.of(Cors.Origin.of("http://cloud.google.com"));
+    List<HttpMethod> httpMethods = ImmutableList.of(HttpMethod.GET);
+    List<String> responseHeaders = ImmutableList.of("Content-Type");
+    Cors cors =
+        Cors.newBuilder()
+            .setOrigins(origins)
+            .setMethods(httpMethods)
+            .setResponseHeaders(responseHeaders)
+            .setMaxAgeSeconds(100)
+            .build();
+    BucketInfo bucketInfo = BucketInfo.newBuilder("b").setCors(ImmutableList.of(cors)).build();
+    Bucket bucket = new Bucket(serviceMockReturnsOptions, new BucketInfo.BuilderImpl(bucketInfo));
+    assertThat(bucket.getCors()).isNotNull();
+    assertThat(bucket.getCors().get(0).getMaxAgeSeconds()).isEqualTo(100);
+    assertThat(bucket.getCors().get(0).getMethods()).isEqualTo(httpMethods);
+    assertThat(bucket.getCors().get(0).getOrigins()).isEqualTo(origins);
+    assertThat(bucket.getCors().get(0).getResponseHeaders()).isEqualTo(responseHeaders);
+
+    // Remove bucket CORS configuration.
+    Bucket expectedUpdatedBucket = bucket.toBuilder().setCors(null).build();
+    when(storage.getOptions()).thenReturn(mockOptions);
+    when(storage.update(expectedUpdatedBucket)).thenReturn(expectedUpdatedBucket);
+    initializeBucket();
+    Bucket updatedBucket = new Bucket(storage, new BucketInfo.BuilderImpl(expectedUpdatedBucket));
+    Bucket actualUpdatedBucket = updatedBucket.update();
+    assertThat(actualUpdatedBucket.getCors()).isEmpty();
+
+    verify(storage, times(2)).getOptions();
+    verify(storage).update(expectedUpdatedBucket);
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BuffersTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BuffersTest.java
index f0d760cb..3721dbc4 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BuffersTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BuffersTest.java
@@ -16,10 +16,15 @@
 
 package com.google.cloud.storage;
 
+import static com.google.cloud.storage.TestUtils.assertAll;
+import static com.google.cloud.storage.TestUtils.xxd;
 import static com.google.common.truth.Truth.assertThat;
 
+import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
 import java.security.SecureRandom;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Test;
 
 public final class BuffersTest {
@@ -72,4 +77,87 @@ public final class BuffersTest {
     ByteBuffer b1 = Buffers.allocateAligned(8, 4);
     assertThat(b1.capacity()).isEqualTo(8);
   }
+
+  @Test
+  public void fillFrom_handles_0SizeRead_someBytesRead() throws Exception {
+    byte[] bytes = new byte[14];
+    ByteBuffer buf = ByteBuffer.wrap(bytes);
+
+    byte[] expected =
+        new byte[] {
+          (byte) 'A',
+          (byte) 'B',
+          (byte) 'C',
+          (byte) 'A',
+          (byte) 'B',
+          (byte) 'A',
+          (byte) 'A',
+          (byte) 'A',
+          (byte) 'B',
+          (byte) 'A',
+          (byte) 'B',
+          (byte) 'C',
+          (byte) 0,
+          (byte) 0
+        };
+
+    int[] acceptSequence = new int[] {3, 2, 1, 0, 0, 1, 2, 3};
+    AtomicInteger readCount = new AtomicInteger(0);
+
+    ReadableByteChannel c =
+        new ReadableByteChannel() {
+          @Override
+          public int read(ByteBuffer dst) throws IOException {
+            int i = readCount.getAndIncrement();
+            if (i == acceptSequence.length) {
+              return -1;
+            }
+            int bytesToRead = acceptSequence[i];
+            if (bytesToRead > 0) {
+              long copy =
+                  Buffers.copy(DataGenerator.base64Characters().genByteBuffer(bytesToRead), dst);
+              assertThat(copy).isEqualTo(bytesToRead);
+            }
+
+            return bytesToRead;
+          }
+
+          @Override
+          public boolean isOpen() {
+            return true;
+          }
+
+          @Override
+          public void close() throws IOException {}
+        };
+    int filled = Buffers.fillFrom(buf, c);
+
+    assertAll(
+        () -> assertThat(filled).isEqualTo(12),
+        () -> assertThat(xxd(bytes)).isEqualTo(xxd(expected)));
+  }
+
+  @Test
+  public void fillFrom_handles_0SizeRead_noBytesRead() throws Exception {
+    ByteBuffer buf = ByteBuffer.allocate(3);
+
+    ReadableByteChannel c =
+        new ReadableByteChannel() {
+          @Override
+          public int read(ByteBuffer dst) throws IOException {
+            return -1;
+          }
+
+          @Override
+          public boolean isOpen() {
+            return true;
+          }
+
+          @Override
+          public void close() throws IOException {}
+        };
+    int filled = Buffers.fillFrom(buf, c);
+
+    assertThat(filled).isEqualTo(-1);
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ChunkSegmenterTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ChunkSegmenterTest.java
index f8fd2283..3a34c2f5 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ChunkSegmenterTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ChunkSegmenterTest.java
@@ -16,18 +16,24 @@
 
 package com.google.cloud.storage;
 
+import static com.google.cloud.storage.TestUtils.assertAll;
 import static com.google.common.truth.Truth.assertThat;
 
 import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.common.collect.ImmutableList;
 import com.google.common.hash.HashCode;
 import com.google.common.hash.Hashing;
+import com.google.protobuf.ByteString;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
+import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.stream.Collectors;
 import net.jqwik.api.Arbitraries;
 import net.jqwik.api.Arbitrary;
+import net.jqwik.api.Example;
 import net.jqwik.api.ForAll;
 import net.jqwik.api.Property;
 import net.jqwik.api.Provide;
@@ -58,6 +64,113 @@ final class ChunkSegmenterTest {
     assertThat(reduce).isAnyOf(Optional.empty(), Optional.of(Crc32cValue.of(td.allCrc32c.asInt())));
   }
 
+  /**
+   *
+   *
+   * <pre>
+   * Given 64 bytes, maxSegmentSize: 10, blockSize: 5
+   * 0                                                              64
+   * |---------------------------------------------------------------|
+   *   Produce 6 10-byte segments
+   * |---------|---------|---------|---------|---------|---------|
+   * </pre>
+   */
+  @Example
+  void allowUnalignedBlocks_false_1() {
+    ChunkSegmenter segmenter =
+        new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.noCopy(), 10, 5);
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(64);
+    List<ByteString> expected =
+        ImmutableList.of(
+            ByteString.copyFrom(bytes, 0, 10),
+            ByteString.copyFrom(bytes, 10, 10),
+            ByteString.copyFrom(bytes, 20, 10),
+            ByteString.copyFrom(bytes, 30, 10),
+            ByteString.copyFrom(bytes, 40, 10),
+            ByteString.copyFrom(bytes, 50, 10));
+
+    ByteBuffer buf = ByteBuffer.wrap(bytes);
+
+    ChunkSegment[] segments = segmenter.segmentBuffers(new ByteBuffer[] {buf}, 0, 1, false);
+    assertThat(buf.remaining()).isEqualTo(4);
+    List<ByteString> actual =
+        Arrays.stream(segments).map(ChunkSegment::getB).collect(Collectors.toList());
+    assertThat(actual).isEqualTo(expected);
+  }
+
+  /**
+   *
+   *
+   * <pre>
+   * Given 64 bytes, maxSegmentSize: 14, blockSize: 7
+   * 0                                                              64
+   * |---------------------------------------------------------------|
+   *   Produce 4 14-byte segments, and one 7 byte segment
+   * |-------------|-------------|-------------|-------------|------|
+   * </pre>
+   */
+  @Example
+  void allowUnalignedBlocks_false_2() throws Exception {
+    ChunkSegmenter segmenter =
+        new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.noCopy(), 14, 7);
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(64);
+    List<ByteString> expected =
+        ImmutableList.of(
+            ByteString.copyFrom(bytes, 0, 14),
+            ByteString.copyFrom(bytes, 14, 14),
+            ByteString.copyFrom(bytes, 28, 14),
+            ByteString.copyFrom(bytes, 42, 14),
+            ByteString.copyFrom(bytes, 56, 7));
+
+    ByteBuffer buf = ByteBuffer.wrap(bytes);
+
+    ChunkSegment[] segments = segmenter.segmentBuffers(new ByteBuffer[] {buf}, 0, 1, false);
+    List<ByteString> actual =
+        Arrays.stream(segments).map(ChunkSegment::getB).collect(Collectors.toList());
+    assertAll(
+        () -> assertThat(buf.remaining()).isEqualTo(1),
+        () -> assertThat(actual).isEqualTo(expected));
+  }
+
+  /**
+   *
+   *
+   * <pre>
+   * Given 60 bytes in one buffer and 4 bytes in a second buffer, maxSegmentSize: 14, blockSize: 7
+   * 0                                                          60   4
+   * |-----------------------------------------------------------|---|
+   *   Produce 4 14-byte segments, and one 7 byte segment
+   * |-------------|-------------|-------------|-------------|------|
+   * </pre>
+   */
+  @Example
+  void allowUnalignedBlocks_false_3() throws Exception {
+    ChunkSegmenter segmenter =
+        new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.noCopy(), 14, 7);
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(64);
+    List<ByteString> expected =
+        ImmutableList.of(
+            ByteString.copyFrom(bytes, 0, 14),
+            ByteString.copyFrom(bytes, 14, 14),
+            ByteString.copyFrom(bytes, 28, 14),
+            ByteString.copyFrom(bytes, 42, 14),
+            ByteString.copyFrom(bytes, 56, 7));
+
+    ByteBuffer buf1 = ByteBuffer.wrap(bytes, 0, 60);
+    ByteBuffer buf2 = ByteBuffer.wrap(bytes, 60, 4);
+
+    ChunkSegment[] segments = segmenter.segmentBuffers(new ByteBuffer[] {buf1, buf2}, 0, 2, false);
+    List<ByteString> actual =
+        Arrays.stream(segments).map(ChunkSegment::getB).collect(Collectors.toList());
+    assertAll(
+        () -> assertThat(buf1.remaining()).isEqualTo(0),
+        () -> assertThat(buf2.remaining()).isEqualTo(1),
+        () -> assertThat(actual).isEqualTo(expected));
+  }
+
   @Provide("TestData")
   static Arbitrary<TestData> arbitraryTestData() {
     return Arbitraries.lazyOf(
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java
new file mode 100644
index 00000000..f245b64b
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java
@@ -0,0 +1,212 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.cloud.RestorableState;
+import com.google.cloud.ServiceOptions;
+import com.google.cloud.storage.spi.StorageRpcFactory;
+import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.cloud.storage.spi.v1.StorageRpc.RewriteRequest;
+import com.google.cloud.storage.spi.v1.StorageRpc.RewriteResponse;
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class CopyWriterTest {
+
+  private static final String SOURCE_BUCKET_NAME = "b";
+  private static final String SOURCE_BLOB_NAME = "n";
+  private static final String DESTINATION_BUCKET_NAME = "b1";
+  private static final String DESTINATION_BLOB_NAME = "n1";
+  private static final BlobId BLOB_ID = BlobId.of(SOURCE_BUCKET_NAME, SOURCE_BLOB_NAME);
+  private static final BlobInfo BLOB_INFO =
+      BlobInfo.newBuilder(DESTINATION_BUCKET_NAME, DESTINATION_BLOB_NAME).build();
+  private static final BlobInfo RESULT_INFO =
+      BlobInfo.newBuilder(DESTINATION_BUCKET_NAME, DESTINATION_BLOB_NAME)
+          .setContentType("type")
+          .build();
+  private static final Map<StorageRpc.Option, ?> EMPTY_OPTIONS = ImmutableMap.of();
+  private static final RewriteRequest REQUEST_WITH_OBJECT =
+      new StorageRpc.RewriteRequest(
+          Conversions.json().blobId().encode(BLOB_ID),
+          EMPTY_OPTIONS,
+          true,
+          Conversions.json().blobInfo().encode(BLOB_INFO),
+          EMPTY_OPTIONS,
+          null);
+  private static final RewriteRequest REQUEST_WITHOUT_OBJECT =
+      new StorageRpc.RewriteRequest(
+          Conversions.json().blobId().encode(BLOB_ID),
+          EMPTY_OPTIONS,
+          false,
+          Conversions.json().blobInfo().encode(BLOB_INFO),
+          EMPTY_OPTIONS,
+          null);
+  private static final RewriteResponse RESPONSE_WITH_OBJECT =
+      new RewriteResponse(REQUEST_WITH_OBJECT, null, 42L, false, "token", 21L);
+  private static final RewriteResponse RESPONSE_WITHOUT_OBJECT =
+      new RewriteResponse(REQUEST_WITHOUT_OBJECT, null, 42L, false, "token", 21L);
+  private static final RewriteResponse RESPONSE_WITH_OBJECT_DONE =
+      new RewriteResponse(
+          REQUEST_WITH_OBJECT,
+          Conversions.json().blobInfo().encode(RESULT_INFO),
+          42L,
+          true,
+          "token",
+          42L);
+  private static final RewriteResponse RESPONSE_WITHOUT_OBJECT_DONE =
+      new RewriteResponse(
+          REQUEST_WITHOUT_OBJECT,
+          Conversions.json().blobInfo().encode(RESULT_INFO),
+          42L,
+          true,
+          "token",
+          42L);
+
+  private HttpStorageOptions options;
+  private StorageRpcFactory rpcFactoryMock;
+  private StorageRpc storageRpcMock;
+  private CopyWriter copyWriter;
+  private Blob result;
+
+  @Before
+  public void setUp() {
+    rpcFactoryMock = Mockito.mock(StorageRpcFactory.class);
+    storageRpcMock = Mockito.mock(StorageRpc.class);
+    when(rpcFactoryMock.create(any(StorageOptions.class))).thenReturn(storageRpcMock);
+    options =
+        HttpStorageOptions.newBuilder()
+            .setProjectId("projectid")
+            .setServiceRpcFactory(rpcFactoryMock)
+            .setRetrySettings(ServiceOptions.getNoRetrySettings())
+            .build();
+    result = new Blob(options.getService(), new BlobInfo.BuilderImpl(RESULT_INFO));
+    verify(rpcFactoryMock).create(any(StorageOptions.class));
+  }
+
+  @Test
+  public void testRewriteWithObject() {
+    when(storageRpcMock.continueRewrite(RESPONSE_WITH_OBJECT))
+        .thenReturn(RESPONSE_WITH_OBJECT_DONE);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITH_OBJECT);
+    assertEquals(result, copyWriter.getResult());
+    assertTrue(copyWriter.isDone());
+    assertEquals(42L, copyWriter.getTotalBytesCopied());
+    assertEquals(42L, copyWriter.getBlobSize());
+    verify(storageRpcMock).continueRewrite(RESPONSE_WITH_OBJECT);
+  }
+
+  @Test
+  public void testRewriteWithoutObject() {
+    when(storageRpcMock.continueRewrite(RESPONSE_WITHOUT_OBJECT))
+        .thenReturn(RESPONSE_WITHOUT_OBJECT_DONE);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITHOUT_OBJECT);
+    assertEquals(result, copyWriter.getResult());
+    assertTrue(copyWriter.isDone());
+    assertEquals(42L, copyWriter.getTotalBytesCopied());
+    assertEquals(42L, copyWriter.getBlobSize());
+    verify(storageRpcMock).continueRewrite(RESPONSE_WITHOUT_OBJECT);
+  }
+
+  @Test
+  public void testRewriteWithObjectMultipleRequests() {
+    when(storageRpcMock.continueRewrite(RESPONSE_WITH_OBJECT))
+        .thenReturn(RESPONSE_WITH_OBJECT, RESPONSE_WITHOUT_OBJECT_DONE);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITH_OBJECT);
+    assertEquals(result, copyWriter.getResult());
+    assertTrue(copyWriter.isDone());
+    assertEquals(42L, copyWriter.getTotalBytesCopied());
+    assertEquals(42L, copyWriter.getBlobSize());
+    verify(storageRpcMock, times(2)).continueRewrite(RESPONSE_WITH_OBJECT);
+  }
+
+  @Test
+  public void testRewriteWithoutObjectMultipleRequests() {
+    when(storageRpcMock.continueRewrite(RESPONSE_WITHOUT_OBJECT))
+        .thenReturn(RESPONSE_WITHOUT_OBJECT, RESPONSE_WITHOUT_OBJECT_DONE);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITHOUT_OBJECT);
+    assertEquals(result, copyWriter.getResult());
+    assertTrue(copyWriter.isDone());
+    assertEquals(42L, copyWriter.getTotalBytesCopied());
+    assertEquals(42L, copyWriter.getBlobSize());
+    verify(storageRpcMock, times(2)).continueRewrite(RESPONSE_WITHOUT_OBJECT);
+  }
+
+  @Test
+  public void testSaveAndRestoreWithObject() {
+    when(storageRpcMock.continueRewrite(RESPONSE_WITH_OBJECT))
+        .thenReturn(RESPONSE_WITH_OBJECT, RESPONSE_WITH_OBJECT_DONE);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITH_OBJECT);
+    copyWriter.copyChunk();
+    assertTrue(!copyWriter.isDone());
+    assertEquals(21L, copyWriter.getTotalBytesCopied());
+    assertEquals(42L, copyWriter.getBlobSize());
+    RestorableState<CopyWriter> rewriterState = copyWriter.capture();
+    CopyWriter restoredRewriter = rewriterState.restore();
+    assertEquals(result, restoredRewriter.getResult());
+    assertTrue(restoredRewriter.isDone());
+    assertEquals(42L, restoredRewriter.getTotalBytesCopied());
+    assertEquals(42L, restoredRewriter.getBlobSize());
+    verify(storageRpcMock, times(2)).continueRewrite(RESPONSE_WITH_OBJECT);
+  }
+
+  @Test
+  public void testSaveAndRestoreWithoutObject() {
+    when(storageRpcMock.continueRewrite(RESPONSE_WITHOUT_OBJECT))
+        .thenReturn(RESPONSE_WITHOUT_OBJECT, RESPONSE_WITHOUT_OBJECT_DONE);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITHOUT_OBJECT);
+    copyWriter.copyChunk();
+    assertTrue(!copyWriter.isDone());
+    assertEquals(21L, copyWriter.getTotalBytesCopied());
+    assertEquals(42L, copyWriter.getBlobSize());
+    RestorableState<CopyWriter> rewriterState = copyWriter.capture();
+    CopyWriter restoredRewriter = rewriterState.restore();
+    assertEquals(result, restoredRewriter.getResult());
+    assertTrue(restoredRewriter.isDone());
+    assertEquals(42L, restoredRewriter.getTotalBytesCopied());
+    assertEquals(42L, restoredRewriter.getBlobSize());
+    verify(storageRpcMock, times(2)).continueRewrite(RESPONSE_WITHOUT_OBJECT);
+  }
+
+  @Test
+  public void testSaveAndRestoreWithResult() {
+    when(storageRpcMock.continueRewrite(RESPONSE_WITH_OBJECT))
+        .thenReturn(RESPONSE_WITH_OBJECT_DONE);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITH_OBJECT);
+    copyWriter.copyChunk();
+    assertEquals(result, copyWriter.getResult());
+    assertTrue(copyWriter.isDone());
+    assertEquals(42L, copyWriter.getTotalBytesCopied());
+    assertEquals(42L, copyWriter.getBlobSize());
+    RestorableState<CopyWriter> rewriterState = copyWriter.capture();
+    CopyWriter restoredRewriter = rewriterState.restore();
+    assertEquals(result, restoredRewriter.getResult());
+    assertTrue(restoredRewriter.isDone());
+    assertEquals(42L, restoredRewriter.getTotalBytesCopied());
+    assertEquals(42L, restoredRewriter.getBlobSize());
+    verify(storageRpcMock).continueRewrite(RESPONSE_WITH_OBJECT);
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITUnbufferedResumableUploadTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITUnbufferedResumableUploadTest.java
index cf1b0d67..34b600d7 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITUnbufferedResumableUploadTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITUnbufferedResumableUploadTest.java
@@ -18,7 +18,9 @@ package com.google.cloud.storage;
 
 import static com.google.common.truth.Truth.assertThat;
 
+import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
+import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.services.storage.model.StorageObject;
 import com.google.cloud.storage.ITUnbufferedResumableUploadTest.ObjectSizes;
 import com.google.cloud.storage.TransportCompatibility.Transport;
@@ -36,6 +38,11 @@ import com.google.cloud.storage.it.runner.annotations.Parameterized.ParametersPr
 import com.google.cloud.storage.it.runner.registry.Generator;
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.collect.ImmutableList;
+import com.google.storage.v2.Object;
+import com.google.storage.v2.StorageClient;
+import com.google.storage.v2.WriteObjectRequest;
+import com.google.storage.v2.WriteObjectResponse;
+import com.google.storage.v2.WriteObjectSpec;
 import java.io.IOException;
 import java.math.BigInteger;
 import java.nio.ByteBuffer;
@@ -103,8 +110,13 @@ public final class ITUnbufferedResumableUploadTest {
     ByteBuffer b = DataGenerator.base64Characters().genByteBuffer(size);
 
     UnbufferedWritableByteChannel open = session.open();
-    int written = open.write(b);
-    assertThat(written).isEqualTo(objectSize);
+    int written1 = open.write(b);
+    assertThat(written1).isEqualTo(objectSize);
+    assertThat(b.remaining()).isEqualTo(additional);
+
+    // no bytes should be consumed if less than 256KiB
+    int written2 = open.write(b);
+    assertThat(written2).isEqualTo(0);
     assertThat(b.remaining()).isEqualTo(additional);
 
     int writtenAndClose = open.writeAndClose(b);
@@ -114,4 +126,70 @@ public final class ITUnbufferedResumableUploadTest {
     StorageObject storageObject = session.getResult().get(2, TimeUnit.SECONDS);
     assertThat(storageObject.getSize()).isEqualTo(BigInteger.valueOf(size));
   }
+
+  @Test
+  @Exclude(transports = Transport.HTTP)
+  public void grpc() throws Exception {
+    BlobInfo blobInfo = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    Opts<ObjectTargetOpt> opts = Opts.empty();
+    BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null);
+    BlobInfo updated = opts.blobInfoMapper().apply(builder).build();
+
+    Object object = Conversions.grpc().blobInfo().encode(updated);
+    Object.Builder objectBuilder =
+        object
+            .toBuilder()
+            // required if the data is changing
+            .clearChecksums()
+            // trimmed to shave payload size
+            .clearGeneration()
+            .clearMetageneration()
+            .clearSize()
+            .clearCreateTime()
+            .clearUpdateTime();
+    WriteObjectSpec.Builder specBuilder = WriteObjectSpec.newBuilder().setResource(objectBuilder);
+
+    WriteObjectRequest.Builder requestBuilder =
+        WriteObjectRequest.newBuilder().setWriteObjectSpec(specBuilder);
+
+    WriteObjectRequest request = opts.writeObjectRequest().apply(requestBuilder).build();
+
+    GrpcCallContext merge = Retrying.newCallContext();
+    StorageClient storageClient = PackagePrivateMethodWorkarounds.maybeGetStorageClient(storage);
+    assertThat(storageClient).isNotNull();
+    ApiFuture<ResumableWrite> start =
+        ResumableMedia.gapic()
+            .write()
+            .resumableWrite(
+                storageClient.startResumableWriteCallable().withDefaultCallContext(merge), request);
+
+    UnbufferedWritableByteChannelSession<WriteObjectResponse> session =
+        ResumableMedia.gapic()
+            .write()
+            .byteChannel(storageClient.writeObjectCallable())
+            .resumable()
+            .unbuffered()
+            .setStartAsync(start)
+            .build();
+
+    int additional = 13;
+    long size = objectSize + additional;
+    ByteBuffer b = DataGenerator.base64Characters().genByteBuffer(size);
+
+    UnbufferedWritableByteChannel open = session.open();
+    int written1 = open.write(b);
+    assertThat(written1).isEqualTo(objectSize);
+    assertThat(b.remaining()).isEqualTo(additional);
+
+    // no bytes should be consumed if less than 256KiB
+    int written2 = open.write(b);
+    assertThat(written2).isEqualTo(0);
+    assertThat(b.remaining()).isEqualTo(additional);
+
+    int writtenAndClose = open.writeAndClose(b);
+    assertThat(writtenAndClose).isEqualTo(additional);
+    open.close();
+    WriteObjectResponse resp = session.getResult().get(2, TimeUnit.SECONDS);
+    assertThat(resp.getResource().getSize()).isEqualTo(size);
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/PackagePrivateMethodWorkarounds.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/PackagePrivateMethodWorkarounds.java
index 23b7f371..66df95a8 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/PackagePrivateMethodWorkarounds.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/PackagePrivateMethodWorkarounds.java
@@ -21,6 +21,7 @@ import com.google.api.core.ApiFutures;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.WriteChannel;
 import com.google.common.collect.ImmutableList;
+import com.google.storage.v2.StorageClient;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Consumer;
@@ -79,6 +80,19 @@ public final class PackagePrivateMethodWorkarounds {
     }
   }
 
+  @Nullable
+  public static StorageClient maybeGetStorageClient(Storage s) {
+    if (s instanceof GrpcStorageImpl) {
+      return ((GrpcStorageImpl) s).storageClient;
+    }
+    // handle instances of AbstractStorageProxy
+    Storage service = s.getOptions().getService();
+    if (service instanceof GrpcStorageImpl) {
+      return ((GrpcStorageImpl) service).storageClient;
+    }
+    return null;
+  }
+
   public static <T> void ifNonNull(@Nullable T t, Consumer<T> c) {
     Utils.ifNonNull(t, c);
   }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/testing/RemoteStorageHelperTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/testing/RemoteStorageHelperTest.java
new file mode 100644
index 00000000..508109e3
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/testing/RemoteStorageHelperTest.java
@@ -0,0 +1,280 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.testing;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.api.gax.paging.Page;
+import com.google.cloud.http.HttpTransportOptions;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import com.google.common.collect.ImmutableList;
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.threeten.bp.Duration;
+
+public class RemoteStorageHelperTest {
+
+  private static final String BUCKET_NAME = "bucket-name";
+  private static final String PROJECT_ID = "project-id";
+  private static final String JSON_KEY =
+      "{\n"
+          + "  \"private_key_id\": \"somekeyid\",\n"
+          + "  \"private_key\": \"-----BEGIN PRIVATE KEY-----\\nMIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggS"
+          + "kAgEAAoIBAQC+K2hSuFpAdrJI\\nnCgcDz2M7t7bjdlsadsasad+fvRSW6TjNQZ3p5LLQY1kSZRqBqylRkzteMOyHg"
+          + "aR\\n0Pmxh3ILCND5men43j3h4eDbrhQBuxfEMalkG92sL+PNQSETY2tnvXryOvmBRwa/\\nQP/9dJfIkIDJ9Fw9N4"
+          + "Bhhhp6mCcRpdQjV38H7JsyJ7lih/oNjECgYAt\\nknddadwkwewcVxHFhcZJO+XWf6ofLUXpRwiTZakGMn8EE1uVa2"
+          + "LgczOjwWHGi99MFjxSer5m9\\n1tCa3/KEGKiS/YL71JvjwX3mb+cewlkcmweBKZHM2JPTk0ZednFSpVZMtycjkbLa"
+          + "\\ndYOS8V85AgMBewECggEBAKksaldajfDZDV6nGqbFjMiizAKJolr/M3OQw16K6o3/\\n0S31xIe3sSlgW0+UbYlF"
+          + "4U8KifhManD1apVSC3csafaspP4RZUHFhtBywLO9pR5c\\nr6S5aLp+gPWFyIp1pfXbWGvc5VY/v9x7ya1VEa6rXvL"
+          + "sKupSeWAW4tMj3eo/64ge\\nsdaceaLYw52KeBYiT6+vpsnYrEkAHO1fF/LavbLLOFJmFTMxmsNaG0tuiJHgjshB\\"
+          + "n82DpMCbXG9YcCgI/DbzuIjsdj2JC1cascSP//3PmefWysucBQe7Jryb6NQtASmnv\\nCdDw/0jmZTEjpe4S1lxfHp"
+          + "lAhHFtdgYTvyYtaLZiVVkCgYEA8eVpof2rceecw/I6\\n5ng1q3Hl2usdWV/4mZMvR0fOemacLLfocX6IYxT1zA1FF"
+          + "JlbXSRsJMf/Qq39mOR2\\nSpW+hr4jCoHeRVYLgsbggtrevGmILAlNoqCMpGZ6vDmJpq6ECV9olliDvpPgWOP+\\nm"
+          + "YPDreFBGxWvQrADNbRt2dmGsrsCgYEAyUHqB2wvJHFqdmeBsaacewzV8x9WgmeX\\ngUIi9REwXlGDW0Mz50dxpxcK"
+          + "CAYn65+7TCnY5O/jmL0VRxU1J2mSWyWTo1C+17L0\\n3fUqjxL1pkefwecxwecvC+gFFYdJ4CQ/MHHXU81Lwl1iWdF"
+          + "Cd2UoGddYaOF+KNeM\\nHC7cmqra+JsCgYEAlUNywzq8nUg7282E+uICfCB0LfwejuymR93CtsFgb7cRd6ak\\nECR"
+          + "8FGfCpH8ruWJINllbQfcHVCX47ndLZwqv3oVFKh6pAS/vVI4dpOepP8++7y1u\\ncoOvtreXCX6XqfrWDtKIvv0vjl"
+          + "HBhhhp6mCcRpdQjV38H7JsyJ7lih/oNjECgYAt\\nkndj5uNl5SiuVxHFhcZJO+XWf6ofLUregtevZakGMn8EE1uVa"
+          + "2AY7eafmoU/nZPT\\n00YB0TBATdCbn/nBSuKDESkhSg9s2GEKQZG5hBmL5uCMfo09z3SfxZIhJdlerreP\\nJ7gSi"
+          + "dI12N+EZxYd4xIJh/HFDgp7RRO87f+WJkofMQKBgGTnClK1VMaCRbJZPriw\\nEfeFCoOX75MxKwXs6xgrw4W//AYG"
+          + "GUjDt83lD6AZP6tws7gJ2IwY/qP7+lyhjEqN\\nHtfPZRGFkGZsdaksdlaksd323423d+15/UvrlRSFPNj1tWQmNKk"
+          + "XyRDW4IG1Oa2p\\nrALStNBx5Y9t0/LQnFI4w3aG\\n-----END PRIVATE KEY-----\\n\",\n"
+          + "  \"client_email\": \"someclientid@developer.gserviceaccount.com\",\n"
+          + "  \"client_id\": \"someclientid.apps.googleusercontent.com\",\n"
+          + "  \"type\": \"service_account\"\n"
+          + "}";
+  private static final InputStream JSON_KEY_STREAM = new ByteArrayInputStream(JSON_KEY.getBytes());
+  private static final StorageException RETRYABLE_EXCEPTION = new StorageException(409, "");
+  private static final StorageException FATAL_EXCEPTION = new StorageException(500, "");
+  private static final String BLOB_NAME2 = "n2";
+  private static final BlobId BLOB_ID1 = BlobId.of(BUCKET_NAME, "n1");
+  private static final BlobId BLOB_ID2 = BlobId.of(BUCKET_NAME, BLOB_NAME2);
+
+  private Blob blob1;
+  private Blob blob2;
+  private List<Blob> blobList;
+  private Page<Blob> blobPage;
+
+  @Before
+  public void setUp() {
+    blob1 = Mockito.mock(Blob.class);
+    blob2 = Mockito.mock(Blob.class);
+    blobList = ImmutableList.of(blob1, blob2);
+    blobPage =
+        new Page<Blob>() {
+
+          @Override
+          public boolean hasNextPage() {
+            return true;
+          }
+
+          @Override
+          public String getNextPageToken() {
+            return "nextPageCursor";
+          }
+
+          @Override
+          public Page<Blob> getNextPage() {
+            return null;
+          }
+
+          @Override
+          public Iterable<Blob> getValues() {
+            return blobList;
+          }
+
+          @Override
+          public Iterable<Blob> iterateAll() {
+            return blobList;
+          }
+        };
+  }
+
+  @Test
+  public void testForceDelete() throws InterruptedException, ExecutionException {
+    Storage storageMock = Mockito.mock(Storage.class);
+    when(blob1.getBlobId()).thenReturn(BLOB_ID1);
+    when(blob2.getBlobId()).thenReturn(BLOB_ID2);
+
+    ArrayList<BlobId> ids = new ArrayList<>();
+    ids.add(BLOB_ID1);
+    ids.add(BLOB_ID2);
+    when(storageMock.delete(ids)).thenReturn(Collections.nCopies(2, true));
+    when(storageMock.list(BUCKET_NAME, BlobListOption.versions(true))).thenReturn(blobPage);
+    when(storageMock.delete(BUCKET_NAME)).thenReturn(true);
+    assertTrue(RemoteStorageHelper.forceDelete(storageMock, BUCKET_NAME, 5, TimeUnit.SECONDS));
+
+    verify(blob1).getBlobId();
+    verify(blob2).getBlobId();
+    verify(storageMock).delete(ids);
+    verify(storageMock).list(BUCKET_NAME, BlobListOption.versions(true));
+    verify(storageMock).delete(BUCKET_NAME);
+  }
+
+  @Test
+  public void testForceDeleteTimeout() throws InterruptedException, ExecutionException {
+    Storage storageMock = Mockito.mock(Storage.class);
+
+    when(blob1.getBlobId()).thenReturn(BLOB_ID1);
+    when(blob2.getBlobId()).thenReturn(BLOB_ID2);
+
+    ArrayList<BlobId> ids = new ArrayList<>();
+    ids.add(BLOB_ID1);
+    ids.add(BLOB_ID2);
+    when(storageMock.delete(ids)).thenReturn(Collections.nCopies(2, true));
+
+    when(storageMock.list(BUCKET_NAME, BlobListOption.versions(true))).thenReturn(blobPage);
+    when(storageMock.delete(BUCKET_NAME)).thenThrow(RETRYABLE_EXCEPTION);
+    assertFalse(
+        RemoteStorageHelper.forceDelete(storageMock, BUCKET_NAME, 50, TimeUnit.MICROSECONDS));
+  }
+
+  @Test
+  public void testForceDeleteFail() throws InterruptedException, ExecutionException {
+    Storage storageMock = Mockito.mock(Storage.class);
+    when(blob1.getBlobId()).thenReturn(BLOB_ID1);
+    when(blob2.getBlobId()).thenReturn(BLOB_ID2);
+    ArrayList<BlobId> ids = new ArrayList<>();
+    ids.add(BLOB_ID1);
+    ids.add(BLOB_ID2);
+    when(storageMock.delete(ids)).thenReturn(Collections.nCopies(2, true));
+    when(storageMock.list(BUCKET_NAME, BlobListOption.versions(true))).thenReturn(blobPage);
+    when(storageMock.delete(BUCKET_NAME)).thenThrow(FATAL_EXCEPTION);
+    try {
+      RemoteStorageHelper.forceDelete(storageMock, BUCKET_NAME, 5, TimeUnit.SECONDS);
+      Assert.fail();
+    } catch (ExecutionException ex) {
+      assertNotNull(ex.getMessage());
+    } finally {
+      verify(blob1).getBlobId();
+      verify(blob2).getBlobId();
+      verify(storageMock).delete(ids);
+      verify(storageMock).list(BUCKET_NAME, BlobListOption.versions(true));
+      verify(storageMock).delete(BUCKET_NAME);
+    }
+  }
+
+  @Test
+  public void testForceDeleteNoTimeout() {
+    Storage storageMock = Mockito.mock(Storage.class);
+    when(blob1.getBlobId()).thenReturn(BLOB_ID1);
+    when(blob2.getBlobId()).thenReturn(BLOB_ID2);
+    ArrayList<BlobId> ids = new ArrayList<>();
+    ids.add(BLOB_ID1);
+    ids.add(BLOB_ID2);
+    when(storageMock.delete(ids)).thenReturn(Collections.nCopies(2, true));
+    when(storageMock.list(BUCKET_NAME, BlobListOption.versions(true))).thenReturn(blobPage);
+    when(storageMock.delete(BUCKET_NAME)).thenReturn(true);
+    RemoteStorageHelper.forceDelete(storageMock, BUCKET_NAME);
+
+    verify(blob1).getBlobId();
+    verify(blob2).getBlobId();
+    verify(storageMock).delete(ids);
+    verify(storageMock).list(BUCKET_NAME, BlobListOption.versions(true));
+    verify(storageMock).delete(BUCKET_NAME);
+  }
+
+  @Test
+  public void testForceDeleteNoTimeoutFail() {
+    Storage storageMock = Mockito.mock(Storage.class);
+    when(blob1.getBlobId()).thenReturn(BLOB_ID1);
+    when(blob2.getBlobId()).thenReturn(BLOB_ID2);
+    ArrayList<BlobId> ids = new ArrayList<>();
+    ids.add(BLOB_ID1);
+    ids.add(BLOB_ID2);
+    when(storageMock.delete(ids)).thenReturn(Collections.nCopies(2, true));
+    when(storageMock.list(BUCKET_NAME, BlobListOption.versions(true))).thenReturn(blobPage);
+    when(storageMock.delete(BUCKET_NAME)).thenThrow(FATAL_EXCEPTION);
+    try {
+      RemoteStorageHelper.forceDelete(storageMock, BUCKET_NAME);
+      Assert.fail();
+    } catch (StorageException ex) {
+      assertNotNull(ex.getMessage());
+    } finally {
+      verify(blob1).getBlobId();
+      verify(blob2).getBlobId();
+      verify(storageMock).delete(ids);
+      verify(storageMock).list(BUCKET_NAME, BlobListOption.versions(true));
+      verify(storageMock).delete(BUCKET_NAME);
+    }
+  }
+
+  @Test
+  public void testForceDeleteRetriesWithUserProject() throws Exception {
+    final String USER_PROJECT = "user-project";
+    Storage storageMock = Mockito.mock(Storage.class);
+    when(blob1.getBlobId()).thenReturn(BLOB_ID1);
+    when(blob2.getBlobId()).thenReturn(BLOB_ID2);
+    ArrayList<BlobId> ids = new ArrayList<>();
+    ids.add(BLOB_ID1);
+    ids.add(BLOB_ID2);
+    when(storageMock.delete(ids)).thenReturn(ImmutableList.of(Boolean.TRUE, Boolean.FALSE));
+    when(storageMock.delete(
+            BUCKET_NAME, BLOB_NAME2, Storage.BlobSourceOption.userProject(USER_PROJECT)))
+        .thenReturn(true);
+    when(storageMock.list(
+            BUCKET_NAME, BlobListOption.versions(true), BlobListOption.userProject(USER_PROJECT)))
+        .thenReturn(blobPage);
+    when(storageMock.delete(BUCKET_NAME, Storage.BucketSourceOption.userProject(USER_PROJECT)))
+        .thenReturn(true);
+    try {
+      RemoteStorageHelper.forceDelete(storageMock, BUCKET_NAME, 5, TimeUnit.SECONDS, USER_PROJECT);
+    } finally {
+      verify(blob1).getBlobId();
+      verify(blob2).getBlobId();
+      verify(storageMock).delete(ids);
+      verify(storageMock)
+          .delete(BUCKET_NAME, BLOB_NAME2, Storage.BlobSourceOption.userProject(USER_PROJECT));
+      verify(storageMock)
+          .list(
+              BUCKET_NAME, BlobListOption.versions(true), BlobListOption.userProject(USER_PROJECT));
+      verify(storageMock).delete(BUCKET_NAME, Storage.BucketSourceOption.userProject(USER_PROJECT));
+    }
+  }
+
+  @Test
+  public void testCreateFromStream() {
+    RemoteStorageHelper helper = RemoteStorageHelper.create(PROJECT_ID, JSON_KEY_STREAM);
+    StorageOptions options = helper.getOptions();
+    assertEquals(PROJECT_ID, options.getProjectId());
+    assertEquals(60000, ((HttpTransportOptions) options.getTransportOptions()).getConnectTimeout());
+    assertEquals(60000, ((HttpTransportOptions) options.getTransportOptions()).getReadTimeout());
+    assertEquals(10, options.getRetrySettings().getMaxAttempts());
+    assertEquals(Duration.ofMillis(30000), options.getRetrySettings().getMaxRetryDelay());
+    assertEquals(Duration.ofMillis(120000), options.getRetrySettings().getTotalTimeout());
+    assertEquals(Duration.ofMillis(250), options.getRetrySettings().getInitialRetryDelay());
+  }
+}