Repository: googleapis/java-storage
Pull request #1571 (2022-08-12 22:18:03)

chore: implement retries for List operations in GrpcStorageImpl
Update TransformPageDecorator to apply automatic retires to individual page fetches.

To accomplish page granularity for retries I've replaced TransformingIterator with
a page producing Stream, where each page can then be expected into a single Iterator.

Add new interface Retrying.RetryingDependencies to provide a marker for the need
of RetrySettings and an ApiClock so we're not coupled to full StorageOptions
instances.

Renamed Retrying.run(HttpStorageOptions, ...) to Retrying.runHttp(HttpStorageOptions, ...)
to remove ambiguity. Ideally we'll be able to remove this in the future once grpc
and http are both able to use the same approach for configuring gccl-invocation-id.

Refactor much of GrpcTransformPageDecoratorTest.java to provide a data driven
implementation of UnaryCallable removing the need to track individual request counts.
Rather than using String for RequestT and ResponseT I've created Req and Resp container
classes to be used and to remove any ambiguity about the directionality of various
operations.




Patch:
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index 086db5a0..5ef4cf8a 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -104,6 +104,10 @@
       <groupId>com.google.api.grpc</groupId>
       <artifactId>proto-google-cloud-storage-v2</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.google.api.grpc</groupId>
+      <artifactId>grpc-google-cloud-storage-v2</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.google.api.grpc</groupId>
       <artifactId>gapic-google-cloud-storage-v2</artifactId>
@@ -249,8 +253,9 @@
               grpc-netty-shaded is a runtime dep, we pull it up to our level to raise its transitive dependencies priority
               that overlap with our pubsub test deps.
               -->
-            <usedDependencies>io.grpc:grpc-netty-shaded,io.grpc:grpc-xds</usedDependencies>
+            <usedDependencies>io.grpc:grpc-xds</usedDependencies>
             <ignoredDependencies>
