Repository: googleapis/java-storage
Pull request #1585 (2022-08-24 20:50:42)

test: add first two low level tests for grpc upload
two new tests that validate correct request packing and response handling when
performing a direct or resumable upload.

### Refactorings

#### ByteSizeConstants
Create a constants class to keep track of the common byte size values used
throughout the library.

#### GapicUnbufferedWritableByteChannel
Reduce the surface area on the constructor, moving Write callable, Hasher,
ByteStringStrategy, etc. out since GapicUnbufferedWritableByteChannel itself
doesn't use them.

Change ChunkSegmenter to no longer be static, instead taking a handle the
Hasher, ByteStringStrategy and maxSegmentSize which are necessary for later
computations of segmentBuffers.

Change FlusherFactory to no longer take the write callable, instead any
dependency is expected to be bound into the state of the factory. Flushers will
now maintain a handle to the write callable when they are constructed.

#### Hasher
Annotate Hasher#hash(*) methods as nullable to ensure the return value is
checked before dereferencing (via intellij warning, and eventually error prone
when we can wire that in).

#### TestUtils#subDivide(byte[], int)
Create utility method to subdivide a byte array into N ByteBuffers, each with a
window over the provided byte[] with a maximum limit of int bytes.



Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteSizeConstants.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteSizeConstants.java
new file mode 100644
index 00000000..4fb27685
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteSizeConstants.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright 2022 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;
+
+final class ByteSizeConstants {
+
+  static final int _1KiB = 1024;
+  static final int _128KiB = 128 * _1KiB;
+  static final int _256KiB = 256 * _1KiB;
+  static final int _384KiB = 384 * _1KiB;
+  static final int _512KiB = 512 * _1KiB;
+  static final int _1MiB = 1024 * _1KiB;
+  static final int _2MiB = 2 * _1MiB;
+  static final int _15MiB = 15 * _1MiB;
+  static final int _16MiB = 16 * _1MiB;
+  static final int _32MiB = 32 * _1MiB;
+
+  private ByteSizeConstants() {}
+}
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 9586eab4..b891bb3b 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
@@ -17,6 +17,7 @@
 package com.google.cloud.storage;
 
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.ByteString;
 import java.nio.ByteBuffer;
 import java.util.ArrayDeque;
