Repository: googleapis/java-storage
Pull request #2715 (2024-09-19 20:36:28)

fix: update GrpcStorageOptions to attempt direct path by default


Patch:
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 82789bcd..fcb6cf1d 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
@@ -20,8 +20,10 @@ import static java.util.Objects.requireNonNull;
 
 import com.google.api.core.ApiFutures;
 import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.cloud.storage.BufferedReadableByteChannelSession.BufferedReadableByteChannel;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.storage.v2.Object;
@@ -43,27 +45,32 @@ final class GapicDownloadSessionBuilder {
     return INSTANCE;
   }
 
-  /**
-   * 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,
+      RetryingDependencies retryingDependencies,
+      ResultRetryAlgorithm<?> resultRetryAlgorithm,
       ResponseContentLifecycleManager responseContentLifecycleManager) {
-    return new ReadableByteChannelSessionBuilder(read, responseContentLifecycleManager);
+    return new ReadableByteChannelSessionBuilder(
+        read, retryingDependencies, resultRetryAlgorithm, responseContentLifecycleManager);
   }
 
   public static final class ReadableByteChannelSessionBuilder {
 
     private final ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read;
+    private final RetryingDependencies retryingDependencies;
+    private final ResultRetryAlgorithm<?> resultRetryAlgorithm;
     private final ResponseContentLifecycleManager responseContentLifecycleManager;
     private boolean autoGzipDecompression;
     private Hasher hasher;
 
     private ReadableByteChannelSessionBuilder(
         ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read,
+        RetryingDependencies retryingDependencies,
+        ResultRetryAlgorithm<?> resultRetryAlgorithm,
         ResponseContentLifecycleManager responseContentLifecycleManager) {
       this.read = read;
+      this.retryingDependencies = retryingDependencies;
+      this.resultRetryAlgorithm = resultRetryAlgorithm;
       this.responseContentLifecycleManager = responseContentLifecycleManager;
       this.hasher = Hasher.noop();
       this.autoGzipDecompression = false;
@@ -105,12 +112,24 @@ final class GapicDownloadSessionBuilder {
         if (autoGzipDecompression) {
           return new GzipReadableByteChannel(
               new GapicUnbufferedReadableByteChannel(
-                  resultFuture, read, object, hasher, responseContentLifecycleManager),
+                  resultFuture,
+                  read,
+                  object,
+                  hasher,
+                  retryingDependencies,
+                  resultRetryAlgorithm,
+                  responseContentLifecycleManager),
               ApiFutures.transform(
                   resultFuture, Object::getContentEncoding, MoreExecutors.directExecutor()));
         } else {
           return new GapicUnbufferedReadableByteChannel(
-              resultFuture, read, object, hasher, responseContentLifecycleManager);
+              resultFuture,
+              read,
+              object,
+              hasher,
+              retryingDependencies,
+              resultRetryAlgorithm,
+              responseContentLifecycleManager);
         }
       };
     }
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 8904dce6..6f4ff260 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
@@ -19,38 +19,55 @@ package com.google.cloud.storage;
 import com.google.api.client.http.HttpStatusCodes;
 import com.google.api.core.ApiFuture;
 import com.google.api.core.SettableApiFuture;
-import com.google.api.gax.rpc.ServerStream;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.gax.rpc.ApiExceptions;
 import com.google.api.gax.rpc.ServerStreamingCallable;
+import com.google.api.gax.rpc.StateCheckingResponseObserver;
+import com.google.api.gax.rpc.StreamController;
+import com.google.cloud.BaseServiceException;
+import com.google.cloud.storage.Conversions.Decoder;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
 import com.google.protobuf.ByteString;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.Object;
 import com.google.storage.v2.ReadObjectRequest;
 import com.google.storage.v2.ReadObjectResponse;
-import java.io.Closeable;
+import io.grpc.Status.Code;
 import java.io.IOException;
+import java.io.InterruptedIOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 import java.nio.channels.ScatteringByteChannel;
-import java.util.Iterator;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import org.checkerframework.checker.nullness.qual.NonNull;
 
 final class GapicUnbufferedReadableByteChannel
     implements UnbufferedReadableByteChannel, ScatteringByteChannel {
+  private static final java.lang.Object EOF_MARKER = new java.lang.Object();
 
   private final SettableApiFuture<Object> result;
   private final ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read;
   private final ReadObjectRequest req;
   private final Hasher hasher;
-  private final LazyServerStreamIterator iter;
   private final ResponseContentLifecycleManager rclm;
+  private final RetryingDependencies retryingDeps;
+  private final ResultRetryAlgorithm<?> alg;
+  private final SimpleBlockingQueue<java.lang.Object> queue;
 
-  private boolean open = true;
-  private boolean complete = false;
-  private long blobOffset;
+  private final AtomicLong fetchOffset;
+  private volatile ReadObjectObserver readObjectObserver;
+  private volatile boolean open = true;
+  private volatile boolean complete = false;
 
+  private long blobOffset;
   private Object metadata;
-
   private ResponseContentLifecycleHandle leftovers;
 
   GapicUnbufferedReadableByteChannel(
@@ -58,14 +75,21 @@ final class GapicUnbufferedReadableByteChannel
       ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read,
       ReadObjectRequest req,
       Hasher hasher,
+      RetryingDependencies retryingDependencies,
+      ResultRetryAlgorithm<?> alg,
       ResponseContentLifecycleManager rclm) {
     this.result = result;
     this.read = read;
     this.req = req;
     this.hasher = hasher;
+    this.fetchOffset = new AtomicLong(req.getReadOffset());
     this.blobOffset = req.getReadOffset();
     this.rclm = rclm;
-    this.iter = new LazyServerStreamIterator();
+    this.retryingDeps = retryingDependencies;
+    this.alg = alg;
+    // The reasoning for 2 elements below allow for a single response and the EOF/error signal
+    // from onComplete or onError. Same thing com.google.api.gax.rpc.QueuingResponseObserver does.
+    this.queue = new SimpleBlockingQueue<>(2);
   }
 
   @Override
@@ -90,48 +114,63 @@ final class GapicUnbufferedReadableByteChannel
         continue;
       }
 
-      if (iter.hasNext()) {
-        ReadObjectResponse resp = iter.next();
-        ResponseContentLifecycleHandle handle = rclm.get(resp);
-        if (resp.hasMetadata()) {
-          Object respMetadata = resp.getMetadata();
-          if (metadata == null) {
-            metadata = respMetadata;
-          } else if (metadata.getGeneration() != respMetadata.getGeneration()) {
-            throw closeWithError(
-                String.format(
-                    "Mismatch Generation between subsequent reads. Expected %d but received %d",
-                    metadata.getGeneration(), respMetadata.getGeneration()));
-          }
-
-          if (!result.isDone()) {
-            result.set(metadata);
-          }
-        }
-        ChecksummedData checksummedData = resp.getChecksummedData();
-        ByteString content = checksummedData.getContent();
-        int contentSize = content.size();
-        // Very important to know whether a crc32c value is set. Without checking, protobuf will
-        // happily return 0, which is a valid crc32c value.
-        if (checksummedData.hasCrc32C()) {
-          Crc32cLengthKnown expected = Crc32cValue.of(checksummedData.getCrc32C(), contentSize);
-          try {
-            hasher.validate(expected, content.asReadOnlyByteBufferList());
-          } catch (IOException e) {
-            close();
-            throw e;
-          }
-        }
-        handle.copy(c, dsts, offset, length);
-        if (handle.hasRemaining()) {
-          leftovers = handle;
+      ensureStreamOpen();
+      java.lang.Object take;
+      try {
+        take = queue.poll();
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        throw new InterruptedIOException();
+      }
+      if (take instanceof Throwable) {
+        Throwable throwable = (Throwable) take;
+        BaseServiceException coalesce = StorageException.coalesce(throwable);
+        if (alg.shouldRetry(coalesce, null)) {
+          readObjectObserver = null;
+          continue;
         } else {
-          handle.close();
+          throw new IOException(coalesce);
         }
-      } else {
+      }
+      if (take == EOF_MARKER) {
         complete = true;
         break;
       }
+      readObjectObserver.request();
+
+      ReadObjectResponse resp = (ReadObjectResponse) take;
+      ResponseContentLifecycleHandle handle = rclm.get(resp);
+      if (resp.hasMetadata()) {
+        Object respMetadata = resp.getMetadata();
+        if (metadata == null) {
+          metadata = respMetadata;
+        } else if (metadata.getGeneration() != respMetadata.getGeneration()) {
+          throw closeWithError(
+              String.format(
+                  "Mismatch Generation between subsequent reads. Expected %d but received %d",
+                  metadata.getGeneration(), respMetadata.getGeneration()));
+        }
+      }
+      ChecksummedData checksummedData = resp.getChecksummedData();
+      ByteString content = checksummedData.getContent();
+      int contentSize = content.size();
+      // Very important to know whether a crc32c value is set. Without checking, protobuf will
+      // happily return 0, which is a valid crc32c value.
+      if (checksummedData.hasCrc32C()) {
+        Crc32cLengthKnown expected = Crc32cValue.of(checksummedData.getCrc32C(), contentSize);
+        try {
+          hasher.validate(expected, content.asReadOnlyByteBufferList());
+        } catch (IOException e) {
+          close();
+          throw e;
+        }
+      }
+      handle.copy(c, dsts, offset, length);
+      if (handle.hasRemaining()) {
+        leftovers = handle;
+      } else {
+        handle.close();
+      }
     }
     long read = c.read();
 
@@ -152,8 +191,64 @@ final class GapicUnbufferedReadableByteChannel
       if (leftovers != null) {
         leftovers.close();
       }
+      ReadObjectObserver obs = readObjectObserver;
+      if (obs != null && !obs.cancellation.isDone()) {
+        obs.cancel();
+        drainQueue();
+        try {
+          // make sure our waiting doesn't lockup permanently
+          obs.cancellation.get(1, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+          InterruptedIOException ioe = new InterruptedIOException();
+          ioe.initCause(e);
+          ioe.addSuppressed(new AsyncStorageTaskException());
+          throw ioe;
+        } catch (ExecutionException e) {
+          Throwable cause = e;
+          if (e.getCause() != null) {
+            cause = e.getCause();
+          }
+          IOException ioException = new IOException(cause);
+          ioException.addSuppressed(new AsyncStorageTaskException());
+          throw ioException;
+        } catch (TimeoutException ignore) {
+        }
+      }
     } finally {
-      iter.close();
+      drainQueue();
+    }
+  }
+
+  private void drainQueue() throws IOException {
+    IOException ioException = null;
+    while (queue.nonEmpty()) {
+      try {
+        java.lang.Object queueValue = queue.poll();
+        if (queueValue instanceof ReadObjectResponse) {
+          ReadObjectResponse resp = (ReadObjectResponse) queueValue;
+          ResponseContentLifecycleHandle handle = rclm.get(resp);
+          handle.close();
+        } else if (queueValue == EOF_MARKER || queueValue instanceof Throwable) {
+          break;
+        }
+      } catch (IOException e) {
+        if (ioException == null) {
+          ioException = e;
+        } else if (ioException != e) {
+          ioException.addSuppressed(e);
+        }
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        if (ioException == null) {
+          ioException = new InterruptedIOException();
+        } else {
+          ioException.addSuppressed(e);
+        }
+      }
+    }
+    if (ioException != null) {
+      throw ioException;
     }
   }
 
@@ -161,6 +256,27 @@ final class GapicUnbufferedReadableByteChannel
     return result;
   }
 
+  private void ensureStreamOpen() {
+    if (readObjectObserver == null) {
+      readObjectObserver =
+          Retrying.run(
+              retryingDeps,
+              alg,
+              () -> {
+                ReadObjectObserver tmp = new ReadObjectObserver();
+                ReadObjectRequest request = req;
+                long currentFetchOffset = fetchOffset.get();
+                if (request.getReadOffset() != currentFetchOffset) {
+                  request = req.toBuilder().setReadOffset(currentFetchOffset).build();
+                }
+                read.call(request, tmp);
+                ApiExceptions.callAndTranslateApiException(tmp.open);
+                return tmp;
+              },
+              Decoder.identity());
+    }
+  }
+
   private IOException closeWithError(String message) throws IOException {
     close();
     StorageException cause =
@@ -168,83 +284,95 @@ final class GapicUnbufferedReadableByteChannel
     throw new IOException(message, cause);
   }
 
-  private final class LazyServerStreamIterator implements Iterator<ReadObjectResponse>, Closeable {
-    private ServerStream<ReadObjectResponse> serverStream;
-    private Iterator<ReadObjectResponse> responseIterator;
+  private final class ReadObjectObserver extends StateCheckingResponseObserver<ReadObjectResponse> {
+
+    private final SettableApiFuture<Void> open = SettableApiFuture.create();
+    private final SettableApiFuture<Throwable> cancellation = SettableApiFuture.create();
+
+    private volatile StreamController controller;
 
-    private volatile boolean streamInitialized = false;
+    void request() {
+      controller.request(1);
+    }
+
+    void cancel() {
+      controller.cancel();
+    }
+
+    @Override
+    protected void onStartImpl(StreamController controller) {
+      this.controller = controller;
+      controller.disableAutoInboundFlowControl();
+      controller.request(1);
+    }
 
     @Override
-    public boolean hasNext() {
+    protected void onResponseImpl(ReadObjectResponse response) {
       try {
-        return ensureResponseIteratorOpen().hasNext();
-      } catch (RuntimeException e) {
-        if (!result.isDone()) {
-          result.setException(StorageException.coalesce(e));
+        open.set(null);
+        queue.offer(response);
+        fetchOffset.addAndGet(response.getChecksummedData().getContent().size());
+        if (response.hasMetadata() && !result.isDone()) {
+          result.set(response.getMetadata());
         }
-        reset();
-        throw e;
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        throw Code.ABORTED.toStatus().withCause(e).asRuntimeException();
       }
     }
 
     @Override
-    public ReadObjectResponse next() {
+    protected void onErrorImpl(Throwable t) {
+      open.setException(t);
+      if (!alg.shouldRetry(t, null)) {
+        result.setException(StorageException.coalesce(t));
+      }
+      if (t instanceof CancellationException) {
+        cancellation.set(t);
+      }
       try {
-        return ensureResponseIteratorOpen().next();
-      } catch (RuntimeException e) {
-        if (!result.isDone()) {
-          result.setException(StorageException.coalesce(e));
-        }
-        reset();
-        throw e;
+        queue.offer(t);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        throw Code.ABORTED.toStatus().withCause(e).asRuntimeException();
       }
     }
 
     @Override
-    public void close() {
-      if (serverStream != null) {
-        serverStream.cancel();
-        if (responseIterator != null) {
-          IOException ioException = null;
-          while (responseIterator.hasNext()) {
-            try {
-              ReadObjectResponse next = responseIterator.next();
-              ResponseContentLifecycleHandle handle = rclm.get(next);
-              handle.close();
-            } catch (IOException e) {
-              if (ioException == null) {
-                ioException = e;
-              } else if (ioException != e) {
-                ioException.addSuppressed(e);
-              }
-            }
-          }
-        }
+    protected void onCompleteImpl() {
+      try {
+        cancellation.set(null);
+        queue.offer(EOF_MARKER);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        throw Code.ABORTED.toStatus().withCause(e).asRuntimeException();
       }
     }
+  }
 
-    private Iterator<ReadObjectResponse> ensureResponseIteratorOpen() {
-      boolean initialized = streamInitialized;
-      if (initialized) {
-        return responseIterator;
-      } else {
-        synchronized (this) {
-          if (!streamInitialized) {
-            if (serverStream == null) {
-              serverStream = read.call(req);
-            }
-            responseIterator = serverStream.iterator();
-            streamInitialized = true;
-          }
-          return responseIterator;
-        }
-      }
+  /**
+   * Simplified wrapper around an {@link java.util.concurrent.ArrayBlockingQueue}. We don't need the
+   * majority of methods/functionality just blocking offer/poll.
+   */
+  static final class SimpleBlockingQueue<T> {
+
+    private final ArrayBlockingQueue<T> queue;
+
+    SimpleBlockingQueue(int poolMaxSize) {
+      this.queue = new ArrayBlockingQueue<>(poolMaxSize);
+    }
+
+    public boolean nonEmpty() {
+      return !queue.isEmpty();
+    }
+
+    @NonNull
+    public T poll() throws InterruptedException {
+      return queue.take();
     }
 
-    private void reset() {
-      serverStream = null;
-      responseIterator = null;
-      streamInitialized = false;
+    public void offer(@NonNull T element) throws InterruptedException {
+      queue.put(element);
     }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java
index 03e3cb51..873f6b6b 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java
@@ -16,10 +16,12 @@
 
 package com.google.cloud.storage;
 
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.RestorableState;
 import com.google.cloud.storage.GapicDownloadSessionBuilder.ReadableByteChannelSessionBuilder;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.storage.v2.Object;
 import com.google.storage.v2.ReadObjectRequest;
 import com.google.storage.v2.ReadObjectResponse;
@@ -28,17 +30,23 @@ import org.checkerframework.checker.nullness.qual.NonNull;
 final class GrpcBlobReadChannel extends BaseStorageReadChannel<Object> {
 
   private final ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read;
+  private final RetryingDependencies retryingDependencies;
+  private final ResultRetryAlgorithm<?> resultRetryAlgorithm;
   private final ResponseContentLifecycleManager responseContentLifecycleManager;
   private final ReadObjectRequest request;
   private final boolean autoGzipDecompression;
 
   GrpcBlobReadChannel(
       ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read,
+      RetryingDependencies retryingDependencies,
+      ResultRetryAlgorithm<?> resultRetryAlgorithm,
       ResponseContentLifecycleManager responseContentLifecycleManager,
       ReadObjectRequest request,
       boolean autoGzipDecompression) {
     super(Conversions.grpc().blobInfo());
     this.read = read;
+    this.retryingDependencies = retryingDependencies;
+    this.resultRetryAlgorithm = resultRetryAlgorithm;
     this.responseContentLifecycleManager = responseContentLifecycleManager;
     this.request = request;
     this.autoGzipDecompression = autoGzipDecompression;
@@ -56,7 +64,11 @@ final class GrpcBlobReadChannel extends BaseStorageReadChannel<Object> {
           ReadableByteChannelSessionBuilder b =
               ResumableMedia.gapic()
                   .read()
-                  .byteChannel(read, responseContentLifecycleManager)
+                  .byteChannel(
+                      read,
+                      retryingDependencies,
+                      resultRetryAlgorithm,
+                      responseContentLifecycleManager)
                   .setHasher(Hasher.noop())
                   .setAutoGzipDecompression(autoGzipDecompression);
           BufferHandle bufferHandle = getBufferHandle();
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 022a3cc1..d76d9ce4 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
@@ -749,6 +749,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
 
     return new GrpcBlobReadChannel(
         storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext),
+        getOptions(),
+        retryAlgorithmManager.getFor(request),
         responseContentLifecycleManager,
         request,
         !opts.autoGzipDecompression());
@@ -1910,6 +1912,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
         .read()
         .byteChannel(
             storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext),
+            getOptions(),
+            retryAlgorithmManager.getFor(readObjectRequest),
             responseContentLifecycleManager)
         .setAutoGzipDecompression(!opts.autoGzipDecompression())
         .unbuffered()
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 f4f3bd6d..10d70669 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
@@ -30,7 +30,6 @@ import com.google.api.gax.grpc.GrpcInterceptorProvider;
 import com.google.api.gax.grpc.GrpcStubCallableFactory;
 import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
 import com.google.api.gax.retrying.RetrySettings;
-import com.google.api.gax.retrying.StreamResumptionStrategy;
 import com.google.api.gax.rpc.ClientContext;
 import com.google.api.gax.rpc.HeaderProvider;
 import com.google.api.gax.rpc.NoHeaderProvider;
@@ -98,7 +97,6 @@ import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Set;
 import org.checkerframework.checker.nullness.qual.NonNull;
-import org.checkerframework.checker.nullness.qual.Nullable;
 import org.threeten.bp.Duration;
 
 /** @since 2.14.0 This new api is in preview and is subject to breaking changes. */
@@ -336,18 +334,12 @@ public final class GrpcStorageOptions extends StorageOptions
         .startResumableWriteSettings()
         .setRetrySettings(baseRetrySettings)
         .setRetryableCodes(startResumableWriteRetryableCodes);
-    // 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.
+    // for ReadObject disable retries and move the total timeout to the idle timeout
     builder
         .readObjectSettings()
         .setRetrySettings(readRetrySettings)
-        // 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())
+        // disable gapic retries because we're handling it ourselves
+        .setRetryableCodes(Collections.emptySet())
         // for reads, the stream can be held open for a long time in order to read all bytes,
         // this is totally valid. instead we want to monitor if the stream is doing work and if not
         // timeout.
@@ -697,7 +689,7 @@ public final class GrpcStorageOptions extends StorageOptions
     /** @since 2.14.0 This new api is in preview and is subject to breaking changes. */
     @BetaApi
     public boolean isAttemptDirectPath() {
-      return false;
+      return true;
     }
 
     /** @since 2.41.0 This new api is in preview and is subject to breaking changes. */
@@ -845,40 +837,6 @@ public final class GrpcStorageOptions extends StorageOptions
     }
   }
 
-  // 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;
-    }
-  }
-
   // setInternalHeaderProvider is protected so we need to open its scope in order to set it
   // we are adding an entry for gccl which is set via this provider
   private static final class GapicStorageSettingsBuilder extends StorageSettings.Builder {


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
index 331856c4..74b1283c 100644
--- 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
@@ -61,6 +61,7 @@ final class FakeServer implements AutoCloseable {
             .setCredentials(NoCredentials.getInstance())
             .setGrpcInterceptorProvider(GrpcPlainRequestLoggingInterceptor.getInterceptorProvider())
             .setEnableGrpcClientMetrics(false)
+            .setAttemptDirectPath(false)
             .build();
     return new FakeServer(server, grpcStorageOptions);
   }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedReadableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedReadableByteChannelTest.java
index 3a4fd7e9..c2901b46 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedReadableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedReadableByteChannelTest.java
@@ -17,13 +17,15 @@
 package com.google.cloud.storage;
 
 import static com.google.cloud.storage.TestUtils.apiException;
-import static com.google.cloud.storage.TestUtils.contextWithRetryForCodes;
 import static com.google.cloud.storage.TestUtils.getChecksummedData;
 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.rpc.StatusCode;
+import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.gax.rpc.ApiException;
+import com.google.api.gax.rpc.DataLossException;
 import com.google.cloud.storage.ChannelSession.UnbufferedReadSession;
 import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
 import com.google.common.collect.ImmutableList;
@@ -122,12 +124,11 @@ public final class ITGapicUnbufferedReadableByteChannelTest {
               (start, resultFuture) ->
                   new GapicUnbufferedReadableByteChannel(
                       resultFuture,
-                      storageClient
-                          .readObjectCallable()
-                          .withDefaultCallContext(
-                              contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
+                      storageClient.readObjectCallable(),
                       start,
                       Hasher.noop(),
+                      server.getGrpcStorageOptions(),
+                      retryOnly(DataLossException.class),
                       ResponseContentLifecycleManager.noop()));
       byte[] actualBytes = new byte[40];
       try (UnbufferedReadableByteChannel c = session.open()) {
@@ -151,12 +152,11 @@ public final class ITGapicUnbufferedReadableByteChannelTest {
               (start, resultFuture) ->
                   new GapicUnbufferedReadableByteChannel(
                       resultFuture,
-                      storageClient
-                          .readObjectCallable()
-                          .withDefaultCallContext(
-                              contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
+                      storageClient.readObjectCallable(),
                       start,
                       Hasher.noop(),
+                      server.getGrpcStorageOptions(),
+                      retryOnly(DataLossException.class),
                       ResponseContentLifecycleManager.noop()));
       byte[] actualBytes = new byte[40];
       ImmutableList<ByteBuffer> buffers = TestUtils.subDivide(actualBytes, 2);
@@ -210,12 +210,11 @@ public final class ITGapicUnbufferedReadableByteChannelTest {
               (start, resultFuture) ->
                   new GapicUnbufferedReadableByteChannel(
                       resultFuture,
-                      storageClient
-                          .readObjectCallable()
-                          .withDefaultCallContext(
-                              contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
+                      storageClient.readObjectCallable(),
                       start,
                       Hasher.noop(),
+                      server.getGrpcStorageOptions(),
+                      retryOnly(DataLossException.class),
                       ResponseContentLifecycleManager.noop()));
       byte[] actualBytes = new byte[40];
       try (UnbufferedReadableByteChannel c = session.open()) {
@@ -258,12 +257,11 @@ public final class ITGapicUnbufferedReadableByteChannelTest {
               (start, resultFuture) ->
                   new GapicUnbufferedReadableByteChannel(
                       resultFuture,
-                      storageClient
-                          .readObjectCallable()
-                          .withDefaultCallContext(
-                              contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
+                      storageClient.readObjectCallable(),
                       start,
                       Hasher.enabled(),
+                      server.getGrpcStorageOptions(),
+                      retryOnly(DataLossException.class),
                       ResponseContentLifecycleManager.noop()));
       byte[] actualBytes = new byte[40];
       try (UnbufferedReadableByteChannel c = session.open()) {
@@ -298,12 +296,11 @@ public final class ITGapicUnbufferedReadableByteChannelTest {
               (start, resultFuture) ->
                   new GapicUnbufferedReadableByteChannel(
                       resultFuture,
-                      storageClient
-                          .readObjectCallable()
-                          .withDefaultCallContext(
-                              contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
+                      storageClient.readObjectCallable(),
                       start,
                       Hasher.enabled(),
+                      server.getGrpcStorageOptions(),
+                      retryOnly(DataLossException.class),
                       ResponseContentLifecycleManager.noop()));
       byte[] actualBytes = new byte[41];
       //noinspection resource
@@ -316,4 +313,14 @@ public final class ITGapicUnbufferedReadableByteChannelTest {
       assertThrows(ClosedChannelException.class, () -> c.read(buf));
     }
   }
+
+  private static <E extends ApiException> ResultRetryAlgorithm<?> retryOnly(Class<E> c) {
+    return new BasicResultRetryAlgorithm<java.lang.Object>() {
+      @Override
+      public boolean shouldRetry(Throwable previousThrowable, java.lang.Object previousResponse) {
+        return previousThrowable instanceof StorageException
+            && c.isAssignableFrom(previousThrowable.getCause().getClass());
+      }
+    };
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGzipReadableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGzipReadableByteChannelTest.java
index 41f6df1d..bfde3e4c 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGzipReadableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGzipReadableByteChannelTest.java
@@ -126,6 +126,8 @@ public class ITGzipReadableByteChannelTest {
               .read()
               .byteChannel(
                   storageClient.getInstance().readObjectCallable(),
+                  fakeServer.getInstance().getGrpcStorageOptions(),
+                  StorageRetryStrategy.getDefaultStorageRetryStrategy().getIdempotentHandler(),
                   ResponseContentLifecycleManager.noop())
               .setHasher(Hasher.noop())
               .setAutoGzipDecompression(true)
@@ -147,6 +149,8 @@ public class ITGzipReadableByteChannelTest {
               .read()
               .byteChannel(
                   storageClient.getInstance().readObjectCallable(),
+                  fakeServer.getInstance().getGrpcStorageOptions(),
+                  StorageRetryStrategy.getDefaultStorageRetryStrategy().getIdempotentHandler(),
                   ResponseContentLifecycleManager.noop())
               .setHasher(Hasher.noop())
               .setAutoGzipDecompression(false)
@@ -199,6 +203,8 @@ public class ITGzipReadableByteChannelTest {
               .read()
               .byteChannel(
                   storageClient.getInstance().readObjectCallable(),
+                  fakeServer.getInstance().getGrpcStorageOptions(),
+                  StorageRetryStrategy.getDefaultStorageRetryStrategy().getIdempotentHandler(),
                   ResponseContentLifecycleManager.noop())
               .setHasher(Hasher.noop())
               .setAutoGzipDecompression(true)
@@ -220,6 +226,8 @@ public class ITGzipReadableByteChannelTest {
               .read()
               .byteChannel(
                   storageClient.getInstance().readObjectCallable(),
+                  fakeServer.getInstance().getGrpcStorageOptions(),
+                  StorageRetryStrategy.getDefaultStorageRetryStrategy().getIdempotentHandler(),
                   ResponseContentLifecycleManager.noop())
               .setHasher(Hasher.noop())
               .setAutoGzipDecompression(false)
@@ -241,6 +249,8 @@ public class ITGzipReadableByteChannelTest {
               .read()
               .byteChannel(
                   storageClient.getInstance().readObjectCallable(),
+                  fakeServer.getInstance().getGrpcStorageOptions(),
+                  StorageRetryStrategy.getDefaultStorageRetryStrategy().getIdempotentHandler(),
                   ResponseContentLifecycleManager.noop())
               .setHasher(Hasher.noop())
               .unbuffered()
@@ -324,7 +334,11 @@ public class ITGzipReadableByteChannelTest {
         ReadableByteChannelSession<?, Object> session =
             ResumableMedia.gapic()
                 .read()
-                .byteChannel(sc.readObjectCallable(), ResponseContentLifecycleManager.noop())
+                .byteChannel(
+                    sc.readObjectCallable(),
+                    fakeServer.getGrpcStorageOptions(),
+                    StorageRetryStrategy.getDefaultStorageRetryStrategy().getIdempotentHandler(),
+                    ResponseContentLifecycleManager.noop())
                 .setHasher(Hasher.noop())
                 .setAutoGzipDecompression(true)
                 .unbuffered()
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITStorageLifecycleTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITStorageLifecycleTest.java
index 1c404ec9..4b4e497c 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITStorageLifecycleTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITStorageLifecycleTest.java
@@ -51,6 +51,7 @@ public final class ITStorageLifecycleTest {
             .setCredentials(NoCredentials.getInstance())
             .setProjectId("test-project-id")
             .setEnableGrpcClientMetrics(false)
+            .setAttemptDirectPath(false)
             .build();
 
     Storage service1 = options.getService();
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
index 5efbae9f..92016330 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
@@ -20,14 +20,12 @@ import static java.util.Objects.requireNonNull;
 
 import com.google.api.core.ApiClock;
 import com.google.api.core.NanoClock;
-import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.grpc.GrpcStatusCode;
 import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
 import com.google.api.gax.retrying.RetrySettings;
 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.RetryHelper;
 import com.google.cloud.RetryHelper.RetryHelperException;
 import com.google.cloud.http.BaseHttpServiceException;
@@ -113,10 +111,6 @@ public final class TestUtils {
         statusRuntimeException, GrpcStatusCode.of(code), true, errorDetails);
   }
 
-  public static GrpcCallContext contextWithRetryForCodes(StatusCode.Code... code) {
-    return GrpcCallContext.createDefault().withRetryableCodes(ImmutableSet.copyOf(code));
-  }
-
   public static ImmutableList<ByteBuffer> subDivide(byte[] bytes, int division) {
     int length = bytes.length;
     int fullDivisions = length / division;
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RetryTestFixture.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RetryTestFixture.java
index 3076b171..38c1f899 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RetryTestFixture.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RetryTestFixture.java
@@ -160,7 +160,8 @@ final class RetryTestFixture extends TestWatcher {
           builder =
               StorageOptions.grpc()
                   .setHost(testBench.getGRPCBaseUri())
-                  .setEnableGrpcClientMetrics(false);
+                  .setEnableGrpcClientMetrics(false)
+                  .setAttemptDirectPath(false);
           break;
         default:
           throw new IllegalStateException(
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcIdempotencyTokenTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcIdempotencyTokenTest.java
index 0aca897a..f6cfb5a3 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcIdempotencyTokenTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcIdempotencyTokenTest.java
@@ -69,6 +69,7 @@ public final class ITGrpcIdempotencyTokenTest {
         StorageOptions.grpc()
             .setGrpcInterceptorProvider(() -> ImmutableList.of(requestAuditing))
             .setEnableGrpcClientMetrics(false)
+            .setAttemptDirectPath(false)
             .build()
             .getService();
   }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITStorageReadChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITStorageReadChannelTest.java
index 5c6bb4ef..ced70f71 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITStorageReadChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITStorageReadChannelTest.java
@@ -117,7 +117,7 @@ public final class ITStorageReadChannelTest {
 
   @Test
   public void storageReadChannel_shouldAllowDisablingBufferingBySettingChunkSize_lteq0()
-      throws IOException {
+      throws Exception {
     int _512KiB = 512 * 1024;
     int _1MiB = 1024 * 1024;
 
@@ -130,6 +130,7 @@ public final class ITStorageReadChannelTest {
     }
 
     try (ReadChannel c = storage.reader(info.getBlobId())) {
+      ApiFuture<BlobInfo> infoFuture = getBlobInfoFromReadChannelFunction(c);
       c.setChunkSize(0);
 
       ByteBuffer buf = ByteBuffer.allocate(_1MiB);
@@ -140,6 +141,8 @@ public final class ITStorageReadChannelTest {
       String actual = xxd(buf);
       String expected = xxd(uncompressedBytes);
       assertThat(actual).isEqualTo(expected);
+      BlobInfo blobInfo = infoFuture.get(3, TimeUnit.SECONDS);
+      assertThat(blobInfo.getBlobId()).isEqualTo(info.getBlobId());
     }
   }
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/StorageNativeCanary.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/StorageNativeCanary.java
index 534764d9..651f0124 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/StorageNativeCanary.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/StorageNativeCanary.java
@@ -57,7 +57,12 @@ public final class StorageNativeCanary {
 
   @Test
   public void canary_happyPath_grpc() throws Exception {
-    assertBehaviorOfPrimaryStorageActions(StorageOptions.grpc().build().getService());
+    assertBehaviorOfPrimaryStorageActions(
+        StorageOptions.grpc()
+            // TODO https://github.com/googleapis/java-storage/issues/2716
+            .setAttemptDirectPath(false)
+            .build()
+            .getService());
   }
 
   /**
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java
index 9739abcc..28a7b877 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java
@@ -118,6 +118,7 @@ final class BackendResources implements ManagedLifecycle {
                       .setGrpcInterceptorProvider(
                           GrpcPlainRequestLoggingInterceptor.getInterceptorProvider())
                       .setEnableGrpcClientMetrics(false)
+                      .setAttemptDirectPath(false)
                       .build();
               return new StorageInstance(built, protectedBucketNames);
             });