+              <dependency>io.grpc:grpc-netty-shaded</dependency>
               <!--
               We depend on `net.jqwik:jqwik` which itself has no classes, but depends on each of the other jars.
               In order to not churn on which of the sub apis we happen to be using at this time, we're flagging
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryConversions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryConversions.java
index b330fad9..37505948 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryConversions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryConversions.java
@@ -649,7 +649,7 @@ final class ApiaryConversions {
     to.setId(from.getId());
     to.setProjectId(from.getProjectId());
     ifNonNull(from.getServiceAccount(), ServiceAccount::getEmail, to::setServiceAccountEmail);
-    ifNonNull(from.getState(), Object::toString, to::setState);
+    ifNonNull(from.getState(), Enum::name, to::setState);
     ifNonNull(from.getCreateTimeOffsetDateTime(), dateTimeCodec::encode, to::setTimeCreated);
     ifNonNull(from.getUpdateTimeOffsetDateTime(), dateTimeCodec::encode, to::setUpdated);
     return to;
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java
index 4062b507..3222bc4e 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java
@@ -49,6 +49,10 @@ abstract class Crc32cValue<Res extends Crc32cValue<Res>> {
 
   public abstract String debugString();
 
+  public boolean eqValue(Crc32cValue<?> other) {
+    return this.getValue() == other.getValue();
+  }
+
   static Crc32cLengthUnknown of(int value) {
     return new Crc32cLengthUnknown(value);
   }
@@ -57,12 +61,8 @@ abstract class Crc32cValue<Res extends Crc32cValue<Res>> {
     return new Crc32cLengthKnown(value, length);
   }
 
-  public static <Res extends Crc32cValue<Res>> Res nullSafeConcat(Res r1, Crc32cLengthKnown r2) {
-    if (r1 == null) {
-      return null;
-    } else {
-      return r1.concat(r2);
-    }
+  static String fmtCrc32cValue(int value1) {
+    return String.format("crc32c{0x%08x}", value1);
   }
 
   static final class Crc32cLengthUnknown extends Crc32cValue<Crc32cLengthUnknown> {
@@ -84,7 +84,7 @@ abstract class Crc32cValue<Res extends Crc32cValue<Res>> {
 
     @Override
     public String toString() {
-      return String.format("crc32c{0x%08x}", value);
+      return Crc32cLengthKnown.fmtCrc32cValue(value);
     }
 
     @Override
@@ -128,7 +128,7 @@ abstract class Crc32cValue<Res extends Crc32cValue<Res>> {
 
     @Override
     public String debugString() {
-      return String.format("crc32c{0x%08x}", value);
+      return fmtCrc32cValue(value);
     }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicDownloadSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicDownloadSessionBuilder.java
index 6e7ea548..8177e4ea 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicDownloadSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicDownloadSessionBuilder.java
@@ -37,6 +37,10 @@ final class GapicDownloadSessionBuilder {
     return new GapicDownloadSessionBuilder();
   }
 
+  /**
+   * Any retry capability must be defined within the provided ServerStreamingCallable. The
+   * ultimately produced channel will not do any retries of its own.
+   */
   public ReadableByteChannelSessionBuilder byteChannel(
       ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read) {
     return new ReadableByteChannelSessionBuilder(read);
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 91031c08..c1079166 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
@@ -16,17 +16,17 @@
 
 package com.google.cloud.storage;
 
-import static com.google.cloud.storage.StorageV2ProtoUtils.seekReadObjectRequest;
 import static com.google.common.base.Preconditions.checkArgument;
 
 import com.google.api.client.http.HttpStatusCodes;
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.rpc.ServerStream;
 import com.google.api.gax.rpc.ServerStreamingCallable;
-import com.google.cloud.storage.Crc32cValue.Crc32cLengthUnknown;
+import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.Object;
+import com.google.storage.v2.ObjectChecksums;
 import com.google.storage.v2.ReadObjectRequest;
 import com.google.storage.v2.ReadObjectResponse;
 import java.io.Closeable;
@@ -49,11 +49,12 @@ final class GapicUnbufferedReadableByteChannel
 
   private boolean open = true;
   private boolean complete = false;
-  private long blobOffset = 0;
-  private long blobLimit = Long.MAX_VALUE;
-  private long totalSize = Long.MAX_VALUE; // initial sentinel value, will be updated upon responses
+  private boolean ioExceptionAlreadyThrown = false;
+  private long blobOffset;
+  private Crc32cLengthKnown cumulativeCrc32c;
 
   private Object metadata;
+  private ObjectChecksums objectChecksums;
 
   private ByteBuffer leftovers;
 
@@ -66,6 +67,7 @@ final class GapicUnbufferedReadableByteChannel
     this.read = read;
     this.req = req;
     this.hasher = hasher;
+    this.blobOffset = req.getReadOffset();
     this.iter = new LazyServerStreamIterator();
   }
 
@@ -90,9 +92,7 @@ final class GapicUnbufferedReadableByteChannel
     }
 
     long totalBufferCapacity = Arrays.stream(dsts).mapToLong(Buffer::remaining).sum();
-    long toRead = Math.min(blobLimit, totalBufferCapacity);
-
-    ReadCursor c = new ReadCursor(blobOffset, blobOffset + toRead);
+    ReadCursor c = new ReadCursor(blobOffset, blobOffset + totalBufferCapacity);
     while (c.hasRemaining()) {
       if (leftovers != null) {
         copy(c, leftovers, dsts, offset, length);
@@ -108,25 +108,38 @@ final class GapicUnbufferedReadableByteChannel
           Object respMetadata = resp.getMetadata();
           if (metadata == null) {
             metadata = respMetadata;
-          } else if (metadata.getMetageneration() != respMetadata.getMetageneration()) {
+          } else if (metadata.getGeneration() != respMetadata.getGeneration()) {
             throw closeWithError(
                 String.format(
-                    "Mismatch Metageneration between subsequent reads. Expected %d but received %d",
-                    metadata.getMetageneration(), respMetadata.getMetageneration()));
+                    "Mismatch Generation between subsequent reads. Expected %d but received %d",
+                    metadata.getGeneration(), respMetadata.getGeneration()));
           }
 
           if (!result.isDone()) {
             result.set(metadata);
           }
-
-          totalSize = metadata.getSize();
+        }
+        if (resp.hasObjectChecksums()) {
+          ObjectChecksums checksums = resp.getObjectChecksums();
+          if (this.objectChecksums == null) {
+            this.objectChecksums = checksums;
+          } else if (!objectChecksums.equals(checksums)) {
+            throw closeWithError(
+                String.format(
+                    "Mismatch checksums between subsequent reads. Expected %s but received %s",
+                    Crc32cValue.fmtCrc32cValue(objectChecksums.getCrc32C()),
+                    Crc32cValue.fmtCrc32cValue(checksums.getCrc32C())));
+          }
         }
         ChecksummedData checksummedData = resp.getChecksummedData();
         ByteBuffer content = checksummedData.getContent().asReadOnlyByteBuffer();
-        Crc32cLengthUnknown expected = Crc32cValue.of(checksummedData.getCrc32C());
+        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;
         }
@@ -153,6 +166,18 @@ final class GapicUnbufferedReadableByteChannel
 
   @Override
   public void close() throws IOException {
+    if (!ioExceptionAlreadyThrown
+        && cumulativeCrc32c != null
+        && objectChecksums != null
+        && objectChecksums.hasCrc32C()) {
+      Crc32cLengthKnown expected = Crc32cValue.of(objectChecksums.getCrc32C(), metadata.getSize());
+      if (!expected.eqValue(cumulativeCrc32c)) {
+        throw new IOException(
+            String.format(
+                "Mismatch checksum value. Expected %s actual %s",
+                expected.debugString(), cumulativeCrc32c.debugString()));
+      }
+    }
     open = false;
     iter.close();
   }
@@ -163,6 +188,7 @@ final class GapicUnbufferedReadableByteChannel
   }
 
   private IOException closeWithError(String message) throws IOException {
+    ioExceptionAlreadyThrown = true;
     close();
     StorageException cause =
         new StorageException(HttpStatusCodes.STATUS_CODE_PRECONDITION_FAILED, message);
@@ -200,7 +226,7 @@ final class GapicUnbufferedReadableByteChannel
 
     @Override
     public String toString() {
-      return String.format("Cursor{begin=%d, offset=%d, limit=%d}", beginning, offset, limit);
+      return String.format("ReadCursor{begin=%d, offset=%d, limit=%d}", beginning, offset, limit);
     }
   }
 
@@ -236,8 +262,7 @@ final class GapicUnbufferedReadableByteChannel
         synchronized (this) {
           if (!streamInitialized) {
             if (serverStream == null) {
-              ReadObjectRequest request = seekReadObjectRequest(req, blobOffset, blobLimit);
-              serverStream = read.call(request);
+              serverStream = read.call(req);
             }
             responseIterator = serverStream.iterator();
             streamInitialized = true;
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 c88a7221..afb6fa65 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
@@ -98,7 +98,7 @@ final class GapicUnbufferedWritableByteChannel<
       long offset = writeCtx.getTotalSentBytes().getAndAdd(contentSize);
       ChecksummedData.Builder checksummedData = ChecksummedData.newBuilder().setContent(b);
       if (crc32c != null) {
-        writeCtx.getCumulativeCrc32c().getAndAccumulate(crc32c, Crc32cValue::nullSafeConcat);
+        writeCtx.getCumulativeCrc32c().getAndAccumulate(crc32c, hasher::nullSafeConcat);
         checksummedData.setCrc32C(crc32c.getValue());
       }
       WriteObjectRequest.Builder builder =
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 9774a076..78f01b47 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
@@ -59,7 +59,8 @@ final class GrpcBlobWriteChannel implements WriteChannel {
 
   @Override
   public void setChunkSize(int chunkSize) {
-    Preconditions.checkState(!isOpen(), "Unable to change chunkSize after write");
+    Preconditions.checkState(
+        !lazyWriteChannel.isOpened(), "Unable to change chunkSize after write");
     this.chunkSize = chunkSize;
   }
 
@@ -75,7 +76,7 @@ final class GrpcBlobWriteChannel implements WriteChannel {
 
   @Override
   public boolean isOpen() {
-    return lazyWriteChannel.getChannel().isOpen();
+    return lazyWriteChannel.isOpened() && lazyWriteChannel.getChannel().isOpen();
   }
 
   @Override
@@ -93,14 +94,25 @@ final class GrpcBlobWriteChannel implements WriteChannel {
     private final Supplier<BufferedWritableByteChannelSession<WriteObjectResponse>> session;
     private final Supplier<BufferedWritableByteChannel> channel;
 
+    private boolean opened = false;
+
     public LazyWriteChannel(
         Supplier<BufferedWritableByteChannelSession<WriteObjectResponse>> session) {
       this.session = session;
-      this.channel = Suppliers.memoize(() -> session.get().open());
+      this.channel =
+          Suppliers.memoize(
+              () -> {
+                opened = true;
+                return session.get().open();
+              });
     }
 
     public BufferedWritableByteChannel getChannel() {
       return channel.get();
     }
+
+    public boolean isOpened() {
+      return opened;
+    }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
index 37987134..df0c4124 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
@@ -20,6 +20,7 @@ import static com.google.cloud.storage.Utils.bucketNameCodec;
 import static com.google.cloud.storage.Utils.durationMillisCodec;
 import static com.google.cloud.storage.Utils.ifNonNull;
 import static com.google.cloud.storage.Utils.lift;
+import static com.google.cloud.storage.Utils.projectNameCodec;
 import static com.google.cloud.storage.Utils.toImmutableListOf;
 import static com.google.cloud.storage.Utils.todo;
 
@@ -678,12 +679,12 @@ final class GrpcConversions {
 
   private HmacKeyMetadata hmacKeyMetadataEncode(HmacKey.HmacKeyMetadata from) {
     HmacKeyMetadata.Builder to = HmacKeyMetadata.newBuilder();
-    to.setAccessId(from.getAccessId());
-    // TODO etag
-    to.setId(from.getId());
-    to.setProject(from.getProjectId());
+    ifNonNull(from.getEtag(), to::setEtag);
+    ifNonNull(from.getId(), to::setId);
+    ifNonNull(from.getAccessId(), to::setAccessId);
+    ifNonNull(from.getProjectId(), projectNameCodec::encode, to::setProject);
     ifNonNull(from.getServiceAccount(), ServiceAccount::getEmail, to::setServiceAccountEmail);
-    ifNonNull(from.getState(), java.lang.Object::toString, to::setState);
+    ifNonNull(from.getState(), Enum::name, to::setState);
     ifNonNull(from.getCreateTimeOffsetDateTime(), timestampCodec::encode, to::setCreateTime);
     ifNonNull(from.getUpdateTimeOffsetDateTime(), timestampCodec::encode, to::setUpdateTime);
     return to.build();
@@ -694,7 +695,7 @@ final class GrpcConversions {
         .setAccessId(from.getAccessId())
         .setCreateTimeOffsetDateTime(timestampCodec.decode(from.getCreateTime()))
         .setId(from.getId())
-        .setProjectId(from.getProject())
+        .setProjectId(projectNameCodec.decode(from.getProject()))
         .setState(HmacKey.HmacKeyState.valueOf(from.getState()))
         .setUpdateTimeOffsetDateTime(timestampCodec.decode(from.getUpdateTime()))
         .build();
@@ -726,11 +727,11 @@ final class GrpcConversions {
   }
 
   private Object blobIdEncode(BlobId from) {
-    return Object.newBuilder()
-        .setName(from.getName())
-        .setBucket(from.getBucket())
-        .setGeneration(from.getGeneration())
-        .build();
+    Object.Builder to = Object.newBuilder();
+    ifNonNull(from.getBucket(), bucketNameCodec::encode, to::setBucket);
+    ifNonNull(from.getName(), to::setName);
+    ifNonNull(from.getGeneration(), to::setGeneration);
+    return to.build();
   }
 
   private BlobId blobIdDecode(Object from) {
@@ -738,11 +739,10 @@ final class GrpcConversions {
   }
 
   private Object blobInfoEncode(BlobInfo from) {
-    Object.Builder toBuilder =
-        Object.newBuilder()
-            .setBucket(from.getBucket())
-            .setName(from.getName())
-            .setGeneration(from.getGeneration());
+    Object.Builder toBuilder = Object.newBuilder();
+    ifNonNull(from.getBucket(), bucketNameCodec::encode, toBuilder::setBucket);
+    ifNonNull(from.getName(), toBuilder::setName);
+    ifNonNull(from.getGeneration(), toBuilder::setGeneration);
     ifNonNull(from.getCacheControl(), toBuilder::setCacheControl);
     ifNonNull(from.getSize(), toBuilder::setSize);
     ifNonNull(from.getContentType(), toBuilder::setContentType);
@@ -761,7 +761,7 @@ final class GrpcConversions {
       }
       toBuilder.setChecksums(objectChecksums.build());
     }
-    toBuilder.setMetageneration(from.getMetageneration());
+    ifNonNull(from.getMetageneration(), toBuilder::setMetageneration);
     ifNonNull(from.getDeleteTimeOffsetDateTime(), timestampCodec::encode, toBuilder::setDeleteTime);
     ifNonNull(from.getUpdateTimeOffsetDateTime(), timestampCodec::encode, toBuilder::setUpdateTime);
     ifNonNull(from.getCreateTimeOffsetDateTime(), timestampCodec::encode, toBuilder::setCreateTime);
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 16b5a5af..53a28e17 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
@@ -25,10 +25,14 @@ import static java.util.Objects.requireNonNull;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.gax.grpc.GrpcStatusCode;
 import com.google.api.gax.paging.AbstractPage;
 import com.google.api.gax.paging.Page;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.rpc.ApiException;
+import com.google.api.gax.rpc.ApiExceptionFactory;
 import com.google.api.gax.rpc.ApiExceptions;
+import com.google.api.gax.rpc.StatusCode;
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.BaseService;
 import com.google.cloud.Policy;
@@ -46,6 +50,7 @@ import com.google.cloud.storage.UnifiedOpts.BucketListOpt;
 import com.google.cloud.storage.UnifiedOpts.BucketSourceOpt;
 import com.google.cloud.storage.UnifiedOpts.BucketTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.HmacKeyListOpt;
+import com.google.cloud.storage.UnifiedOpts.HmacKeySourceOpt;
 import com.google.cloud.storage.UnifiedOpts.HmacKeyTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectListOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
@@ -53,16 +58,20 @@ import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.cloud.storage.UnifiedOpts.ProjectId;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.io.BaseEncoding;
 import com.google.common.io.ByteStreams;
+import com.google.protobuf.ByteString;
 import com.google.protobuf.FieldMask;
 import com.google.protobuf.Message;
 import com.google.storage.v2.ComposeObjectRequest;
 import com.google.storage.v2.ComposeObjectRequest.SourceObject;
 import com.google.storage.v2.CreateBucketRequest;
+import com.google.storage.v2.CreateHmacKeyRequest;
 import com.google.storage.v2.DeleteBucketRequest;
 import com.google.storage.v2.DeleteHmacKeyRequest;
 import com.google.storage.v2.DeleteObjectRequest;
 import com.google.storage.v2.GetBucketRequest;
+import com.google.storage.v2.GetHmacKeyRequest;
 import com.google.storage.v2.GetObjectRequest;
 import com.google.storage.v2.GetServiceAccountRequest;
 import com.google.storage.v2.ListBucketsRequest;
@@ -71,6 +80,7 @@ import com.google.storage.v2.ListObjectsRequest;
 import com.google.storage.v2.Object;
 import com.google.storage.v2.ProjectName;
 import com.google.storage.v2.ReadObjectRequest;
+import com.google.storage.v2.StorageClient;
 import com.google.storage.v2.StorageClient.ListBucketsPage;
 import com.google.storage.v2.StorageClient.ListBucketsPagedResponse;
 import com.google.storage.v2.StorageClient.ListHmacKeysPage;
@@ -78,11 +88,12 @@ import com.google.storage.v2.StorageClient.ListHmacKeysPagedResponse;
 import com.google.storage.v2.StorageClient.ListObjectsPage;
 import com.google.storage.v2.StorageClient.ListObjectsPagedResponse;
 import com.google.storage.v2.UpdateBucketRequest;
+import com.google.storage.v2.UpdateHmacKeyRequest;
 import com.google.storage.v2.UpdateObjectRequest;
 import com.google.storage.v2.WriteObjectRequest;
 import com.google.storage.v2.WriteObjectResponse;
 import com.google.storage.v2.WriteObjectSpec;
-import com.google.storage.v2.stub.GrpcStorageStub;
+import io.grpc.Status.Code;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -99,11 +110,19 @@ import java.nio.file.OpenOption;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.Arrays;
-import java.util.Iterator;
 import java.util.List;
+import java.util.Objects;
 import java.util.Set;
+import java.util.Spliterator;
+import java.util.Spliterators.AbstractSpliterator;
+import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.function.UnaryOperator;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
 
 final class GrpcStorageImpl extends BaseService<StorageOptions> implements Storage {
 
@@ -114,19 +133,33 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
           StandardOpenOption.WRITE,
           StandardOpenOption.CREATE,
           StandardOpenOption.TRUNCATE_EXISTING);
+  /**
+   * For use in {@link #resultRetryAlgorithmToCodes(ResultRetryAlgorithm)}. Resolve all codes and
+   * construct corresponding ApiExceptions.
+   *
+   * <p>Constructing the exceptions will walk the stack for each one. In order to avoid the stack
+   * walking overhead for every Code for every invocation of read, construct the set of exceptions
+   * only once and keep in this value.
+   */
+  private static final Set<ApiException> CODE_API_EXCEPTIONS =
+      Arrays.stream(StatusCode.Code.values())
+          .map(GrpcStorageImpl::statusCodeFor)
+          .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 GrpcStorageStub grpcStorageStub;
+  private final StorageClient storageClient;
   private final GrpcConversions codecs;
   private final GrpcRetryAlgorithmManager retryAlgorithmManager;
   private final SyntaxDecoders syntaxDecoders;
 
-  private final transient ProjectId defaultProjectId;
+  @Deprecated private final transient ProjectId defaultProjectId;
 
-  GrpcStorageImpl(GrpcStorageOptions options, GrpcStorageStub grpcStorageStub) {
+  GrpcStorageImpl(GrpcStorageOptions options, StorageClient storageClient) {
     super(options);
-    this.grpcStorageStub = grpcStorageStub;
+    this.storageClient = storageClient;
     this.codecs = Conversions.grpc();
     this.retryAlgorithmManager = options.getRetryAlgorithmManager();
     this.syntaxDecoders = new SyntaxDecoders();
@@ -135,7 +168,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public void close() throws Exception {
-    try (GrpcStorageStub s = grpcStorageStub) {
+    try (StorageClient s = storageClient) {
       s.shutdownNow();
     }
   }
@@ -155,7 +188,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> grpcStorageStub.createBucketCallable().call(req, grpcCallContext),
+        () -> storageClient.createBucketCallable().call(req, grpcCallContext),
         syntaxDecoders.bucket);
   }
 
@@ -184,7 +217,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
           ResumableMedia.gapic()
               .write()
               .byteChannel(
-                  grpcStorageStub.writeObjectCallable().withDefaultCallContext(grpcCallContext))
+                  storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
               .setByteStringStrategy(ByteStringStrategy.noCopy())
               .setHasher(Hasher.enabled())
               .direct()
@@ -233,7 +266,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         ResumableMedia.gapic()
             .write()
             .byteChannel(
-                grpcStorageStub.writeObjectCallable().withDefaultCallContext(grpcCallContext))
+                storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
             .setHasher(Hasher.enabled())
             .setByteStringStrategy(ByteStringStrategy.noCopy());
 
@@ -248,7 +281,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
           ResumableMedia.gapic()
               .write()
               .resumableWrite(
-                  grpcStorageStub
+                  storageClient
                       .startResumableWriteCallable()
                       .withDefaultCallContext(grpcCallContext),
                   req);
@@ -290,16 +323,14 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         ResumableMedia.gapic()
             .write()
             .resumableWrite(
-                grpcStorageStub
-                    .startResumableWriteCallable()
-                    .withDefaultCallContext(grpcCallContext),
+                storageClient.startResumableWriteCallable().withDefaultCallContext(grpcCallContext),
                 req);
 
     BufferedWritableByteChannelSession<WriteObjectResponse> session =
         ResumableMedia.gapic()
             .write()
             .byteChannel(
-                grpcStorageStub.writeObjectCallable().withDefaultCallContext(grpcCallContext))
+                storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
             .setHasher(Hasher.enabled())
             .setByteStringStrategy(ByteStringStrategy.noCopy())
             .resumable()
@@ -331,7 +362,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> grpcStorageStub.getBucketCallable().call(req, grpcCallContext),
+        () -> storageClient.getBucketCallable().call(req, grpcCallContext),
         syntaxDecoders.bucket);
   }
 
@@ -356,7 +387,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> grpcStorageStub.getObjectCallable().call(req, grpcCallContext),
+        () -> storageClient.getObjectCallable().call(req, grpcCallContext),
         syntaxDecoders.blob);
   }
 
@@ -368,22 +399,26 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   @Override
   public Page<Bucket> list(BucketListOption... options) {
     UnaryCallable<ListBucketsRequest, ListBucketsPagedResponse> listBucketsCallable =
-        grpcStorageStub.listBucketsPagedCallable();
+        storageClient.listBucketsPagedCallable();
     Opts<BucketListOpt> opts = Opts.unwrap(options);
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
-    ListBucketsRequest.Builder builder =
-        ListBucketsRequest.newBuilder().setParent(projectNameCodec.encode(defaultProjectId.val));
-    ListBucketsRequest request = opts.listBucketsRequest().apply(builder).build();
+    ListBucketsRequest request =
+        defaultProjectId
+            .listBuckets()
+            .andThen(opts.listBucketsRequest())
+            .apply(ListBucketsRequest.newBuilder())
+            .build();
     ListBucketsPagedResponse call = listBucketsCallable.call(request, grpcCallContext);
     ListBucketsPage page = call.getPage();
-    return new TransformingPageDecorator<>(page, syntaxDecoders.bucket);
+    return new TransformingPageDecorator<>(
+        page, syntaxDecoders.bucket, getOptions(), retryAlgorithmManager.getFor(request));
   }
 
   @Override
   public Page<Blob> list(String bucket, BlobListOption... options) {
     UnaryCallable<ListObjectsRequest, ListObjectsPagedResponse> listObjectsCallable =
-        grpcStorageStub.listObjectsPagedCallable();
+        storageClient.listObjectsPagedCallable();
     Opts<ObjectListOpt> opts = Opts.unwrap(options);
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
@@ -392,7 +427,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     ListObjectsRequest req = opts.listObjectsRequest().apply(builder).build();
     ListObjectsPagedResponse call = listObjectsCallable.call(req, grpcCallContext);
     ListObjectsPage page = call.getPage();
-    return new TransformingPageDecorator<>(page, syntaxDecoders.blob);
+    return new TransformingPageDecorator<>(
+        page, syntaxDecoders.blob, getOptions(), retryAlgorithmManager.getFor(req));
   }
 
   @Override
@@ -411,7 +447,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> grpcStorageStub.updateBucketCallable().call(req, grpcCallContext),
+        () -> storageClient.updateBucketCallable().call(req, grpcCallContext),
         syntaxDecoders.bucket);
   }
 
@@ -430,7 +466,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> grpcStorageStub.updateObjectCallable().call(req, grpcCallContext),
+        () -> storageClient.updateObjectCallable().call(req, grpcCallContext),
         syntaxDecoders.blob);
   }
 
@@ -451,7 +487,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
       Retrying.run(
           getOptions(),
           retryAlgorithmManager.getFor(req),
-          () -> grpcStorageStub.deleteBucketCallable().call(req, grpcCallContext),
+          () -> storageClient.deleteBucketCallable().call(req, grpcCallContext),
           Decoder.identity());
       return true;
     } catch (ApiException e) {
@@ -478,7 +514,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
       Retrying.run(
           getOptions(),
           retryAlgorithmManager.getFor(req),
-          () -> grpcStorageStub.deleteObjectCallable().call(req, grpcCallContext),
+          () -> storageClient.deleteObjectCallable().call(req, grpcCallContext),
           Decoder.identity());
       return true;
     } catch (ApiException e) {
@@ -513,7 +549,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> grpcStorageStub.composeObjectCallable().call(req, grpcCallContext),
+        () -> storageClient.composeObjectCallable().call(req, grpcCallContext),
         syntaxDecoders.blob);
   }
 
@@ -531,10 +567,14 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) {
     Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob);
     ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts);