@@ -28,6 +29,26 @@ import org.checkerframework.checker.nullness.qual.Nullable;
  * segments. Each resulting segment will then become an individual message.
  */
 final class ChunkSegmenter {
+  private final Hasher hasher;
+  private final ByteStringStrategy bss;
+  private final int maxSegmentSize;
+  private final int blockSize;
+
+  ChunkSegmenter(Hasher hasher, ByteStringStrategy bss, int maxSegmentSize) {
+    this(hasher, bss, maxSegmentSize, ByteSizeConstants._256KiB);
+  }
+
+  @VisibleForTesting
+  ChunkSegmenter(Hasher hasher, ByteStringStrategy bss, int maxSegmentSize, int blockSize) {
+    this.hasher = hasher;
+    this.bss = bss;
+    this.maxSegmentSize = maxSegmentSize;
+    this.blockSize = blockSize;
+  }
+
+  Hasher getHasher() {
+    return hasher;
+  }
 
   /**
    * Given {@code bbs}, yield N segments, where each segment is at most {@code maxSegmentSize}
@@ -47,20 +68,13 @@ final class ChunkSegmenter {
    *
    * Each segment will conditionally compute a crc32c value depending upon {@code hasher}.
    *
-   * @see #segmentBuffers(ByteBuffer[], Hasher, ByteStringStrategy, int, int, int)
+   * @see #segmentBuffers(ByteBuffer[], int, int)
    */
-  static ChunkSegment[] segmentBuffers(
-      ByteBuffer[] bbs, Hasher hasher, ByteStringStrategy bss, int maxSegmentSize) {
-    return segmentBuffers(bbs, hasher, bss, maxSegmentSize, 0, bbs.length);
+  ChunkSegment[] segmentBuffers(ByteBuffer[] bbs) {
+    return segmentBuffers(bbs, 0, bbs.length);
   }
 
-  static ChunkSegment[] segmentBuffers(
-      ByteBuffer[] bbs,
-      Hasher hasher,
-      ByteStringStrategy bss,
-      int maxSegmentSize,
-      int offset,
-      int length) {
+  ChunkSegment[] segmentBuffers(ByteBuffer[] bbs, int offset, int length) {
     Deque<ChunkSegment> data = new ArrayDeque<>();
 
     for (int i = offset; i < length; i++) {
@@ -71,13 +85,13 @@ final class ChunkSegmenter {
         ChunkSegment peekLast = data.peekLast();
         if (peekLast == null || peekLast.b.size() == maxSegmentSize) {
           int limit = Math.min(remaining, maxSegmentSize);
-          ChunkSegment datum = getData(buffer, hasher, bss, limit);
+          ChunkSegment datum = newSegment(buffer, limit);
           data.addLast(datum);
         } else {
           ChunkSegment chunkSoFar = data.pollLast();
           //noinspection ConstantConditions -- covered by peekLast check above
           int limit = Math.min(remaining, maxSegmentSize - chunkSoFar.b.size());
-          ChunkSegment datum = getData(buffer, hasher, bss, limit);
+          ChunkSegment datum = newSegment(buffer, limit);
           ChunkSegment plus = chunkSoFar.concat(datum);
           data.addLast(plus);
         }
@@ -87,8 +101,7 @@ final class ChunkSegmenter {
     return data.toArray(new ChunkSegment[0]);
   }
 
-  private static ChunkSegment getData(
-      ByteBuffer buffer, Hasher hasher, ByteStringStrategy bss, int limit) {
+  private ChunkSegment newSegment(ByteBuffer buffer, int limit) {
     final ByteBuffer slice = buffer.slice();
     slice.limit(limit);
 
@@ -99,12 +112,14 @@ final class ChunkSegmenter {
     return new ChunkSegment(byteString, hash);
   }
 
-  static final class ChunkSegment {
+  final class ChunkSegment {
     private final ByteString b;
     @Nullable private final Crc32cLengthKnown crc32c;
+    private final boolean onlyFullBlocks;
 
-    public ChunkSegment(ByteString b, @Nullable Crc32cLengthKnown crc32c) {
+    private ChunkSegment(ByteString b, @Nullable Crc32cLengthKnown crc32c) {
       this.b = b;
+      this.onlyFullBlocks = b.size() % blockSize == 0;
       this.crc32c = crc32c;
     }
 
@@ -113,7 +128,8 @@ final class ChunkSegmenter {
       if (crc32c != null && other.crc32c != null) {
         newCrc = crc32c.concat(other.crc32c);
       }
-      return new ChunkSegment(b.concat(other.b), newCrc);
+      ByteString concat = b.concat(other.b);
+      return new ChunkSegment(concat, newCrc);
     }
 
     public ByteString getB() {
@@ -124,5 +140,9 @@ final class ChunkSegmenter {
     public Crc32cLengthKnown getCrc32c() {
       return crc32c;
     }
+
+    public boolean isOnlyFullBlocks() {
+      return onlyFullBlocks;
+    }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannel.java
index b1f2150a..daea9281 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannel.java
@@ -134,15 +134,19 @@ final class GapicUnbufferedReadableByteChannel
         }
         ChecksummedData checksummedData = resp.getChecksummedData();
         ByteBuffer content = checksummedData.getContent().asReadOnlyByteBuffer();
-        Crc32cLengthKnown expected =
-            Crc32cValue.of(checksummedData.getCrc32C(), checksummedData.getContent().size());
-        cumulativeCrc32c = hasher.nullSafeConcat(cumulativeCrc32c, expected);
-        try {
-          hasher.validate(expected, content::duplicate);
-        } catch (IOException e) {
-          ioExceptionAlreadyThrown = true;
-          close();
-          throw e;
+        // very important to know whether a crc32c value is set. Without checking, protobuf will
+        // happily return 0, which is a valid crc32c value.
+        if (checksummedData.hasCrc32C()) {
+          Crc32cLengthKnown expected =
+              Crc32cValue.of(checksummedData.getCrc32C(), checksummedData.getContent().size());
+          cumulativeCrc32c = hasher.nullSafeConcat(cumulativeCrc32c, expected);
+          try {
+            hasher.validate(expected, content::duplicate);
+          } catch (IOException e) {
+            ioExceptionAlreadyThrown = true;
+            close();
+            throw e;
+          }
         }
         copy(c, content, dsts, offset, length);
         if (content.hasRemaining()) {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
index afb6fa65..240113b8 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
@@ -17,7 +17,6 @@
 package com.google.cloud.storage;
 
 import com.google.api.core.SettableApiFuture;
-import com.google.api.gax.rpc.ClientStreamingCallable;
 import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
@@ -27,7 +26,6 @@ import com.google.cloud.storage.WriteFlushStrategy.FlusherFactory;
 import com.google.protobuf.ByteString;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.ObjectChecksums;
-import com.google.storage.v2.ServiceConstants.Values;
 import com.google.storage.v2.WriteObjectRequest;
 import com.google.storage.v2.WriteObjectResponse;
 import java.io.IOException;
@@ -41,11 +39,9 @@ final class GapicUnbufferedWritableByteChannel<
     implements UnbufferedWritableByteChannel {
 
   private final SettableApiFuture<WriteObjectResponse> resultFuture;
-  private final ByteStringStrategy byteStringStrategy;
-  private final Hasher hasher;
-  private final WriteCtx<RequestFactoryT> writeCtx;
-  private final int perMessageLimit;
+  private final ChunkSegmenter chunkSegmenter;
 
+  private final WriteCtx<RequestFactoryT> writeCtx;
   private final Flusher flusher;
 
   private boolean open = true;
@@ -53,19 +49,15 @@ final class GapicUnbufferedWritableByteChannel<
 
   GapicUnbufferedWritableByteChannel(
       SettableApiFuture<WriteObjectResponse> resultFuture,
-      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+      ChunkSegmenter chunkSegmenter,
       RequestFactoryT requestFactory,
-      ByteStringStrategy byteStringStrategy,
-      Hasher hasher,
-      FlusherFactory flusher) {
+      FlusherFactory flusherFactory) {
     this.resultFuture = resultFuture;
-    this.byteStringStrategy = byteStringStrategy;
-    this.hasher = hasher;
+    this.chunkSegmenter = chunkSegmenter;
 
     this.writeCtx = new WriteCtx<>(requestFactory);
-    this.perMessageLimit = Values.MAX_WRITE_CHUNK_BYTES_VALUE;
     this.flusher =
-        flusher.newFlusher(write, writeCtx.getConfirmedBytes()::addAndGet, resultFuture::set);
+        flusherFactory.newFlusher(writeCtx.getConfirmedBytes()::addAndGet, resultFuture::set);
   }
 
   @Override
@@ -84,9 +76,7 @@ final class GapicUnbufferedWritableByteChannel<
       throw new ClosedChannelException();
     }
 
-    ChunkSegment[] data =
-        ChunkSegmenter.segmentBuffers(
-            srcs, hasher, byteStringStrategy, perMessageLimit, srcsOffset, srcLength);
+    ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcLength);
 
     List<WriteObjectRequest> messages = new ArrayList<>();
 
@@ -98,7 +88,9 @@ final class GapicUnbufferedWritableByteChannel<
       long offset = writeCtx.getTotalSentBytes().getAndAdd(contentSize);
       ChecksummedData.Builder checksummedData = ChecksummedData.newBuilder().setContent(b);
       if (crc32c != null) {
-        writeCtx.getCumulativeCrc32c().getAndAccumulate(crc32c, hasher::nullSafeConcat);
+        writeCtx
+            .getCumulativeCrc32c()
+            .getAndAccumulate(crc32c, chunkSegmenter.getHasher()::nullSafeConcat);
         checksummedData.setCrc32C(crc32c.getValue());
       }
       WriteObjectRequest.Builder builder =
@@ -106,7 +98,7 @@ final class GapicUnbufferedWritableByteChannel<
               .newRequestBuilder()
               .setWriteOffset(offset)
               .setChecksummedData(checksummedData.build());
-      if (contentSize < perMessageLimit) {
+      if (!datum.isOnlyFullBlocks()) {
         builder.setFinishWrite(true);
         if (crc32c != null) {
           builder.setObjectChecksums(
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
index 553cda89..f34c1c67 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
@@ -27,6 +27,7 @@ import com.google.cloud.storage.ChannelSession.UnbufferedWriteSession;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
 import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
 import com.google.cloud.storage.WriteFlushStrategy.FlusherFactory;
+import com.google.storage.v2.ServiceConstants.Values;
 import com.google.storage.v2.WriteObjectRequest;
 import com.google.storage.v2.WriteObjectResponse;
 import java.nio.ByteBuffer;
@@ -35,7 +36,7 @@ import java.util.function.Function;
 
 final class GapicWritableByteChannelSessionBuilder {
 
-  private static final int DEFAULT_BUFFER_CAPACITY = 16 * 1024 * 1024;
+  private static final int DEFAULT_BUFFER_CAPACITY = ByteSizeConstants._16MiB;
   private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
   private Hasher hasher;
   private ByteStringStrategy byteStringStrategy;
@@ -124,7 +125,10 @@ final class GapicWritableByteChannelSessionBuilder {
     Hasher boundHasher = hasher;
     return (start, resultFuture) ->
         new GapicUnbufferedWritableByteChannel<>(
-            resultFuture, write, f.apply(start), boundStrategy, boundHasher, flusherFactory);
+            resultFuture,
+            new ChunkSegmenter(boundHasher, boundStrategy, Values.MAX_WRITE_CHUNK_BYTES_VALUE),
+            f.apply(start),
+            flusherFactory);
   }
 
   final class DirectUploadBuilder {
@@ -212,7 +216,8 @@ final class GapicWritableByteChannelSessionBuilder {
     UnbufferedWritableByteChannelSession<WriteObjectResponse> build() {
       return new UnbufferedWriteSession<>(
           ApiFutures.immediateFuture(requireNonNull(req, "req must be non null")),
-          bindFunction(WriteFlushStrategy::fsyncOnClose, WriteObjectRequestBuilderFactory::simple)
+          bindFunction(
+                  WriteFlushStrategy.fsyncOnClose(write), WriteObjectRequestBuilderFactory::simple)
               .andThen(StorageByteChannels.writable()::createSynchronized));
     }
   }
@@ -235,7 +240,8 @@ final class GapicWritableByteChannelSessionBuilder {
     BufferedWritableByteChannelSession<WriteObjectResponse> build() {
       return new BufferedWriteSession<>(
           ApiFutures.immediateFuture(requireNonNull(req, "req must be non null")),
-          bindFunction(WriteFlushStrategy::fsyncOnClose, WriteObjectRequestBuilderFactory::simple)
+          bindFunction(
+                  WriteFlushStrategy.fsyncOnClose(write), WriteObjectRequestBuilderFactory::simple)
               .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
               .andThen(StorageByteChannels.writable()::createSynchronized));
     }
@@ -257,7 +263,7 @@ final class GapicWritableByteChannelSessionBuilder {
     UnbufferedWritableByteChannelSession<WriteObjectResponse> build() {
       return new UnbufferedWriteSession<>(
           requireNonNull(start, "start must be non null"),
-          bindFunction(WriteFlushStrategy::fsyncEveryFlush, ResumableWrite::identity)
+          bindFunction(WriteFlushStrategy.fsyncEveryFlush(write), ResumableWrite::identity)
               .andThen(StorageByteChannels.writable()::createSynchronized));
     }
   }
@@ -284,7 +290,7 @@ final class GapicWritableByteChannelSessionBuilder {
     BufferedWritableByteChannelSession<WriteObjectResponse> build() {
       return new BufferedWriteSession<>(
           requireNonNull(start, "start must be non null"),
-          bindFunction(WriteFlushStrategy::fsyncEveryFlush, ResumableWrite::identity)
+          bindFunction(WriteFlushStrategy.fsyncEveryFlush(write), ResumableWrite::identity)
               .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
               .andThen(StorageByteChannels.writable()::createSynchronized));
     }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java
index 78f01b47..14fa1c2e 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java
@@ -16,8 +16,8 @@
 
 package com.google.cloud.storage;
 
-import static com.google.cloud.storage.GrpcStorageImpl._15MiB;
-import static com.google.cloud.storage.GrpcStorageImpl._256KiB;
+import static com.google.cloud.storage.ByteSizeConstants._15MiB;
+import static com.google.cloud.storage.ByteSizeConstants._256KiB;
 import static com.google.cloud.storage.Utils.todo;
 
 import com.google.api.core.ApiFuture;
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index d6e77ea9..dcdf2952 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -16,6 +16,8 @@
 
 package com.google.cloud.storage;
 
+import static com.google.cloud.storage.ByteSizeConstants._15MiB;
+import static com.google.cloud.storage.ByteSizeConstants._256KiB;
 import static com.google.cloud.storage.Utils.bucketNameCodec;
 import static com.google.cloud.storage.Utils.ifNonNull;
 import static com.google.cloud.storage.Utils.projectNameCodec;
@@ -147,9 +149,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
           .map(c -> ApiExceptionFactory.createException(null, c, false))
           .collect(Collectors.toSet());
 
-  static final int _256KiB = 256 * 1024;
-  static final int _15MiB = 15 * 1024 * 1024;
-
   private final StorageClient storageClient;
   private final GrpcConversions codecs;
   private final GrpcRetryAlgorithmManager retryAlgorithmManager;
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Hasher.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Hasher.java
index b89fe7c6..06fca041 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Hasher.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Hasher.java
@@ -26,10 +26,12 @@ import org.checkerframework.checker.nullness.qual.Nullable;
 
 interface Hasher {
 
+  @Nullable
   default Crc32cLengthKnown hash(Supplier<ByteBuffer> b) {
     return hash(b.get());
   }
 
+  @Nullable
   Crc32cLengthKnown hash(ByteBuffer b);
 
   void validate(Crc32cValue<?> expected, Supplier<ByteBuffer> b) throws IOException;
@@ -77,6 +79,7 @@ interface Hasher {
       return Crc32cValue.of(Hashing.crc32c().hashBytes(b).asInt(), remaining);
     }
 
+    @SuppressWarnings("ConstantConditions")
     @Override
     public void validate(Crc32cValue<?> expected, Supplier<ByteBuffer> b) throws IOException {
       Crc32cLengthKnown actual = hash(b);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
index 17e599e0..b804ba4c 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
@@ -35,12 +35,11 @@ final class ResumableWrite implements WriteObjectRequestBuilderFactory {
   public ResumableWrite(StartResumableWriteRequest req, StartResumableWriteResponse res) {
     this.req = req;
     this.res = res;
-    this.writeRequest =
-        WriteObjectRequest.newBuilder()
-            .setUploadId(res.getUploadId())
-            .setCommonObjectRequestParams(req.getCommonObjectRequestParams())
-            // REVIEW: how does kms come into play here?
-            .build();
+    WriteObjectRequest.Builder b = WriteObjectRequest.newBuilder().setUploadId(res.getUploadId());
+    if (req.hasCommonObjectRequestParams()) {
+      b.setCommonObjectRequestParams(req.getCommonObjectRequestParams());
+    }
+    this.writeRequest = b.build();
   }
 
   public StartResumableWriteRequest getReq() {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
index e8a7978f..8c22ba6e 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
@@ -3471,7 +3471,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
-  @TransportCompatibility({Transport.HTTP})
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   HmacKey createHmacKey(ServiceAccount serviceAccount, CreateHmacKeyOption... options);
 
   /**
@@ -3523,7 +3523,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
-  @TransportCompatibility({Transport.HTTP})
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   HmacKeyMetadata getHmacKey(String accessId, GetHmacKeyOption... options);
 
   /**
@@ -3560,7 +3560,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
-  @TransportCompatibility({Transport.HTTP})
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   HmacKeyMetadata updateHmacKeyState(
       final HmacKeyMetadata hmacKeyMetadata,
       final HmacKey.HmacKeyState state,
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
index 07207b07..708de195 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
@@ -88,7 +88,11 @@ public final class StorageException extends BaseHttpServiceException {
   }
 
   private static StorageException getStorageException(Throwable t) {
-    return new StorageException(UNKNOWN_CODE, t.getMessage(), t.getCause());
+    // unwrap a RetryHelperException if that is what is being translated
+    if (t instanceof RetryHelperException) {
+      return new StorageException(UNKNOWN_CODE, t.getMessage(), t.getCause());
+    }
+    return new StorageException(UNKNOWN_CODE, t.getMessage(), t);
   }
 
   /**
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
index 1104c18b..8dfe99f2 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
@@ -49,24 +49,24 @@ final class WriteFlushStrategy {
    * Create a {@link Flusher} which will "fsync" every time {@link Flusher#flush(List)} is called
    * along with {@link Flusher#close(WriteObjectRequest)}.
    */
-  static Flusher fsyncEveryFlush(
-      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
-      LongConsumer committedTotalBytesCallback,
-      Consumer<WriteObjectResponse> onSuccessCallback) {
-    return new FsyncEveryFlusher(write, committedTotalBytesCallback, onSuccessCallback);
+  static FlusherFactory fsyncEveryFlush(
+      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write) {
+    return (LongConsumer committedTotalBytesCallback,
+        Consumer<WriteObjectResponse> onSuccessCallback) ->
+        new FsyncEveryFlusher(write, committedTotalBytesCallback, onSuccessCallback);
   }
 
   /**
    * Create a {@link Flusher} which will "fsync" only on {@link Flusher#close(WriteObjectRequest)}.
    * Calls to {@link Flusher#flush(List)} will be sent but not synced.
    *
-   * @see FlusherFactory#newFlusher(ClientStreamingCallable, LongConsumer, Consumer)
+   * @see FlusherFactory#newFlusher(LongConsumer, Consumer)
    */
-  static Flusher fsyncOnClose(
-      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
-      LongConsumer committedTotalBytesCallback,
-      Consumer<WriteObjectResponse> onSuccessCallback) {
-    return new FsyncOnClose(write, committedTotalBytesCallback, onSuccessCallback);
+  static FlusherFactory fsyncOnClose(
+      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write) {
+    return (LongConsumer committedTotalBytesCallback,
+        Consumer<WriteObjectResponse> onSuccessCallback) ->
+        new FsyncOnClose(write, committedTotalBytesCallback, onSuccessCallback);
   }
 
   @FunctionalInterface
@@ -77,9 +77,7 @@ final class WriteFlushStrategy {
      * @param onSuccessCallback Callback to signal success, and provide the final response.
      */
     Flusher newFlusher(
-        ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
-        LongConsumer committedTotalBytesCallback,
-        Consumer<WriteObjectResponse> onSuccessCallback);
+        LongConsumer committedTotalBytesCallback, Consumer<WriteObjectResponse> onSuccessCallback);
   }
 
   interface Flusher {
@@ -201,9 +199,7 @@ final class WriteFlushStrategy {
       if (value.hasPersistedSize()) {
         sizeCallback.accept(value.getPersistedSize());
       } else if (value.hasResource()) {
-        // testbench_seems to return this even on finish
-      } else {
-        // serverState.confirmedBytes.set(serverState.totalSentBytes.get());
+        sizeCallback.accept(value.getResource().getSize());
       }
       last = value;
     }


Test patch:
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 7da2f705..9a1a6a03 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
@@ -41,8 +41,8 @@ final class ChunkSegmenterTest {
     System.out.println("td = " + td);
 
     ChunkSegment[] data =
-        ChunkSegmenter.segmentBuffers(
-            td.buffers, Hasher.noop(), ByteStringStrategy.noCopy(), td.chunkSize);
+        new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.noCopy(), td.chunkSize)
+            .segmentBuffers(td.buffers);
 
     long dataTotalSize = Arrays.stream(data).mapToLong(d -> d.getB().size()).sum();
     Optional<Crc32cLengthKnown> reduce =
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannelTest.java
index 092f5405..27603447 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannelTest.java
@@ -26,6 +26,7 @@ import com.google.api.core.ApiFutures;
 import com.google.api.gax.rpc.StatusCode;
 import com.google.cloud.storage.ChannelSession.UnbufferedReadSession;
 import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
+import com.google.common.collect.ImmutableList;
 import com.google.common.hash.Hashing;
 import com.google.protobuf.ByteString;
 import com.google.storage.v2.ContentRange;
@@ -40,13 +41,10 @@ import io.grpc.stub.StreamObserver;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
-import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
 import org.junit.Test;
 
 public final class GapicUnbufferedReadableByteChannelTest {
@@ -73,20 +71,24 @@ public final class GapicUnbufferedReadableByteChannelTest {
           .setContentRange(ContentRange.newBuilder().setStart(0).build())
           .setObjectChecksums(
               ObjectChecksums.newBuilder().setCrc32C(Hashing.crc32c().hashBytes(bytes).asInt()))
-          .setChecksummedData(getChecksummedData(data1))
+          .setChecksummedData(getChecksummedData(data1, Hasher.enabled()))
           .build();
   private final ReadObjectResponse resp2 =
-      ReadObjectResponse.newBuilder().setChecksummedData(getChecksummedData(data2)).build();
+      ReadObjectResponse.newBuilder()
+          .setChecksummedData(getChecksummedData(data2, Hasher.enabled()))
+          .build();
   private final ReadObjectResponse resp3 =
       ReadObjectResponse.newBuilder()
           .setMetadata(expectedResult)
           .setContentRange(ContentRange.newBuilder().setStart(20).build())
           .setObjectChecksums(
               ObjectChecksums.newBuilder().setCrc32C(Hashing.crc32c().hashBytes(bytes).asInt()))
-          .setChecksummedData(getChecksummedData(data3))
+          .setChecksummedData(getChecksummedData(data3, Hasher.enabled()))
           .build();
   private final ReadObjectResponse resp4 =
-      ReadObjectResponse.newBuilder().setChecksummedData(getChecksummedData(data4)).build();
+      ReadObjectResponse.newBuilder()
+          .setChecksummedData(getChecksummedData(data4, Hasher.enabled()))
+          .build();
 
   /** Define a Storage service that will always return an error during the first readObject */
   private final StorageGrpc.StorageImplBase fakeStorage =
@@ -155,11 +157,7 @@ public final class GapicUnbufferedReadableByteChannelTest {
                       start,
                       Hasher.noop()));
       byte[] actualBytes = new byte[40];
-      List<ByteBuffer> buffers =
-          IntStream.iterate(0, i -> i + 2)
-              .limit(actualBytes.length / 2)
-              .mapToObj(i -> ByteBuffer.wrap(actualBytes, i, 2))
-              .collect(Collectors.toList());
+      ImmutableList<ByteBuffer> buffers = TestUtils.subDivide(actualBytes, 2);
       try (UnbufferedReadableByteChannel c = session.open()) {
         for (ByteBuffer buf : buffers) {
           c.read(buf);
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
new file mode 100644
index 00000000..bc32ccea
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
@@ -0,0 +1,237 @@
+/*
+ * Copyright 2022 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.common.truth.Truth.assertThat;
+
+import com.google.api.core.SettableApiFuture;
+import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.protobuf.ByteString;
+import com.google.storage.v2.Object;
+import com.google.storage.v2.StartResumableWriteRequest;
+import com.google.storage.v2.StartResumableWriteResponse;
+import com.google.storage.v2.StorageClient;
+import com.google.storage.v2.StorageGrpc.StorageImplBase;
+import com.google.storage.v2.WriteObjectRequest;
+import com.google.storage.v2.WriteObjectResponse;
+import com.google.storage.v2.WriteObjectSpec;
+import io.grpc.Status.Code;
+import io.grpc.stub.CallStreamObserver;
+import io.grpc.stub.StreamObserver;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import org.junit.Test;
+
+public final class GapicUnbufferedWritableByteChannelTest {
+
+  ChunkSegmenter segmenter = new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.copy(), 10, 5);
+
+  @Test
+  public void directUpload() throws IOException, InterruptedException, ExecutionException {
+    Object obj = Object.newBuilder().setBucket("buck").setName("obj").build();
+    WriteObjectSpec spec = WriteObjectSpec.newBuilder().setResource(obj).build();
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(40);
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder()
+            .setWriteObjectSpec(spec)
+            .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 0, 10)))
+            .build();
+    WriteObjectRequest req2 =
+        WriteObjectRequest.newBuilder()
+            .setWriteOffset(10)
+            .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 10, 10)))
+            .build();
+    WriteObjectRequest req3 =
+        WriteObjectRequest.newBuilder()
+            .setWriteOffset(20)
+            .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 20, 10)))
+            .build();
+    WriteObjectRequest req4 =
+        WriteObjectRequest.newBuilder()
+            .setWriteOffset(30)
+            .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 30, 10)))
+            .build();
+    WriteObjectRequest req5 =
+        WriteObjectRequest.newBuilder().setWriteOffset(40).setFinishWrite(true).build();
+
+    WriteObjectResponse resp =
+        WriteObjectResponse.newBuilder().setResource(obj.toBuilder().setSize(40)).build();
+
+    WriteObjectRequest base = WriteObjectRequest.newBuilder().setWriteObjectSpec(spec).build();
+    WriteObjectRequestBuilderFactory reqFactory = base::toBuilder;
+
+    StorageImplBase service =
+        new DirectWriteService(
+            ImmutableMap.of(ImmutableList.of(req1, req2, req3, req4, req5), resp));
+    try (FakeServer fake = FakeServer.of(service);
+        StorageClient sc = StorageClient.create(fake.storageSettings())) {
+      SettableApiFuture<WriteObjectResponse> result = SettableApiFuture.create();
+      try (GapicUnbufferedWritableByteChannel<?> c =
+          new GapicUnbufferedWritableByteChannel<>(
+              result,
+              segmenter,
+              reqFactory,
+              WriteFlushStrategy.fsyncOnClose(sc.writeObjectCallable()))) {
+        c.write(ByteBuffer.wrap(bytes));
+      }
+      assertThat(result.get()).isEqualTo(resp);
+    }
+  }
+
+  @Test
+  public void resumableUpload() throws IOException, InterruptedException, ExecutionException {
+    String uploadId = "upload-id";
+
+    Object obj = Object.newBuilder().setBucket("buck").setName("obj").build();
+    WriteObjectSpec spec = WriteObjectSpec.newBuilder().setResource(obj).build();
+
+    StartResumableWriteRequest startReq =
+        StartResumableWriteRequest.newBuilder().setWriteObjectSpec(spec).build();
+    StartResumableWriteResponse startResp =
+        StartResumableWriteResponse.newBuilder().setUploadId(uploadId).build();
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(40);
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 0, 10)))
+            .build();
+    WriteObjectRequest req2 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(10)
+            .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 10, 10)))
+            .build();
+    WriteObjectRequest req3 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(20)
+            .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 20, 10)))
+            .build();
+    WriteObjectRequest req4 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(30)
+            .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 30, 10)))
+            .build();
+    WriteObjectRequest req5 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(40)
+            .setFinishWrite(true)
+            .build();
+
+    WriteObjectResponse resp1 = WriteObjectResponse.newBuilder().setPersistedSize(10).build();
+    WriteObjectResponse resp2 = WriteObjectResponse.newBuilder().setPersistedSize(20).build();
+    WriteObjectResponse resp3 = WriteObjectResponse.newBuilder().setPersistedSize(30).build();
+    WriteObjectResponse resp4 = WriteObjectResponse.newBuilder().setPersistedSize(40).build();
+    WriteObjectResponse resp5 =
+        WriteObjectResponse.newBuilder().setResource(obj.toBuilder().setSize(40)).build();
+
+    WriteObjectRequestBuilderFactory reqFactory = new ResumableWrite(startReq, startResp);
+
+    ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> writes =
+        ImmutableMap.<List<WriteObjectRequest>, WriteObjectResponse>builder()
+            .put(ImmutableList.of(req1), resp1)
+            .put(ImmutableList.of(req2), resp2)
+            .put(ImmutableList.of(req3), resp3)
+            .put(ImmutableList.of(req4), resp4)
+            .put(ImmutableList.of(req5), resp5)
+            .build();
+    StorageImplBase service = new DirectWriteService(writes);
+    try (FakeServer fake = FakeServer.of(service);
+        StorageClient sc = StorageClient.create(fake.storageSettings())) {
+      SettableApiFuture<WriteObjectResponse> result = SettableApiFuture.create();
+      try (GapicUnbufferedWritableByteChannel<?> c =
+          new GapicUnbufferedWritableByteChannel<>(
+              result,
+              segmenter,
+              reqFactory,
+              WriteFlushStrategy.fsyncEveryFlush(sc.writeObjectCallable()))) {
+        ImmutableList<ByteBuffer> buffers = TestUtils.subDivide(bytes, 10);
+        for (ByteBuffer buf : buffers) {
+          c.write(buf);
+        }
+      }
+      assertThat(result.get()).isEqualTo(resp5);
+    }
+  }
+
+  private static class DirectWriteService extends StorageImplBase {
+    private final ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> writes;
+
+    private ImmutableList.Builder<WriteObjectRequest> requests;
+
+    private DirectWriteService(ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> writes) {
+      this.writes = writes;
+      this.requests = new ImmutableList.Builder<>();
+    }
+
+    @Override
+    public StreamObserver<WriteObjectRequest> writeObject(StreamObserver<WriteObjectResponse> obs) {
+      return new Adapter() {
+        @Override
+        public void onNext(WriteObjectRequest value) {
+          requests.add(value);
+        }
+
+        @Override
+        public void onError(Throwable t) {}
+
+        @Override
+        public void onCompleted() {
+          ImmutableList<WriteObjectRequest> build = requests.build();
+          if (writes.containsKey(build)) {
+            requests = new ImmutableList.Builder<>();
+            obs.onNext(writes.get(build));
+            obs.onCompleted();
+          } else {
+            obs.onError(
+                TestUtils.apiException(Code.PERMISSION_DENIED, "Unexpected request chain."));
+          }
+        }
+      };
+    }
+  }
+
+  private abstract static class Adapter extends CallStreamObserver<WriteObjectRequest> {
+
+    private Adapter() {}
+
+    @Override
+    public boolean isReady() {
+      return true;
+    }
+
+    @Override
+    public void setOnReadyHandler(Runnable onReadyHandler) {}
+
+    @Override
+    public void disableAutoInboundFlowControl() {}
+
+    @Override
+    public void request(int count) {}
+
+    @Override
+    public void setMessageCompression(boolean enable) {}
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
index 28f0ae7d..573c2781 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
@@ -22,9 +22,9 @@ import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.ApiExceptionFactory;
 import com.google.api.gax.rpc.ErrorDetails;
 import com.google.api.gax.rpc.StatusCode;
+import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
-import com.google.common.hash.Hashing;
 import com.google.protobuf.Any;
 import com.google.protobuf.ByteString;
 import com.google.rpc.DebugInfo;
@@ -34,6 +34,11 @@ import io.grpc.StatusRuntimeException;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.nio.Buffer;
+import java.nio.ByteBuffer;
+import java.util.function.Function;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
 import java.util.zip.GZIPOutputStream;
 
 public final class TestUtils {
@@ -53,14 +58,27 @@ public final class TestUtils {
   }
 
   public static ChecksummedData getChecksummedData(ByteString content) {
-    int crc32c = Hashing.crc32c().hashBytes(content.asReadOnlyByteBuffer()).asInt();
-    return ChecksummedData.newBuilder().setContent(content).setCrc32C(crc32c).build();
+    return ChecksummedData.newBuilder().setContent(content).build();
+  }
+
+  public static ChecksummedData getChecksummedData(ByteString content, Hasher hasher) {
+    ChecksummedData.Builder b = ChecksummedData.newBuilder().setContent(content);
+    Crc32cLengthKnown hash = hasher.hash(content.asReadOnlyByteBuffer());
+    if (hash != null) {
+      int crc32c = hash.getValue();
+      b.setCrc32C(crc32c);
+    }
+    return b.build();
   }
 
   public static ApiException apiException(Code code) {
+    return apiException(code, "");
+  }
+
+  public static ApiException apiException(Code code, String message) {
     StatusRuntimeException statusRuntimeException = code.toStatus().asRuntimeException();
     DebugInfo debugInfo =
-        DebugInfo.newBuilder().setDetail("forced failure |~| " + code.name()).build();
+        DebugInfo.newBuilder().setDetail("forced failure |~| " + code.name() + message).build();
     ErrorDetails errorDetails =
         ErrorDetails.builder().setRawErrorMessages(ImmutableList.of(Any.pack(debugInfo))).build();
     return ApiExceptionFactory.createException(
@@ -70,4 +88,24 @@ public final class TestUtils {
   public static GrpcCallContext contextWithRetryForCodes(StatusCode.Code... code) {
     return GrpcCallContext.createDefault().withRetryableCodes(ImmutableSet.copyOf(code));
   }
+
+  public static ImmutableList<ByteBuffer> subDivide(byte[] bytes, int division) {
+    int length = bytes.length;
+    int fullDivisions = length / division;
+    int x = division * fullDivisions;
+    int remaining = length - x;
+
+    if ((fullDivisions == 1 && remaining == 0) || (fullDivisions == 0 && remaining == 1)) {
+      return ImmutableList.of(ByteBuffer.wrap(bytes));
+    } else {
+      return Stream.of(
+              IntStream.iterate(0, i -> i + division)
+                  .limit(fullDivisions)
+                  .mapToObj(i -> ByteBuffer.wrap(bytes, i, division)),
+              Stream.of(ByteBuffer.wrap(bytes, x, remaining)))
+          .flatMap(Function.identity())
+          .filter(Buffer::hasRemaining)
+          .collect(ImmutableList.toImmutableList());
+    }
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/TransportCompatibilityTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/TransportCompatibilityTest.java
index e2042b01..9b346aca 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/TransportCompatibilityTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/TransportCompatibilityTest.java
@@ -31,12 +31,13 @@ public final class TransportCompatibilityTest {
 
   @Test
   public void verifyUnsupportedMethodsGenerateMeaningfulException() {
-    Storage s =
+    GrpcStorageOptions options =
         StorageOptions.grpc()
             .setProjectId("blank")
             .setCredentials(NoCredentials.getInstance())
-            .build()
-            .getService();
+            .build();
+    @SuppressWarnings("resource")
+    Storage s = new GrpcStorageImpl(options, null);
     ImmutableList<String> messages =
         Stream.<Supplier<?>>of(
                 s::batch,