Repository: googleapis/java-storage
Pull request #1573 (2022-08-17 17:21:24)

test: add error handling and retry tests for GapicUnbufferedReadableByteChannel
### Testing
Add new FakeServer which allows defining an in process grpc storage service rather than trying to fight with mocks.

Add test for GapicUnbufferedReadableByteChannel to verify:
* retries happening at the ServerStream level return sound data
* Generation changing between a resumed read results in an IOException rather than proceeding
* checksum mismatch for an individual ChecksummedData
* checksum mismatch across all cumulative ChecksummedData

#### Refactoring
* Move Crc32cValue#nullSafeConcat into Hasher so that it can be context aware
  whether to actually perform any computation

### Retries

Wire in Automatic Retries for ReadObject operations

The ServerStream provided by gax isn't friendly to decoration for its iterator.
Instead, leverage the built-in stream resumption provided by gax.

In order to provide compatibility with our existing public apis we need to
translate a ResultRetryAlgorithm<?> into a Set<StatusCode.Code> which can
be passed to GrpcCallContext#withRetryableCodes(Set<StatusCode.Code>). To do
this we construct an ApiException for each Code then test against the provided
ResultRetryAlgorithm.

Additionally, a StreamResumptionStrategy must be provided which can construct
and updated request for resume. Unfortunately, at this time this Resumption
Strategy must be set globally at StorageSettings construction time.

One side effect of using gax's built in ServerStream resumption is it only
works with an instance of StorageClient constructed from StorageSettings.
If the StorageClient is constructed from a GrpcStorageStub the retries don't
get wired in. As a consequence, all usage of GrpcStorageStub in GrpcStorageImpl
has been replaced with StorageClient.



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/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/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index 16b5a5af..6cd82865 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;
@@ -71,6 +75,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;
@@ -82,7 +87,7 @@ 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;
@@ -114,19 +119,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;
 
-  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 +154,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 +174,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 +203,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 +252,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 +267,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
           ResumableMedia.gapic()
               .write()
               .resumableWrite(
-                  grpcStorageStub
+                  storageClient
                       .startResumableWriteCallable()
                       .withDefaultCallContext(grpcCallContext),
                   req);
@@ -290,16 +309,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 +348,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 +373,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,7 +385,7 @@ 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());
@@ -383,7 +400,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   @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());
@@ -411,7 +428,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 +447,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 +468,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 +495,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 +530,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 +548,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 +584,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 +599,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 +627,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 +654,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));
@@ -846,7 +879,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   @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());
@@ -880,7 +913,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         getOptions(),
         retryAlgorithmManager.getFor(req),
         () -> {
-          grpcStorageStub.deleteHmacKeyCallable().call(req, grpcCallContext);
+          storageClient.deleteHmacKeyCallable().call(req, grpcCallContext);
           return null;
         },
         Decoder.identity());
@@ -915,7 +948,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());
   }
 
@@ -1085,4 +1118,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..97abe512 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,11 +38,16 @@ 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)
@@ -54,12 +61,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 +74,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 +347,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 +394,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);
       }
     }
   }


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/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));
+  }
+}