+    Set<StatusCode.Code> codes =
+        GrpcStorageImpl.resultRetryAlgorithmToCodes(
+            retryAlgorithmManager.getFor(readObjectRequest));
+    GrpcCallContext grpcCallContext = GrpcCallContext.createDefault().withRetryableCodes(codes);
     UnbufferedReadableByteChannelSession<Object> session =
         ResumableMedia.gapic()
             .read()
-            .byteChannel(grpcStorageStub.readObjectCallable())
+            .byteChannel(storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext))
             .unbuffered()
             .setReadObjectRequest(readObjectRequest)
             .build();
@@ -563,7 +603,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   public GrpcBlobReadChannel reader(BlobId blob, BlobSourceOption... options) {
     Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob);
     ReadObjectRequest request = getReadObjectRequest(blob, opts);
-    return new GrpcBlobReadChannel(grpcStorageStub.readObjectCallable(), request);
+    Set<StatusCode.Code> codes =
+        GrpcStorageImpl.resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(request));
+    GrpcCallContext grpcCallContext = GrpcCallContext.createDefault().withRetryableCodes(codes);
+    return new GrpcBlobReadChannel(
+        storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext), request);
   }
 
   @Override
@@ -574,10 +618,14 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     // TODO: handle StorageRpc.Option.RETURN_RAW_INPUT_STREAM impacts
 
     ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts);
+    Set<StatusCode.Code> codes =
+        GrpcStorageImpl.resultRetryAlgorithmToCodes(
+            retryAlgorithmManager.getFor(readObjectRequest));
+    GrpcCallContext grpcCallContext = GrpcCallContext.createDefault().withRetryableCodes(codes);
     UnbufferedReadableByteChannelSession<Object> session =
         ResumableMedia.gapic()
             .read()
-            .byteChannel(grpcStorageStub.readObjectCallable())
+            .byteChannel(storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext))
             .unbuffered()
             .setReadObjectRequest(readObjectRequest)
             .build();
@@ -598,10 +646,14 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     // TODO: handle StorageRpc.Option.RETURN_RAW_INPUT_STREAM impacts
 
     ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts);
+    Set<StatusCode.Code> codes =
+        GrpcStorageImpl.resultRetryAlgorithmToCodes(
+            retryAlgorithmManager.getFor(readObjectRequest));
+    GrpcCallContext grpcCallContext = GrpcCallContext.createDefault().withRetryableCodes(codes);
     UnbufferedReadableByteChannelSession<Object> session =
         ResumableMedia.gapic()
             .read()
-            .byteChannel(grpcStorageStub.readObjectCallable())
+            .byteChannel(storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext))
             .unbuffered()
             .setReadObjectRequest(readObjectRequest)
             .build();
@@ -621,12 +673,12 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
     return new GrpcBlobWriteChannel(
-        grpcStorageStub.writeObjectCallable(),
+        storageClient.writeObjectCallable(),
         () ->
             ResumableMedia.gapic()
                 .write()
                 .resumableWrite(
-                    grpcStorageStub
+                    storageClient
                         .startResumableWriteCallable()
                         .withDefaultCallContext(grpcCallContext),
                     req));
@@ -840,30 +892,66 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public HmacKey createHmacKey(ServiceAccount serviceAccount, CreateHmacKeyOption... options) {
-    return todo();
+    Opts<HmacKeyTargetOpt> opts = Opts.unwrap(options);
+    GrpcCallContext grpcCallContext =
+        opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
+    CreateHmacKeyRequest request =
+        defaultProjectId
+            .createHmacKey()
+            .andThen(opts.createHmacKeysRequest())
+            .apply(CreateHmacKeyRequest.newBuilder())
+            .setServiceAccountEmail(serviceAccount.getEmail())
+            .build();
+    return Retrying.run(
+        getOptions(),
+        retryAlgorithmManager.getFor(request),
+        () -> storageClient.createHmacKeyCallable().call(request, grpcCallContext),
+        resp -> {
+          ByteString secretKeyBytes = resp.getSecretKeyBytes();
+          String b64SecretKey = BaseEncoding.base64().encode(secretKeyBytes.toByteArray());
+          return HmacKey.newBuilder(b64SecretKey)
+              .setMetadata(codecs.hmacKeyMetadata().decode(resp.getMetadata()))
+              .build();
+        });
   }
 
   @Override
   public Page<HmacKeyMetadata> listHmacKeys(ListHmacKeysOption... options) {
     UnaryCallable<ListHmacKeysRequest, ListHmacKeysPagedResponse> listHmacKeysCallable =
-        grpcStorageStub.listHmacKeysPagedCallable();
+        storageClient.listHmacKeysPagedCallable();
     Opts<HmacKeyListOpt> opts = Opts.unwrap(options);
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
 
-    ProjectId projectId = opts.projectId().orElse(defaultProjectId);
-
-    ListHmacKeysRequest.Builder builder =
-        projectId.listHmacKeys().apply(ListHmacKeysRequest.newBuilder());
-    ListHmacKeysRequest request = opts.listHmacKeysRequest().apply(builder).build();
+    ListHmacKeysRequest request =
+        defaultProjectId
+            .listHmacKeys()
+            .andThen(opts.listHmacKeysRequest())
+            .apply(ListHmacKeysRequest.newBuilder())
+            .build();
     ListHmacKeysPagedResponse call = listHmacKeysCallable.call(request, grpcCallContext);
     ListHmacKeysPage page = call.getPage();
-    return new TransformingPageDecorator<>(page, codecs.hmacKeyMetadata());
+    return new TransformingPageDecorator<>(
+        page, codecs.hmacKeyMetadata(), getOptions(), retryAlgorithmManager.getFor(request));
   }
 
   @Override
   public HmacKeyMetadata getHmacKey(String accessId, GetHmacKeyOption... options) {
-    return todo();
+    Opts<HmacKeySourceOpt> opts = Opts.unwrap(options);
+    GrpcCallContext grpcCallContext =
+        opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
+    GetHmacKeyRequest request =
+        defaultProjectId
+            .getHmacKey()
+            .andThen(opts.getHmacKeysRequest())
+            .apply(GetHmacKeyRequest.newBuilder())
+            .setAccessId(accessId)
+            .build();
+    return Retrying.run(
+        getOptions(),
+        retryAlgorithmManager.getFor(request),
+        () -> storageClient.getHmacKeyCallable().call(request, grpcCallContext),
+        codecs.hmacKeyMetadata());
   }
 
   @Override
@@ -874,13 +962,13 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     DeleteHmacKeyRequest req =
         DeleteHmacKeyRequest.newBuilder()
             .setAccessId(hmacKeyMetadata.getAccessId())
-            .setProject(hmacKeyMetadata.getProjectId())
+            .setProject(projectNameCodec.encode(hmacKeyMetadata.getProjectId()))
             .build();
     Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
         () -> {
-          grpcStorageStub.deleteHmacKeyCallable().call(req, grpcCallContext);
+          storageClient.deleteHmacKeyCallable().call(req, grpcCallContext);
           return null;
         },
         Decoder.identity());
@@ -889,7 +977,21 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   @Override
   public HmacKeyMetadata updateHmacKeyState(
       HmacKeyMetadata hmacKeyMetadata, HmacKeyState state, UpdateHmacKeyOption... options) {
-    return todo();
+    Opts<HmacKeyTargetOpt> opts = Opts.unwrap(options);
+    GrpcCallContext grpcCallContext =
+        opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
+    com.google.storage.v2.HmacKeyMetadata encode =
+        codecs.hmacKeyMetadata().encode(hmacKeyMetadata).toBuilder().setState(state.name()).build();
+
+    UpdateHmacKeyRequest.Builder builder =
+        opts.updateHmacKeysRequest().apply(UpdateHmacKeyRequest.newBuilder()).setHmacKey(encode);
+    UpdateHmacKeyRequest request =
+        builder.setUpdateMask(FieldMask.newBuilder().addPaths("state").build()).build();
+    return Retrying.run(
+        getOptions(),
+        retryAlgorithmManager.getFor(request),
+        () -> storageClient.updateHmacKeyCallable().call(request, grpcCallContext),
+        codecs.hmacKeyMetadata());
   }
 
   @Override
@@ -915,7 +1017,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> grpcStorageStub.getServiceAccountCallable().call(req),
+        () -> storageClient.getServiceAccountCallable().call(req),
         codecs.serviceAccount());
   }
 
@@ -969,10 +1071,18 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
     private final PageT page;
     private final Decoder<ResourceT, ModelT> translator;
-
-    public TransformingPageDecorator(PageT page, Decoder<ResourceT, ModelT> translator) {
+    private final Retrying.RetryingDependencies deps;
+    private final ResultRetryAlgorithm<?> resultRetryAlgorithm;
+
+    TransformingPageDecorator(
+        PageT page,
+        Decoder<ResourceT, ModelT> translator,
+        Retrying.RetryingDependencies deps,
+        ResultRetryAlgorithm<?> resultRetryAlgorithm) {
       this.page = page;
       this.translator = translator;
+      this.deps = deps;
+      this.resultRetryAlgorithm = resultRetryAlgorithm;
     }
 
     @Override
@@ -987,43 +1097,84 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
     @Override
     public Page<ModelT> getNextPage() {
-      return new TransformingPageDecorator<>(page.getNextPage(), translator);
+      return new TransformingPageDecorator<>(
+          page.getNextPage(), translator, deps, resultRetryAlgorithm);
     }
 
+    @SuppressWarnings({"Convert2MethodRef"})
     @Override
     public Iterable<ModelT> iterateAll() {
-      return () -> {
-        final Iterator<ResourceT> iter = page.iterateAll().iterator();
-        return new TransformingIterator(iter);
-      };
+      // iterateAll on AbstractPage isn't very friendly to decoration, as getNextPage isn't actually
+      // ever called. This means we aren't able to apply our retry wrapping there.
+      // Instead, what we do is create a stream which will attempt to call getNextPage repeatedly
+      // until we meet some condition of exhaustion. At that point we can apply our retry logic.
+      return () ->
+          streamIterate(
+                  page,
+                  p -> p != null && p.hasNextPage(),
+                  prev -> {
+                    // explicitly define this callable rather than using the method reference to
+                    // prevent a javac 1.8 exception
+                    // https://bugs.java.com/bugdatabase/view_bug.do?bug_id=8056984
+                    Callable<PageT> c = () -> prev.getNextPage();
+                    return Retrying.run(deps, resultRetryAlgorithm, c, Decoder.identity());
+                  })
+              .filter(Objects::nonNull)
+              .flatMap(p -> StreamSupport.stream(p.getValues().spliterator(), false))
+              .map(translator::decode)
+              .iterator();
     }
 
     @Override
     public Iterable<ModelT> getValues() {
-      return () -> {
-        final Iterator<ResourceT> inter = page.getValues().iterator();
-        return new TransformingIterator(inter);
-      };
+      return () ->
+          StreamSupport.stream(page.getValues().spliterator(), false)
+              .map(translator::decode)
+              .iterator();
     }
 
-    private class TransformingIterator implements Iterator<ModelT> {
-
-      private final Iterator<ResourceT> iter;
-
-      public TransformingIterator(Iterator<ResourceT> iter) {
-        this.iter = iter;
-      }
-
-      @Override
-      public boolean hasNext() {
-        return iter.hasNext();
-      }
-
-      @Override
-      public ModelT next() {
-        ResourceT next = iter.next();
-        return translator.decode(next);
-      }
+    private static <T> Stream<T> streamIterate(
+        T seed, Predicate<? super T> shouldComputeNext, UnaryOperator<T> computeNext) {
+      requireNonNull(seed, "seed must be non null");
+      requireNonNull(shouldComputeNext, "shouldComputeNext must be non null");
+      requireNonNull(computeNext, "computeNext must be non null");
+      Spliterator<T> spliterator =
+          new AbstractSpliterator<T>(Long.MAX_VALUE, 0) {
+            T prev;
+            boolean started = false;
+            boolean done = false;
+
+            @Override
+            public boolean tryAdvance(Consumer<? super T> action) {
+              // if we haven't started, emit our seed and return
+              if (!started) {
+                started = true;
+                action.accept(seed);
+                prev = seed;
+                return true;
+              }
+              // if we've previously finished quickly return
+              if (done) {
+                return false;
+              }
+              // test whether we should try and compute the next value
+              if (shouldComputeNext.test(prev)) {
+                // compute the next value and figure out if we can use it
+                T next = computeNext.apply(prev);
+                if (next != null) {
+                  action.accept(next);
+                  prev = next;
+                  return true;
+                }
+              }
+
+              // fallthrough, if we haven't taken an action by now consider the stream done and
+              // return
+              done = true;
+              return false;
+            }
+          };
+      return StreamSupport.stream(spliterator, false);
     }
   }
 
@@ -1085,4 +1236,58 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
                 .collect(Collectors.toList()))
         .build();
   }
+
+  /**
+   * When using the retry features of the Gapic client, we are only allowed to provide a {@link
+   * Set}{@code <}{@link StatusCode.Code}{@code >}. Given {@link StatusCode.Code} is an enum, we can
+   * resolve the set of values from a given {@link ResultRetryAlgorithm} by evaluating each one as
+   * an {@link ApiException}.
+   */
+  static Set<StatusCode.Code> resultRetryAlgorithmToCodes(ResultRetryAlgorithm<?> alg) {
+    return CODE_API_EXCEPTIONS.stream()
+        .filter(e -> alg.shouldRetry(e, null))
+        .map(e -> e.getStatusCode().getCode())
+        .collect(Collectors.toSet());
+  }
+
+  private static GrpcStatusCode statusCodeFor(StatusCode.Code code) {
+    switch (code) {
+      case OK:
+        return GrpcStatusCode.of(Code.OK);
+      case CANCELLED:
+        return GrpcStatusCode.of(Code.CANCELLED);
+      case UNKNOWN:
+        return GrpcStatusCode.of(Code.UNKNOWN);
+      case INVALID_ARGUMENT:
+        return GrpcStatusCode.of(Code.INVALID_ARGUMENT);
+      case DEADLINE_EXCEEDED:
+        return GrpcStatusCode.of(Code.DEADLINE_EXCEEDED);
+      case NOT_FOUND:
+        return GrpcStatusCode.of(Code.NOT_FOUND);
+      case ALREADY_EXISTS:
+        return GrpcStatusCode.of(Code.ALREADY_EXISTS);
+      case PERMISSION_DENIED:
+        return GrpcStatusCode.of(Code.PERMISSION_DENIED);
+      case RESOURCE_EXHAUSTED:
+        return GrpcStatusCode.of(Code.RESOURCE_EXHAUSTED);
+      case FAILED_PRECONDITION:
+        return GrpcStatusCode.of(Code.FAILED_PRECONDITION);
+      case ABORTED:
+        return GrpcStatusCode.of(Code.ABORTED);
+      case OUT_OF_RANGE:
+        return GrpcStatusCode.of(Code.OUT_OF_RANGE);
+      case UNIMPLEMENTED:
+        return GrpcStatusCode.of(Code.UNIMPLEMENTED);
+      case INTERNAL:
+        return GrpcStatusCode.of(Code.INTERNAL);
+      case UNAVAILABLE:
+        return GrpcStatusCode.of(Code.UNAVAILABLE);
+      case DATA_LOSS:
+        return GrpcStatusCode.of(Code.DATA_LOSS);
+      case UNAUTHENTICATED:
+        return GrpcStatusCode.of(Code.UNAUTHENTICATED);
+      default:
+        throw new IllegalStateException("Unrecognized status code: " + code);
+    }
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
index 8ceef950..875231e3 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
@@ -21,9 +21,11 @@ import static java.util.Objects.requireNonNull;
 import com.google.api.core.ApiClock;
 import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
+import com.google.api.gax.core.CredentialsProvider;
 import com.google.api.gax.core.FixedCredentialsProvider;
+import com.google.api.gax.core.NoCredentialsProvider;
 import com.google.api.gax.retrying.RetrySettings;
-import com.google.api.gax.rpc.ClientContext;
+import com.google.api.gax.retrying.StreamResumptionStrategy;
 import com.google.api.gax.rpc.HeaderProvider;
 import com.google.auth.Credentials;
 import com.google.cloud.NoCredentials;
@@ -36,15 +38,21 @@ import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.spi.StorageRpcFactory;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableSet;
+import com.google.storage.v2.ReadObjectRequest;
+import com.google.storage.v2.ReadObjectResponse;
+import com.google.storage.v2.StorageClient;
 import com.google.storage.v2.StorageSettings;
-import com.google.storage.v2.stub.GrpcStorageStub;
+import io.grpc.ManagedChannelBuilder;
 import java.io.IOException;
 import java.net.URI;
 import java.util.Set;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 @BetaApi
 @TransportCompatibility(Transport.GRPC)
-public final class GrpcStorageOptions extends StorageOptions {
+public final class GrpcStorageOptions extends StorageOptions
+    implements Retrying.RetryingDependencies {
 
   private static final long serialVersionUID = 4165732727259088956L;
   private static final String GCS_SCOPE = "https://www.googleapis.com/auth/devstorage.full_control";
@@ -54,12 +62,12 @@ public final class GrpcStorageOptions extends StorageOptions {
   private final GrpcRetryAlgorithmManager retryAlgorithmManager;
 
   @BetaApi
-  public GrpcStorageOptions(Builder builder, StorageDefaults serviceDefaults) {
+  public GrpcStorageOptions(Builder builder, GrpcStorageDefaults serviceDefaults) {
     super(builder, serviceDefaults);
     this.retryAlgorithmManager =
         new GrpcRetryAlgorithmManager(
             MoreObjects.firstNonNull(
-                builder.storageRetryStrategy, defaults().getStorageRetryStrategy()));
+                builder.storageRetryStrategy, serviceDefaults.getStorageRetryStrategy()));
   }
 
   @Override
@@ -67,10 +75,60 @@ public final class GrpcStorageOptions extends StorageOptions {
     return SCOPES;
   }
 
+  @InternalApi
   GrpcRetryAlgorithmManager getRetryAlgorithmManager() {
     return retryAlgorithmManager;
   }
 
+  @InternalApi
+  StorageSettings getStorageSettings() throws IOException {
+    URI uri = URI.create(getHost());
+    String scheme = uri.getScheme();
+    int port = uri.getPort() > 0 ? uri.getPort() : scheme.equals("http") ? 80 : 443;
+    String endpoint = String.format("%s:%d", uri.getHost(), port);
+
+    CredentialsProvider credentialsProvider;
+    if (credentials instanceof NoCredentials) {
+      credentialsProvider = NoCredentialsProvider.create();
+    } else {
+      credentialsProvider = FixedCredentialsProvider.create(credentials);
+    }
+
+    StorageSettings.Builder builder =
+        StorageSettings.newBuilder()
+            .setEndpoint(endpoint)
+            .setCredentialsProvider(credentialsProvider);
+
+    if (scheme.equals("http")) {
+      builder.setTransportChannelProvider(
+          com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.newBuilder()
+              .setEndpoint(endpoint)
+              .setChannelConfigurator(ManagedChannelBuilder::usePlaintext)
+              .build());
+    }
+    RetrySettings retrySettings = getRetrySettings();
+    RetrySettings attemptOnce = retrySettings.toBuilder().setMaxAttempts(1).build();
+    // all retries for unary methods are handled at a different level
+    builder.applyToAllUnaryMethods(
+        input -> {
+          input.setRetrySettings(attemptOnce);
+          return null;
+        });
+    // for ReadObject we are configuring the server stream handling to do its own retries, so wire
+    // things through. Retryable codes will be controlled closer to the use site as idempotency
+    // considerations need to be made.
+    builder
+        .readObjectSettings()
+        .setRetrySettings(retrySettings)
+        // even though we might want to default to the empty set for retryable codes, don't ever
+        // actually do this. Doing so prevents any retry capability from being wired into the stream
+        // pipeline, ever.
+        // For our use, we will always set it one way or the other to ensure it's appropriate
+        // DO NOT: .setRetryableCodes(Collections.emptySet())
+        .setResumptionStrategy(new ReadObjectResumptionStrategy());
+    return builder.build();
+  }
+
   @BetaApi
   @Override
   public GrpcStorageOptions.Builder toBuilder() {
@@ -290,41 +348,8 @@ public final class GrpcStorageOptions extends StorageOptions {
       if (options instanceof GrpcStorageOptions) {
         GrpcStorageOptions grpcStorageOptions = (GrpcStorageOptions) options;
         try {
-          URI uri = URI.create(options.getHost());
-          String h = uri.getHost();
-          int p = uri.getPort() > 0 ? uri.getPort() : uri.getScheme().equals("http") ? 80 : 443;
-          String hp = String.format("%s:%d", h, p);
-          StorageSettings.Builder builder = StorageSettings.newBuilder().setEndpoint(hp);
-          builder.setCredentialsProvider(FixedCredentialsProvider.create(options.getCredentials()));
-          if (!"storage.googleapis.com:443".equals(hp)) { // TODO: make this more formal
-            options =
-                options
-                    .toBuilder()
-                    .setProjectId(
-                        options.getProjectId() == null ? "local-project" : options.getProjectId())
-                    .setCredentials(NoCredentials.getInstance())
-                    .build();
-            builder
-                .setCredentialsProvider(com.google.api.gax.core.NoCredentialsProvider.create())
-                .setTransportChannelProvider(
-                    com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.newBuilder()
-                        .setEndpoint(hp)
-                        .setChannelConfigurator(c -> c.usePlaintext())
-                        .build());
-          }
-          final StorageOptions finalOpts = options;
-          StorageSettings settings =
-              builder
-                  .applyToAllUnaryMethods(
-                      input -> {
-                        input.setRetrySettings(finalOpts.getRetrySettings());
-                        return null;
-                      })
-                  .build();
-          ClientContext clientContext = ClientContext.create(settings);
-          GrpcStorageStub grpcStorageStub = GrpcStorageStub.create(clientContext);
-
-          return new GrpcStorageImpl(grpcStorageOptions, grpcStorageStub);
+          StorageSettings storageSettings = grpcStorageOptions.getStorageSettings();
+          return new GrpcStorageImpl(grpcStorageOptions, StorageClient.create(storageSettings));
         } catch (IOException e) {
           throw new IllegalStateException(
               "Unable to instantiate gRPC com.google.cloud.storage.Storage client.", e);
@@ -370,4 +395,38 @@ public final class GrpcStorageOptions extends StorageOptions {
       throw new IllegalStateException("No supported for grpc");
     }
   }
+
+  // TODO: See if we can change gax to allow shifting this to callable.withContext so it doesn't
+  //   have to be set globally
+  private static class ReadObjectResumptionStrategy
+      implements StreamResumptionStrategy<ReadObjectRequest, ReadObjectResponse> {
+    private long readOffset = 0;
+
+    @NonNull
+    @Override
+    public StreamResumptionStrategy<ReadObjectRequest, ReadObjectResponse> createNew() {
+      return new ReadObjectResumptionStrategy();
+    }
+
+    @NonNull
+    @Override
+    public ReadObjectResponse processResponse(ReadObjectResponse response) {
+      readOffset += response.getChecksummedData().getContent().size();
+      return response;
+    }
+
+    @Nullable
+    @Override
+    public ReadObjectRequest getResumeRequest(ReadObjectRequest originalRequest) {
+      if (readOffset != 0) {
+        return originalRequest.toBuilder().setReadOffset(readOffset).build();
+      }
+      return originalRequest;
+    }
+
+    @Override
+    public boolean canResume() {
+      return true;
+    }
+  }
 }
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 2c617abd..b89fe7c6 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
@@ -22,6 +22,7 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.function.Supplier;
 import javax.annotation.concurrent.Immutable;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 interface Hasher {
 
@@ -33,6 +34,9 @@ interface Hasher {
 
   void validate(Crc32cValue<?> expected, Supplier<ByteBuffer> b) throws IOException;
 
+  @Nullable
+  Crc32cLengthKnown nullSafeConcat(Crc32cLengthKnown r1, Crc32cLengthKnown r2);
+
   static Hasher noop() {
     return NoOpHasher.INSTANCE;
   }
@@ -54,6 +58,11 @@ interface Hasher {
 
     @Override
     public void validate(Crc32cValue<?> expected, Supplier<ByteBuffer> b) {}
+
+    @Override
+    public @Nullable Crc32cLengthKnown nullSafeConcat(Crc32cLengthKnown r1, Crc32cLengthKnown r2) {
+      return null;
+    }
   }
 
   @Immutable
@@ -71,9 +80,21 @@ interface Hasher {
     @Override
     public void validate(Crc32cValue<?> expected, Supplier<ByteBuffer> b) throws IOException {
       Crc32cLengthKnown actual = hash(b);
-      if (actual.getValue() != expected.getValue()) {
+      if (!actual.eqValue(expected)) {
         throw new IOException(
-            String.format("Mismatch checksum value. Expected %s actual %s", expected, actual));
+            String.format(
+                "Mismatch checksum value. Expected %s actual %s",
+                expected.debugString(), actual.debugString()));
+      }
+    }
+
+    @Override
+    @Nullable
+    public Crc32cLengthKnown nullSafeConcat(Crc32cLengthKnown r1, Crc32cLengthKnown r2) {
+      if (r1 == null) {
+        return r2;
+      } else {
+        return r1.concat(r2);
       }
     }
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java
index b1811a33..e529ee9f 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java
@@ -71,7 +71,7 @@ final class Retrying {
    *
    * @param <T> The result type of {@code c}
    * @param <U> The result type of any mapping that takes place via {@code f}
-   * @param options The {@link HttpStorageOptions} which {@link RetrySettings} and {@link ApiClock}
+   * @param deps The {@link RetryingDependencies} which {@link RetrySettings} and {@link ApiClock}
    *     will be resolved from.
    * @param algorithm The {@link ResultRetryAlgorithm} to use when determining if a retry is
    *     possible
@@ -83,15 +83,26 @@ final class Retrying {
    * @throws StorageException if {@code c} fails due to any retry exhaustion
    */
   static <T, U> U run(
-      GrpcStorageOptions options,
+      RetryingDependencies deps,
       ResultRetryAlgorithm<?> algorithm,
       Callable<T> c,
       Decoder<T, U> f) {
     try {
-      T result = runWithRetries(c, options.getRetrySettings(), algorithm, options.getClock());
+      T result = runWithRetries(c, deps.getRetrySettings(), algorithm, deps.getClock());
       return result == null ? null : f.decode(result);
     } catch (RetryHelperException e) {
       throw StorageException.coalesce(e.getCause());
     }
   }
+
+  /**
+   * Rather than requiring a full set of {@link StorageOptions} to be passed specify what we
+   * actually need and have StorageOptions implement this interface.
+   */
+  interface RetryingDependencies {
+
+    RetrySettings getRetrySettings();
+
+    ApiClock getClock();
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
index 32b2e42b..57fc397c 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
@@ -57,7 +57,6 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.Locale;
 import java.util.Objects;
-import java.util.Optional;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
@@ -1104,7 +1103,7 @@ final class UnifiedOpts {
    */
   @Deprecated
   static final class ProjectId extends RpcOptVal<String>
-      implements HmacKeySourceOpt, HmacKeyTargetOpt, HmacKeyListOpt {
+      implements HmacKeySourceOpt, HmacKeyTargetOpt, HmacKeyListOpt, BucketListOpt {
     private static final long serialVersionUID = 1471462503030451598L;
 
     private ProjectId(String val) {
@@ -1122,9 +1121,14 @@ final class UnifiedOpts {
     }
 
     @Override
-    public Mapper<DeleteHmacKeyRequest.Builder> deleteHmacKey() {
+    public Mapper<CreateHmacKeyRequest.Builder> createHmacKey() {
       return b -> b.setProject(projectNameCodec.encode(val));
     }
+
+    @Override
+    public Mapper<ListBucketsRequest.Builder> listBuckets() {
+      return b -> b.setParent(projectNameCodec.encode(val));
+    }
   }
 
   static final class Projection extends RpcOptVal<String> implements BucketTargetOpt {
@@ -1958,11 +1962,6 @@ final class UnifiedOpts {
       return fuseMappers(ObjectTargetOpt.class, ObjectTargetOpt::blobInfo);
     }
 
-    @Deprecated
-    Optional<ProjectId> projectId() {
-      return filterTo(ProjectId.class).findFirst();
-    }
-
     private Mapper<ImmutableMap.Builder<StorageRpc.Option, Object>> rpcOptionMapper() {
       return fuseMappers(RpcOptVal.class, RpcOptVal::mapper);
     }


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/FakeServer.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/FakeServer.java
new file mode 100644
index 00000000..506a013d
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/FakeServer.java
@@ -0,0 +1,60 @@
+/*
+ * 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 com.google.cloud.NoCredentials;
+import com.google.storage.v2.StorageGrpc;
+import com.google.storage.v2.StorageSettings;
+import io.grpc.Server;
+import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.concurrent.TimeUnit;
+
+final class FakeServer implements AutoCloseable {
+
+  private final Server server;
+  private final GrpcStorageOptions grpcStorageOptions;
+
+  FakeServer(Server server, GrpcStorageOptions grpcStorageOptions) {
+    this.server = server;
+    this.grpcStorageOptions = grpcStorageOptions;
+  }
+
+  StorageSettings storageSettings() throws IOException {
+    return grpcStorageOptions.getStorageSettings();
+  }
+
+  @Override
+  public void close() throws InterruptedException {
+    server.shutdownNow().awaitTermination(10, TimeUnit.SECONDS);
+  }
+
+  static FakeServer of(StorageGrpc.StorageImplBase service) throws IOException {
+    InetSocketAddress address = new InetSocketAddress("localhost", 0);
+    Server server = NettyServerBuilder.forAddress(address).addService(service).build();
+    server.start();
+    String endpoint = String.format("%s:%d", address.getHostString(), server.getPort());
+    GrpcStorageOptions grpcStorageOptions =
+        StorageOptions.grpc()
+            .setHost("http://" + endpoint)
+            .setProjectId("test-proj")
+            .setCredentials(NoCredentials.getInstance())
+            .build();
+    return new FakeServer(server, grpcStorageOptions);
+  }
+}
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
new file mode 100644
index 00000000..5907edf0
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannelTest.java
@@ -0,0 +1,383 @@
+/*
+ * 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 static org.junit.Assert.assertThrows;
+
+import com.google.api.core.ApiFutures;
+import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.gax.grpc.GrpcStatusCode;
+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.ChannelSession.UnbufferedReadSession;
+import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
+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;
+import com.google.storage.v2.ChecksummedData;
+import com.google.storage.v2.ContentRange;
+import com.google.storage.v2.Object;
+import com.google.storage.v2.ObjectChecksums;
+import com.google.storage.v2.ReadObjectRequest;
+import com.google.storage.v2.ReadObjectResponse;
+import com.google.storage.v2.StorageClient;
+import com.google.storage.v2.StorageGrpc;
+import io.grpc.Status.Code;
+import io.grpc.StatusRuntimeException;
+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 {
+  private final byte[] bytes = DataGenerator.base64Characters().genBytes(40);
+  private final ByteString data1 = ByteString.copyFrom(bytes, 0, 10);
+  private final ByteString data2 = ByteString.copyFrom(bytes, 10, 10);
+  private final ByteString data3 = ByteString.copyFrom(bytes, 20, 10);
+  private final ByteString data4 = ByteString.copyFrom(bytes, 30, 10);
+
+  private final String objectName = "name";
+  private final Object expectedResult =
+      Object.newBuilder()
+          .setName(objectName)
+          .setGeneration(3L)
+          .setContentType("application/octet-stream")
+          .build();
+
+  private final ReadObjectRequest req1 =
+      ReadObjectRequest.newBuilder().setObject(objectName).setReadOffset(0).build();
+  private final ReadObjectRequest req2 = req1.toBuilder().setReadOffset(20).build();
+  private final ReadObjectResponse resp1 =
+      ReadObjectResponse.newBuilder()
+          .setMetadata(expectedResult)
+          .setContentRange(ContentRange.newBuilder().setStart(0).build())
+          .setObjectChecksums(
+              ObjectChecksums.newBuilder().setCrc32C(Hashing.crc32c().hashBytes(bytes).asInt()))
+          .setChecksummedData(getChecksummedData(data1))
+          .build();
+  private final ReadObjectResponse resp2 =
+      ReadObjectResponse.newBuilder().setChecksummedData(getChecksummedData(data2)).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))
+          .build();
+  private final ReadObjectResponse resp4 =
+      ReadObjectResponse.newBuilder().setChecksummedData(getChecksummedData(data4)).build();
+
+  /** Define a Storage service that will always return an error during the first readObject */
+  private final StorageGrpc.StorageImplBase fakeStorage =
+      new StorageGrpc.StorageImplBase() {
+        @Override
+        public void readObject(
+            ReadObjectRequest request, StreamObserver<ReadObjectResponse> responseObserver) {
+          if (request.equals(req1)) {
+            responseObserver.onNext(resp1);
+            responseObserver.onNext(resp2);
+            responseObserver.onError(apiException(Code.DATA_LOSS));
+          } else if (request.equals(req2)) {
+            responseObserver.onNext(resp3);
+            responseObserver.onNext(resp4);
+            responseObserver.onCompleted();
+          } else {
+            responseObserver.onError(apiException(Code.PERMISSION_DENIED));
+          }
+        }
+      };
+
+  @Test
+  public void readRetriesAreProperlyOrdered_readLargerThanMessageSize()
+      throws IOException, ExecutionException, InterruptedException, TimeoutException {
+    try (FakeServer server = FakeServer.of(fakeStorage);
+        StorageClient storageClient = StorageClient.create(server.storageSettings())) {
+
+      UnbufferedReadableByteChannelSession<Object> session =
+          new UnbufferedReadSession<>(
+              ApiFutures.immediateFuture(req1),
+              (start, resultFuture) ->
+                  new GapicUnbufferedReadableByteChannel(
+                      resultFuture,
+                      storageClient
+                          .readObjectCallable()
+                          .withDefaultCallContext(
+                              contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
+                      start,
+                      Hasher.noop()));
+      byte[] actualBytes = new byte[40];
+      try (UnbufferedReadableByteChannel c = session.open()) {
+        c.read(ByteBuffer.wrap(actualBytes));
+      }
+      Object actualResult = session.getResult().get(1000, TimeUnit.MILLISECONDS);
+      assertThat(actualResult).isEqualTo(expectedResult);
+      assertThat(actualBytes).isEqualTo(bytes);
+    }
+  }
+
+  @Test
+  public void readRetriesAreProperlyOrdered_readSmallerThanMessageSize()
+      throws IOException, ExecutionException, InterruptedException, TimeoutException {
+    try (FakeServer server = FakeServer.of(fakeStorage);
+        StorageClient storageClient = StorageClient.create(server.storageSettings())) {
+
+      UnbufferedReadableByteChannelSession<Object> session =
+          new UnbufferedReadSession<>(
+              ApiFutures.immediateFuture(req1),
+              (start, resultFuture) ->
+                  new GapicUnbufferedReadableByteChannel(
+                      resultFuture,
+                      storageClient
+                          .readObjectCallable()
+                          .withDefaultCallContext(
+                              contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
+                      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());
+      try (UnbufferedReadableByteChannel c = session.open()) {
+        for (ByteBuffer buf : buffers) {
+          c.read(buf);
+        }
+      }
+      Object actualResult = session.getResult().get(1000, TimeUnit.MILLISECONDS);
+      assertThat(actualResult).isEqualTo(expectedResult);
+      assertThat(actualBytes).isEqualTo(bytes);
+    }
+  }
+
+  @Test
+  public void ioException_if_generation_changes() throws IOException, InterruptedException {
+    StorageGrpc.StorageImplBase fakeStorage =
+        new StorageGrpc.StorageImplBase() {
+
+          final AtomicInteger invocationCount = new AtomicInteger(0);
+
+          @Override
+          public void readObject(
+              ReadObjectRequest request, StreamObserver<ReadObjectResponse> responseObserver) {
+            int count = invocationCount.getAndIncrement();
+            if (request.equals(req1)) {
+              if (count == 0) {
+                responseObserver.onNext(resp1);
+                responseObserver.onNext(resp2);
+                responseObserver.onError(apiException(Code.DATA_LOSS));
+              }
+            } else if (request.equals(req2)) {
+              ReadObjectResponse.Builder builder = resp3.toBuilder();
+              // increment the generation, as if it had been updated between initial read and retry
+              builder.getMetadataBuilder().setGeneration(expectedResult.getGeneration() + 1);
+              responseObserver.onNext(builder.build());
+              responseObserver.onNext(resp4);
+              responseObserver.onCompleted();
+            } else {
+              responseObserver.onError(apiException(Code.PERMISSION_DENIED));
+            }
+          }
+        };
+
+    try (FakeServer server = FakeServer.of(fakeStorage);
+        StorageClient storageClient = StorageClient.create(server.storageSettings())) {
+
+      UnbufferedReadableByteChannelSession<Object> session =
+          new UnbufferedReadSession<>(
+              ApiFutures.immediateFuture(req1),
+              (start, resultFuture) ->
+                  new GapicUnbufferedReadableByteChannel(
+                      resultFuture,
+                      storageClient
+                          .readObjectCallable()
+                          .withDefaultCallContext(
+                              contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
+                      start,
+                      Hasher.noop()));
+      byte[] actualBytes = new byte[40];
+      try (UnbufferedReadableByteChannel c = session.open()) {
+        IOException ioException =
+            assertThrows(IOException.class, () -> c.read(ByteBuffer.wrap(actualBytes)));
+
+        assertThat(ioException).hasMessageThat().containsMatch(".*Generation.*3.*4.*");
+      }
+    }
+  }
+
+  @Test
+  public void ioException_if_crc32c_mismatch_individual_message()
+      throws IOException, InterruptedException {
+    StorageGrpc.StorageImplBase fakeStorage =
+        new StorageGrpc.StorageImplBase() {
+          @Override
+          public void readObject(
+              ReadObjectRequest request, StreamObserver<ReadObjectResponse> responseObserver) {
+            if (request.equals(req1)) {
+              responseObserver.onNext(resp1);
+              ReadObjectResponse.Builder b = resp2.toBuilder();
+              // set a bad checksum value
+              b.getChecksummedDataBuilder().setCrc32C(1);
+              responseObserver.onNext(b.build());
+              responseObserver.onNext(resp3);
+              responseObserver.onNext(resp4);
+              responseObserver.onCompleted();
+            } else {
+              responseObserver.onError(apiException(Code.PERMISSION_DENIED));
+            }
+          }
+        };
+    try (FakeServer server = FakeServer.of(fakeStorage);
+        StorageClient storageClient = StorageClient.create(server.storageSettings())) {
+
+      UnbufferedReadableByteChannelSession<Object> session =
+          new UnbufferedReadSession<>(
+              ApiFutures.immediateFuture(req1),
+              (start, resultFuture) ->
+                  new GapicUnbufferedReadableByteChannel(
+                      resultFuture,
+                      storageClient
+                          .readObjectCallable()
+                          .withDefaultCallContext(
+                              contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
+                      start,
+                      Hasher.enabled()));
+      byte[] actualBytes = new byte[40];
+      try (UnbufferedReadableByteChannel c = session.open()) {
+        IOException ioException =
+            assertThrows(IOException.class, () -> c.read(ByteBuffer.wrap(actualBytes)));
+
+        assertThat(ioException).hasMessageThat().contains("Mismatch checksum");
+      }
+    }
+  }
+
+  @Test
+  public void ioException_if_crc32c_mismatch_cumulative() throws IOException, InterruptedException {
+    StorageGrpc.StorageImplBase fakeStorage =
+        new StorageGrpc.StorageImplBase() {
+          @Override
+          public void readObject(
+              ReadObjectRequest request, StreamObserver<ReadObjectResponse> responseObserver) {
+            if (request.equals(req1)) {
+              responseObserver.onNext(resp1);
+              responseObserver.onCompleted();
+            } else {
+              responseObserver.onError(apiException(Code.PERMISSION_DENIED));
+            }
+          }
+        };
+    try (FakeServer server = FakeServer.of(fakeStorage);
+        StorageClient storageClient = StorageClient.create(server.storageSettings())) {
+
+      UnbufferedReadableByteChannelSession<Object> session =
+          new UnbufferedReadSession<>(
+              ApiFutures.immediateFuture(req1),
+              (start, resultFuture) ->
+                  new GapicUnbufferedReadableByteChannel(
+                      resultFuture,
+                      storageClient
+                          .readObjectCallable()
+                          .withDefaultCallContext(
+                              contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
+                      start,
+                      Hasher.enabled()));
+      byte[] actualBytes = new byte[40];
+      //noinspection resource
+      UnbufferedReadableByteChannel c = session.open();
+      c.read(ByteBuffer.wrap(actualBytes));
+
+      IOException ioException = assertThrows(IOException.class, c::close);
+      assertThat(ioException).hasMessageThat().contains("Mismatch checksum");
+    }
+  }
+
+  @Test
+  public void overRead_handledProperly() throws IOException, InterruptedException {
+    StorageGrpc.StorageImplBase fakeStorage =
+        new StorageGrpc.StorageImplBase() {
+          @Override
+          public void readObject(
+              ReadObjectRequest request, StreamObserver<ReadObjectResponse> responseObserver) {
+            responseObserver.onNext(resp1);
+            responseObserver.onNext(resp2);
+            responseObserver.onNext(resp3);
+            responseObserver.onNext(resp4);
+            responseObserver.onCompleted();
+          }
+        };
+    try (FakeServer server = FakeServer.of(fakeStorage);
+        StorageClient storageClient = StorageClient.create(server.storageSettings())) {
+
+      UnbufferedReadableByteChannelSession<Object> session =
+          new UnbufferedReadSession<>(
+              ApiFutures.immediateFuture(req1),
+              (start, resultFuture) ->
+                  new GapicUnbufferedReadableByteChannel(
+                      resultFuture,
+                      storageClient
+                          .readObjectCallable()
+                          .withDefaultCallContext(
+                              contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
+                      start,
+                      Hasher.enabled()));
+      byte[] actualBytes = new byte[41];
+      //noinspection resource
+      UnbufferedReadableByteChannel c = session.open();
+      ByteBuffer buf = ByteBuffer.wrap(actualBytes);
+      int read1 = c.read(buf);
+      assertThat(read1).isAtLeast(1);
+      int read2 = c.read(buf);
+      assertThat(read2).isEqualTo(-1);
+      assertThrows(ClosedChannelException.class, () -> c.read(buf));
+    }
+  }
+
+  private static ChecksummedData getChecksummedData(ByteString content) {
+    int crc32c = Hashing.crc32c().hashBytes(content.asReadOnlyByteBuffer()).asInt();
+    return ChecksummedData.newBuilder().setContent(content).setCrc32C(crc32c).build();
+  }
+
+  private static ApiException apiException(Code code) {
+    StatusRuntimeException statusRuntimeException = code.toStatus().asRuntimeException();
+    DebugInfo debugInfo =
+        DebugInfo.newBuilder().setDetail("forced failure |~| " + code.name()).build();
+    ErrorDetails errorDetails =
+        ErrorDetails.builder().setRawErrorMessages(ImmutableList.of(Any.pack(debugInfo))).build();
+    return ApiExceptionFactory.createException(
+        statusRuntimeException, GrpcStatusCode.of(code), true, errorDetails);
+  }
+
+  private static GrpcCallContext contextWithRetryForCodes(StatusCode.Code... code) {
+    return GrpcCallContext.createDefault().withRetryableCodes(ImmutableSet.copyOf(code));
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcTransformPageDecoratorTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcTransformPageDecoratorTest.java
index 56f50015..187348a9 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcTransformPageDecoratorTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcTransformPageDecoratorTest.java
@@ -18,111 +18,298 @@ package com.google.cloud.storage;
 
 import static com.google.common.truth.Truth.assertThat;
 
+import com.google.api.core.ApiClock;
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
+import com.google.api.core.NanoClock;
 import com.google.api.gax.paging.AbstractPage;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.gax.retrying.RetrySettings;
+import com.google.api.gax.retrying.TimedAttemptSettings;
 import com.google.api.gax.rpc.ApiCallContext;
 import com.google.api.gax.rpc.PageContext;
 import com.google.api.gax.rpc.PagedListDescriptor;
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.storage.GrpcStorageImpl.TransformingPageDecorator;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
+import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Streams;
-import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
 import org.junit.Test;
 import org.mockito.Mockito;
 
 public class GrpcTransformPageDecoratorTest {
 
+  // define some common data used across our tests
+  private final ApiCallContext apiCallContext = Mockito.mock(ApiCallContext.class);
+
+  // Initial values for the first page
+  private final ImmutableList<String> page1 = ImmutableList.of("string1", "string2", "string3");
+  // Request which will be appended to the second page
+  private final ImmutableList<String> page2 = ImmutableList.of("string4");
+  // Request which will be appended to the second page
+  private final ImmutableList<String> page3 = ImmutableList.of("string5");
+
+  // Expected values after the translation
+  private final List<String> expectedValues =
+      Streams.concat(page1.stream(), page2.stream(), page3.stream())
+          .map(String::toUpperCase)
+          .collect(ImmutableList.toImmutableList());
+
+  private final Req req1 = new Req("req1");
+  private final Resp resp1 = new Resp("req2", page1);
+  private final Req req2 = new Req("req2");
+  private final Resp resp2 = new Resp("req3", page2);
+  private final Req req3 = new Req("req3");
+  private final Resp resp3 = new Resp(null, page3);
+  private final ImmutableMap<Req, Resp> data =
+      ImmutableMap.of(
+          req1, resp1,
+          req2, resp2,
+          req3, resp3);
+
   @Test
   public void valueTranslationTest() {
-    // Initial values for the first page
-    List<String> initialValues = ImmutableList.of("string1", "string2", "string3");
-    // Request which will be appended to the second page
-    String request = "string4";
-
-    // Expected values after the translation
-    List<String> expectedValuesPageOne = ImmutableList.of("STRING1", "STRING2", "STRING3");
-    List<String> expectedValuesPageTwo = ImmutableList.of("STRING4");
-    List<String> expectedPagesValuesMerged =
-        Streams.concat(expectedValuesPageOne.stream(), expectedValuesPageTwo.stream())
-            .collect(ImmutableList.toImmutableList());
-
-    StringPagedListDescriptor descriptor = new StringPagedListDescriptor();
-    UnaryCallable<String, List<String>> callable = new StringPagedCallable();
-    PageContext<String, List<String>, String> context =
-        PageContext.create(callable, descriptor, request, Mockito.mock(ApiCallContext.class));
-    ListStringPage page = new ListStringPage(context, initialValues);
-    TransformingPageDecorator<String, List<String>, String, ListStringPage, String> decorator =
-        new TransformingPageDecorator<>(page, String::toUpperCase);
+    UnaryCallable<Req, Resp> callable = new DataDrivenCallable(data);
+    ReqRespDescriptor descriptor = new ReqRespDescriptor();
+    PageContext<Req, Resp, String> context =
+        PageContext.create(callable, descriptor, req1, apiCallContext);
+    ReqRespPage page = new ReqRespPage(context, resp1);
+    TransformingPageDecorator<Req, Resp, String, ReqRespPage, String> decorator =
+        new TransformingPageDecorator<>(
+            page,
+            String::toUpperCase,
+            retryingDeps(),
+            StorageRetryStrategy.getUniformStorageRetryStrategy().getIdempotentHandler());
 
     assertThat(ImmutableList.copyOf(decorator.getValues().iterator()))
-        .containsExactlyElementsIn(expectedValuesPageOne);
+        .containsExactlyElementsIn(
+            page1.stream().map(String::toUpperCase).collect(Collectors.toList()));
     assertThat(ImmutableList.copyOf(decorator.iterateAll().iterator()))
-        .containsExactlyElementsIn(expectedPagesValuesMerged);
+        .containsExactlyElementsIn(expectedValues);
   }
 
-  private static class ListStringPage
-      extends AbstractPage<String, List<String>, String, ListStringPage> {
+  @Test
+  public void retryWorks() {
+    FailureInducingCallable callable = new FailureInducingCallable(new DataDrivenCallable(data));
+
+    ShouldRetryExceptionAlgorithm alg = new ShouldRetryExceptionAlgorithm();
+
+    ReqRespDescriptor descriptor = new ReqRespDescriptor();
+    PageContext<Req, Resp, String> context =
+        PageContext.create(callable, descriptor, req1, apiCallContext);
+    ReqRespPage page = new ReqRespPage(context, resp1);
+    TransformingPageDecorator<Req, Resp, String, ReqRespPage, String> decorator =
+        new TransformingPageDecorator<>(page, String::toUpperCase, retryingDeps(), alg);
 
-    public ListStringPage(
-        PageContext<String, List<String>, String> context, List<String> response) {
+    ImmutableList<String> actual = ImmutableList.copyOf(decorator.iterateAll().iterator());
+    assertThat(actual).containsExactlyElementsIn(expectedValues);
+    // sometimes shouldRetry is called multiple times, not totally sure why
+    assertThat(alg.shouldRetryCallCount.get()).isAtLeast(1);
+    // we expect to attempt the RPC twice, first attempt fails, latter calls succeed
+    assertThat(callable.callableCallCount.get()).isEqualTo(3);
+  }
+
+  static RetryingDependencies retryingDeps() {
+    return new RetryingDependencies() {
+      @Override
+      public RetrySettings getRetrySettings() {
+        return StorageOptions.getDefaultRetrySettings();
+      }
+
+      @Override
+      public ApiClock getClock() {
+        return NanoClock.getDefaultClock();
+      }
+    };
+  }
+
+  private static class ReqRespPage extends AbstractPage<Req, Resp, String, ReqRespPage> {
+
+    private final Resp response;
+
+    public ReqRespPage(PageContext<Req, Resp, String> context, Resp response) {
       super(context, response);
+      this.response = response;
+    }
+
+    @Override
+    public ReqRespPage createPage(PageContext<Req, Resp, String> context, Resp response) {
+      return new ReqRespPage(context, response);
     }
 
     @Override
-    public ListStringPage createPage(
-        PageContext<String, List<String>, String> context, List<String> response) {
-      return new ListStringPage(context, response);
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("response", response).toString();
     }
   }
 
-  private static class StringPagedListDescriptor
-      implements PagedListDescriptor<String, List<String>, String> {
+  private static class ReqRespDescriptor implements PagedListDescriptor<Req, Resp, String> {
+
     @Override
     public String emptyToken() {
       return "";
     }
 
     @Override
-    public String injectToken(String payload, String token) {
-      return payload;
+    public Req injectToken(Req payload, String token) {
+      return payload.withToken(token);
     }
 
     @Override
-    public String injectPageSize(String payload, int pageSize) {
-      return null;
+    public Req injectPageSize(Req payload, int pageSize) {
+      return payload.withPageSize(pageSize);
     }
 
     @Override
-    public Integer extractPageSize(String payload) {
-      return null;
+    public Integer extractPageSize(Req payload) {
+      return payload.pageSize;
     }
 
     @Override
-    public String extractNextToken(List<String> payload) {
-      return payload.size() > 0 ? payload.get(0) : emptyToken();
+    public String extractNextToken(Resp payload) {
+      return payload.nextToken != null ? payload.nextToken : emptyToken();
+    }
+
+    @Override
+    public Iterable<String> extractResources(Resp payload) {
+      return payload.resources;
+    }
+  }
+
+  /**
+   * Provide an immutable map of {@link Req} to {@link Resp}, when {@link #futureCall(Req,
+   * ApiCallContext)} is invoked attempt to retrieve the Resp from the provided map, if it's present
+   * return it, if not return an exception.
+   */
+  private static class DataDrivenCallable extends UnaryCallable<Req, Resp> {
+    private final ImmutableMap<Req, Resp> data;
+
+    private DataDrivenCallable(ImmutableMap<Req, Resp> data) {
+      this.data = data;
     }
 
     @Override
-    public Iterable<String> extractResources(List<String> payload) {
-      return payload;
+    public ApiFuture<Resp> futureCall(Req request, ApiCallContext context) {
+      if (data.containsKey(request)) {
+        return ApiFutures.immediateFuture(data.get(request));
+      }
+      return ApiFutures.immediateFailedFuture(
+          new IllegalStateException("No matching response for request: " + request));
     }
   }
 
-  private static class StringPagedCallable extends UnaryCallable<String, List<String>> {
-    // We only want to add one additional page with the same value as the request,
-    // this is kind of hacky, but I wanted to validate we are performing iterate all
-    // properly.
-    private int numberOfPages = 1;
+  private static final class Req {
+    private final String token;
+    private final int pageSize;
+
+    Req(String token) {
+      this(token, 5);
+    }
+
+    Req(String token, int pageSize) {
+      this.token = token;
+      this.pageSize = pageSize;
+    }
+
+    Req withToken(String t) {
+      return new Req(t, pageSize);
+    }
+
+    Req withPageSize(int i) {
+      return new Req(token, i);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof Req)) {
+        return false;
+      }
+      Req req = (Req) o;
+      return pageSize == req.pageSize && Objects.equals(token, req.token);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(token, pageSize);
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this)
+          .add("token", token)
+          .add("pageSize", pageSize)
+          .toString();
+    }
+  }
+
+  private static final class Resp {
+    private final String nextToken;
+    private final ImmutableList<String> resources;
+
+    public Resp(String nextToken, ImmutableList<String> resources) {
+      this.resources = resources;
+      this.nextToken = nextToken;
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this)
+          .add("nextToken", nextToken)
+          .add("resources", resources)
+          .toString();
+    }
+  }
+
+  private static final class ShouldRetryException extends RuntimeException {}
+
+  private static final class ShouldRetryExceptionAlgorithm implements ResultRetryAlgorithm<Object> {
+    private final AtomicInteger shouldRetryCallCount;
+
+    public ShouldRetryExceptionAlgorithm() {
+      shouldRetryCallCount = new AtomicInteger(0);
+    }
+
+    @Override
+    public TimedAttemptSettings createNextAttempt(
+        Throwable prevThrowable, Object prevResponse, TimedAttemptSettings prevSettings) {
+      return null;
+    }
+
+    @Override
+    public boolean shouldRetry(Throwable prevThrowable, Object prevResponse)
+        throws CancellationException {
+      shouldRetryCallCount.incrementAndGet();
+      return prevThrowable instanceof ShouldRetryException;
+    }
+  }
+
+  private static final class FailureInducingCallable extends UnaryCallable<Req, Resp> {
+
+    private final AtomicInteger callableCallCount;
+    private final UnaryCallable<Req, Resp> delegate;
+
+    FailureInducingCallable(UnaryCallable<Req, Resp> delegate) {
+      this.delegate = delegate;
+      this.callableCallCount = new AtomicInteger(0);
+    }
 
     @Override
-    public ApiFuture<List<String>> futureCall(String request, ApiCallContext context) {
-      if (numberOfPages-- > 0) {
-        return ApiFutures.immediateFuture(ImmutableList.of(request));
+    public ApiFuture<Resp> futureCall(Req request, ApiCallContext context) {
+      int callCount = callableCallCount.getAndIncrement();
+      if (callCount == 0) {
+        throw new ShouldRetryException();
       }
-      return ApiFutures.immediateFuture(Collections.emptyList());
+      return delegate.futureCall(request, context);
     }
   }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ResultRetryAlgorithmCompatibilityTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ResultRetryAlgorithmCompatibilityTest.java
new file mode 100644
index 00000000..717500d7
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ResultRetryAlgorithmCompatibilityTest.java
@@ -0,0 +1,43 @@
+/*
+ * 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.gax.retrying.BasicResultRetryAlgorithm;
+import com.google.api.gax.rpc.DataLossException;
+import com.google.api.gax.rpc.StatusCode.Code;
+import com.google.common.collect.ImmutableSet;
+import java.util.Set;
+import org.junit.Test;
+
+public final class ResultRetryAlgorithmCompatibilityTest {
+  @Test
+  public void validateConversion() {
+    BasicResultRetryAlgorithm<Object> alg =
+        new BasicResultRetryAlgorithm<java.lang.Object>() {
+          @Override
+          public boolean shouldRetry(
+              Throwable previousThrowable, java.lang.Object previousResponse) {
+            return previousThrowable instanceof DataLossException;
+          }
+        };
+    Set<Code> codes = GrpcStorageImpl.resultRetryAlgorithmToCodes(alg);
+
+    assertThat(codes).isEqualTo(ImmutableSet.of(Code.DATA_LOSS));
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java
index 8099fe36..6e384bb3 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java
@@ -34,8 +34,8 @@ import com.google.protobuf.ByteString;
 import com.google.storage.v2.CommonObjectRequestParams;
 import com.google.storage.v2.ComposeObjectRequest;
 import com.google.storage.v2.CreateBucketRequest;
+import com.google.storage.v2.CreateHmacKeyRequest;
 import com.google.storage.v2.DeleteBucketRequest;
-import com.google.storage.v2.DeleteHmacKeyRequest;
 import com.google.storage.v2.DeleteObjectRequest;
 import com.google.storage.v2.GetBucketRequest;
 import com.google.storage.v2.GetHmacKeyRequest;
@@ -828,14 +828,14 @@ public final class UnifiedOptsGrpcTest {
       }
 
       @Test
-      public void deleteHmacKey_nonResource() {
-        DeleteHmacKeyRequest expected =
-            DeleteHmacKeyRequest.newBuilder().setProject("projects/proj").build();
+      public void createHmacKey_nonResource() {
+        CreateHmacKeyRequest expected =
+            CreateHmacKeyRequest.newBuilder().setProject("projects/proj").build();
 
-        DeleteHmacKeyRequest actual =
+        CreateHmacKeyRequest actual =
             UnifiedOpts.projectId("proj")
-                .deleteHmacKey()
-                .apply(DeleteHmacKeyRequest.newBuilder())
+                .createHmacKey()
+                .apply(CreateHmacKeyRequest.newBuilder())
                 .build();
         assertThat(actual).isEqualTo(expected);
       }
@@ -867,14 +867,40 @@ public final class UnifiedOptsGrpcTest {
       }
 
       @Test
-      public void deleteHmacKey_resource() {
-        DeleteHmacKeyRequest expected =
-            DeleteHmacKeyRequest.newBuilder().setProject("projects/proj").build();
+      public void createHmacKey_resource() {
+        CreateHmacKeyRequest expected =
+            CreateHmacKeyRequest.newBuilder().setProject("projects/proj").build();
 
-        DeleteHmacKeyRequest actual =
+        CreateHmacKeyRequest actual =
             UnifiedOpts.projectId("projects/proj")
-                .deleteHmacKey()
-                .apply(DeleteHmacKeyRequest.newBuilder())
+                .createHmacKey()
+                .apply(CreateHmacKeyRequest.newBuilder())
+                .build();
+        assertThat(actual).isEqualTo(expected);
+      }
+
+      @Test
+      public void listBuckets_nonResource() {
+        ListBucketsRequest expected =
+            ListBucketsRequest.newBuilder().setParent("projects/proj").build();
+
+        ListBucketsRequest actual =
+            UnifiedOpts.projectId("proj")
+                .listBuckets()
+                .apply(ListBucketsRequest.newBuilder())
+                .build();
+        assertThat(actual).isEqualTo(expected);
+      }
+
+      @Test
+      public void listBuckets_resource() {
+        ListBucketsRequest expected =
+            ListBucketsRequest.newBuilder().setParent("projects/proj").build();
+
+        ListBucketsRequest actual =
+            UnifiedOpts.projectId("projects/proj")
+                .listBuckets()
+                .apply(ListBucketsRequest.newBuilder())
                 .build();
         assertThat(actual).isEqualTo(expected);
       }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
index 4147f68b..cca6645d 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
@@ -24,12 +24,19 @@ import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.Bucket;
 import com.google.cloud.storage.BucketFixture;
 import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.HmacKey;
+import com.google.cloud.storage.HmacKey.HmacKeyMetadata;
+import com.google.cloud.storage.HmacKey.HmacKeyState;
+import com.google.cloud.storage.ServiceAccount;
+import com.google.cloud.storage.Storage.CreateHmacKeyOption;
+import com.google.cloud.storage.Storage.ListHmacKeysOption;
 import com.google.cloud.storage.StorageFixture;
 import com.google.cloud.storage.StorageOptions;
 import com.google.cloud.storage.conformance.retry.CleanupStrategy;
 import com.google.cloud.storage.conformance.retry.TestBench;
 import com.google.cloud.storage.testing.RemoteStorageHelper;
 import com.google.common.collect.ImmutableList;
+import java.util.stream.IntStream;
 import java.util.stream.StreamSupport;
 import org.junit.ClassRule;
 import org.junit.Test;
@@ -86,4 +93,67 @@ public final class ITGrpcTest {
 
     assertThat(bucketNames).contains(bucketFixture.getBucketInfo().getName());
   }
+
+  @Test
+  public void createHmacKey() {
+    ServiceAccount serviceAccount = ServiceAccount.of("x@y.z");
+    HmacKey hmacKey = storageFixture.getInstance().createHmacKey(serviceAccount);
+    assertThat(hmacKey).isNotNull();
+    assertThat(hmacKey.getSecretKey()).isNotNull();
+    assertThat(hmacKey.getMetadata().getServiceAccount()).isEqualTo(serviceAccount);
+  }
+
+  @Test
+  public void getHmacKey() {
+    ServiceAccount serviceAccount = ServiceAccount.of("x@y.z");
+    HmacKey hmacKey = storageFixture.getInstance().createHmacKey(serviceAccount);
+    HmacKeyMetadata actual =
+        storageFixture.getInstance().getHmacKey(hmacKey.getMetadata().getAccessId());
+    assertThat(actual).isEqualTo(hmacKey.getMetadata());
+  }
+
+  @Test
+  public void listHmacKeys() {
+    ImmutableList<HmacKey> keys =
+        IntStream.rangeClosed(1, 4)
+            .mapToObj(i -> ServiceAccount.of(String.format("x-%d@y.z", i)))
+            .map(
+                sa ->
+                    storageFixture
+                        .getInstance()
+                        .createHmacKey(sa, CreateHmacKeyOption.projectId("proj")))
+            .collect(ImmutableList.toImmutableList());
+
+    ImmutableList<HmacKeyMetadata> expected =
+        keys.stream().map(HmacKey::getMetadata).collect(ImmutableList.toImmutableList());
+
+    Page<HmacKeyMetadata> page =
+        storageFixture.getInstance().listHmacKeys(ListHmacKeysOption.projectId("proj"));
+
+    ImmutableList<HmacKeyMetadata> actual =
+        StreamSupport.stream(page.iterateAll().spliterator(), false)
+            .collect(ImmutableList.toImmutableList());
+
+    assertThat(actual).isEqualTo(expected);
+  }
+
+  @Test
+  public void updateHmacKey() {
+    ServiceAccount serviceAccount = ServiceAccount.of("x@y.z");
+    HmacKey hmacKey = storageFixture.getInstance().createHmacKey(serviceAccount);
+    HmacKeyMetadata updated =
+        storageFixture
+            .getInstance()
+            .updateHmacKeyState(hmacKey.getMetadata(), HmacKeyState.INACTIVE);
+    assertThat(updated.getServiceAccount()).isEqualTo(serviceAccount);
+    assertThat(updated.getState()).isEqualTo(HmacKeyState.INACTIVE);
+  }
+
+  @Test
+  public void deleteHmacKey() {
+    ServiceAccount serviceAccount = ServiceAccount.of("x@y.z");
+    HmacKey hmacKey = storageFixture.getInstance().createHmacKey(serviceAccount);
+    storageFixture.getInstance().updateHmacKeyState(hmacKey.getMetadata(), HmacKeyState.INACTIVE);
+    storageFixture.getInstance().deleteHmacKey(hmacKey.getMetadata());
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/jqwik/HmacKeyMetadataArbitraryProvider.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/jqwik/HmacKeyMetadataArbitraryProvider.java
index fd305e6d..bf114ae0 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/jqwik/HmacKeyMetadataArbitraryProvider.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/jqwik/HmacKeyMetadataArbitraryProvider.java
@@ -54,7 +54,7 @@ public class HmacKeyMetadataArbitraryProvider implements ArbitraryProvider {
                 (accessId, projectID, state, createTime, updateTime, email) ->
                     HmacKeyMetadata.newBuilder()
                         .setAccessId(accessId)
-                        .setProject(projectID.get())
+                        .setProject("projects/" + projectID.get())
                         .setId(projectID.get() + "/" + accessId)
                         .setState(state.toString())
                         .setCreateTime(createTime)