Repository: googleapis/java-storage
Pull request #3021 (2025-04-03 23:00:20)

chore: improve read object range error handling and retry robustness
When a read causes an error it is delivered as a stream error. Depending on the kind of error (DATA_LOSS OUT_OF_RANGE) this could lead to a terminal state for the stream. If the error is for a specific range(s) mark those individual reads as terminally failed, but do not fail any others. Instead, raise a client side retryable ABORTED to restart the stream and any remaining reads for it.




Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedReadableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedReadableByteChannel.java
index b8879d18..188f53a0 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedReadableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedReadableByteChannel.java
@@ -222,6 +222,8 @@ class ApiaryUnbufferedReadableByteChannel implements UnbufferedReadableByteChann
         if (statusCode == 404) {
           throw new StorageException(404, "Failure while trying to resume download", e);
         }
+      } else if (e.getStatusCode() == 416) {
+        returnEOF = true;
       }
       throw StorageException.translate(e);
     } catch (IOException e) {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/AsyncSessionClosedException.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/AsyncSessionClosedException.java
new file mode 100644
index 00000000..982796f1
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/AsyncSessionClosedException.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright 2025 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.api.core.BetaApi;
+
+/**
+ * Root exception for async tasks which fail due to a session being closed.
+ *
+ * @see BlobReadSession
+ */
+@BetaApi
+public final class AsyncSessionClosedException extends RuntimeException {
+  AsyncSessionClosedException(String msg) {
+    super(msg);
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadSession.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadSession.java
index 41c80bb5..eb424fcd 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadSession.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadSession.java
@@ -62,7 +62,7 @@ public interface BlobReadSession extends AutoCloseable, Closeable {
    *
    * <p>If a projection is not fully consumed/resolved it will be transitioned to a failed state.
    *
-   * <p>This method MUST be closed to ensure cleanup of any inflight buffers, and to avoid a memory
+   * <p>This method MUST be called to ensure cleanup of any inflight buffers, and to avoid a memory
    * leak.
    *
    * @since 2.51.0 This new api is in preview and is subject to breaking changes.
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 9d1863f7..8f33b1e5 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
@@ -22,6 +22,7 @@ import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.rpc.ApiExceptions;
+import com.google.api.gax.rpc.OutOfRangeException;
 import com.google.api.gax.rpc.StateCheckingResponseObserver;
 import com.google.api.gax.rpc.StreamController;
 import com.google.api.gax.rpc.WatchdogTimeoutException;
@@ -365,6 +366,15 @@ final class GapicUnbufferedReadableByteChannel
 
     @Override
     protected void onErrorImpl(Throwable t) {
+      if (t instanceof OutOfRangeException) {
+        try {
+          queue.offer(EOF_MARKER);
+          open.set(null);
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+          throw Code.ABORTED.toStatus().withCause(e).asRuntimeException();
+        }
+      }
       if (t instanceof CancellationException) {
         cancellation.set(t);
       }
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 0d76d001..352843f7 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
@@ -36,6 +36,7 @@ import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
 import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.retrying.RetrySettings;
+import com.google.api.gax.rpc.AbortedException;
 import com.google.api.gax.rpc.BidiStreamingCallable;
 import com.google.api.gax.rpc.ClientContext;
 import com.google.api.gax.rpc.HeaderProvider;
@@ -1309,11 +1310,12 @@ public final class GrpcStorageOptions extends StorageOptions
     }
 
     @Override
-    public boolean shouldRetry(Throwable previousThrowable, Object previousResponse) {
+    public boolean shouldRetry(Throwable t, Object previousResponse) {
       // this is only retryable with read object range, not other requests
-      return previousThrowable instanceof UncheckedChecksumMismatchException
-          || previousThrowable instanceof OutOfRangeException
-          || delegate.shouldRetry(StorageException.coalesce(previousThrowable), null);
+      return t instanceof UncheckedChecksumMismatchException
+          || (t instanceof OutOfRangeException && ((OutOfRangeException) t).isRetryable())
+          || (t instanceof AbortedException && ((AbortedException) t).isRetryable())
+          || delegate.shouldRetry(StorageException.coalesce(t), null);
     }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionImpl.java
index bf387357..a4d54a17 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionImpl.java
@@ -83,7 +83,7 @@ final class ObjectReadSessionImpl implements ObjectReadSession {
   public <Projection> Projection readAs(ReadProjectionConfig<Projection> config) {
     lock.lock();
     try {
-      checkState(open, "stream already closed");
+      checkState(open, "Session already closed");
       switch (config.getType()) {
         case STREAM_READ:
           long readId = state.newReadId();
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionStream.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionStream.java
index d44d6c41..4d1c20ee 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionStream.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionStream.java
@@ -47,7 +47,6 @@ import com.google.storage.v2.ReadRangeError;
 import io.grpc.Status.Code;
 import io.grpc.StatusRuntimeException;
 import java.io.IOException;
-import java.nio.channels.AsynchronousCloseException;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
@@ -124,7 +123,7 @@ final class ObjectReadSessionStream
     }
     int updatedLeaseCount = openLeases.decrementAndGet();
     if (updatedLeaseCount == 0) {
-      AsynchronousCloseException cause = new AsynchronousCloseException();
+      AsyncSessionClosedException cause = new AsyncSessionClosedException("Session already closed");
       ApiFuture<?> f = failAll(() -> new StorageException(0, "Parent stream shutdown", cause));
       return ApiFutures.transformAsync(f, ignore -> ApiFutures.immediateFuture(null), executor);
     } else {
@@ -378,37 +377,52 @@ final class ObjectReadSessionStream
     @Override
     public void onError(Throwable t) {
       requestStream = null;
-      try {
-        BidiReadObjectError error = GrpcUtils.getBidiReadObjectError(t);
-        if (error == null) {
-          return;
-        }
+      BidiReadObjectError error = GrpcUtils.getBidiReadObjectError(t);
+      if (error == null) {
+        // if there isn't a BidiReadObjectError that may contain more narrow failures, propagate
+        // the failure as is to the stream.
+        streamRetryContext.recordError(
+            t, ObjectReadSessionStream.this::restart, ObjectReadSessionStream.this::failAll);
+        return;
+      }
 
-        List<ReadRangeError> rangeErrors = error.getReadRangeErrorsList();
-        if (rangeErrors.isEmpty()) {
-          return;
-        }
-        for (ReadRangeError rangeError : rangeErrors) {
-          Status status = rangeError.getStatus();
-          long id = rangeError.getReadId();
-          ObjectReadSessionStreamRead<?> read = state.getOutstandingRead(id);
-          if (read == null) {
-            continue;
-          }
-          // mark read as failed, but don't resolve its future now. Schedule the delivery of the
-          // failure in executor to ensure any downstream future doesn't block this IO thread.
-          read.preFail();
-          executor.execute(
-              StorageException.liftToRunnable(
-                  () ->
-                      state
-                          .removeOutstandingReadOnFailure(id, read::fail)
-                          .onFailure(GrpcUtils.statusToApiException(status))));
-        }
-      } finally {
+      List<ReadRangeError> rangeErrors = error.getReadRangeErrorsList();
+      if (rangeErrors.isEmpty()) {
+        // if there aren't any specific read id's that contain errors, propagate the error as is to
+        // the stream.
         streamRetryContext.recordError(
             t, ObjectReadSessionStream.this::restart, ObjectReadSessionStream.this::failAll);
+        return;
+      }
+      for (ReadRangeError rangeError : rangeErrors) {
+        Status status = rangeError.getStatus();
+        long id = rangeError.getReadId();
+        ObjectReadSessionStreamRead<?> read = state.getOutstandingRead(id);
+        if (read == null) {
+          continue;
+        }
+        // mark read as failed, but don't resolve its future now. Schedule the delivery of the
+        // failure in executor to ensure any downstream future doesn't block this IO thread.
+        read.preFail();
+        executor.execute(
+            StorageException.liftToRunnable(
+                () ->
+                    state
+                        .removeOutstandingReadOnFailure(id, read::fail)
+                        .onFailure(GrpcUtils.statusToApiException(status))));
       }
+      // now that we've failed specific reads, raise a retryable ABORTED error to the stream to
+      // cause it to retry and pending remaining reads.
+      ApiException apiException =
+          ApiExceptionFactory.createException(
+              "Stream error, reclassifying as ABORTED for reads not specified in BidiReadObjectError",
+              t,
+              GrpcStatusCode.of(Code.ABORTED),
+              true);
+      streamRetryContext.recordError(
+          apiException,
+          ObjectReadSessionStream.this::restart,
+          ObjectReadSessionStream.this::failAll);
     }
 
     private OnSuccess restartReadFromCurrentOffset(long id) {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadAsFutureBytes.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadAsFutureBytes.java
index dcea76b9..722b01bc 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadAsFutureBytes.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadAsFutureBytes.java
@@ -37,7 +37,8 @@ import javax.annotation.concurrent.Immutable;
  */
 @BetaApi
 @Immutable
-final class ReadAsFutureBytes extends BaseConfig<ApiFuture<byte[]>, AccumulatingRead<byte[]>> {
+public final class ReadAsFutureBytes
+    extends BaseConfig<ApiFuture<byte[]>, AccumulatingRead<byte[]>> {
 
   static final ReadAsFutureBytes INSTANCE =
       new ReadAsFutureBytes(RangeSpec.all(), Hasher.enabled());
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadProjectionConfigs.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadProjectionConfigs.java
index 212519d7..1156cf7e 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadProjectionConfigs.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadProjectionConfigs.java
@@ -83,7 +83,7 @@ public final class ReadProjectionConfigs {
    * @since 2.51.0 This new api is in preview and is subject to breaking changes.
    */
   @BetaApi
-  static ReadAsFutureBytes asFutureBytes() {
+  public static ReadAsFutureBytes asFutureBytes() {
     return ReadAsFutureBytes.INSTANCE;
   }
 
@@ -104,7 +104,7 @@ public final class ReadProjectionConfigs {
    * @since 2.51.0 This new api is in preview and is subject to breaking changes.
    */
   @BetaApi
-  static ReadAsFutureByteString asFutureByteString() {
+  public static ReadAsFutureByteString asFutureByteString() {
     return ReadAsFutureByteString.INSTANCE;
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
index 927960ca..6bd3bfc6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
@@ -177,8 +177,8 @@ public final class StorageException extends BaseHttpServiceException {
   }
 
   private static void attachErrorDetails(ApiException ae) {
-    if (ae != null && ae.getErrorDetails() != null) {
-      final StringBuilder sb = new StringBuilder();
+    if (ae != null && ae.getErrorDetails() != null && !errorDetailsAttached(ae)) {
+      StringBuilder sb = new StringBuilder();
       ErrorDetails ed = ae.getErrorDetails();
       sb.append("ErrorDetails {\n");
       Stream.of(
@@ -202,6 +202,16 @@ public final class StorageException extends BaseHttpServiceException {
     }
   }
 
+  private static boolean errorDetailsAttached(ApiException ae) {
+    Throwable[] suppressed = ae.getSuppressed();
+    for (Throwable throwable : suppressed) {
+      if (throwable instanceof ApiExceptionErrorDetailsComment) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   /**
    * Translate IOException to a StorageException representing the cause of the error. This method
    * defaults to idempotent always being {@code true}. Additionally, this method translates
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ZeroCopySupport.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ZeroCopySupport.java
index eaeede2e..8ac0ed61 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ZeroCopySupport.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ZeroCopySupport.java
@@ -16,22 +16,45 @@
 
 package com.google.cloud.storage;
 
-import com.google.api.core.InternalApi;
+import com.google.api.core.BetaApi;
 import com.google.api.core.InternalExtensionOnly;
 import com.google.protobuf.ByteString;
 import java.io.Closeable;
 import java.io.IOException;
 
-@InternalApi
-@InternalExtensionOnly
-interface ZeroCopySupport {
+/**
+ * Public components which exist to support zero-copy data access
+ *
+ * @since 2.51.0 This new api is in preview and is subject to breaking changes.
+ */
+@BetaApi
+public abstract class ZeroCopySupport {
+
+  private ZeroCopySupport() {}
 
-  @InternalApi
+  /**
+   * Represents an object that can be accessed as a {@link ByteString}, but has a lifecycle that
+   * requires being explicitly closed in order to free up resources.
+   *
+   * <p>Instances of this class should be used in a try-with-resources to ensure they are released.
+   *
+   * <pre>{@code
+   * try (DisposableByteString disposableByteString = ...) {
+   *   System.out.println(disposableByteString.byteString().size());
+   * }
+   * }</pre>
+   *
+   * @see ReadProjectionConfigs#asFutureByteString()
+   * @since 2.51.0 This new api is in preview and is subject to breaking changes.
+   */
+  @BetaApi
   @InternalExtensionOnly
-  interface DisposableByteString extends AutoCloseable, Closeable {
+  public interface DisposableByteString extends AutoCloseable, Closeable {
 
+    /** Get the ByteString representation of the underlying resources */
     ByteString byteString();
 
+    /** Signal the underlying resources that they can be released. */
     @Override
     void close() throws IOException;
   }


Test patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadAsFutureByteString.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadAsFutureByteString.java
index fe283a34..aa7eee53 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadAsFutureByteString.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadAsFutureByteString.java
@@ -42,7 +42,7 @@ import javax.annotation.concurrent.Immutable;
  */
 @BetaApi
 @Immutable
-final class ReadAsFutureByteString
+public final class ReadAsFutureByteString
     extends BaseConfig<ApiFuture<DisposableByteString>, AccumulatingRead<DisposableByteString>> {
 
   static final ReadAsFutureByteString INSTANCE =
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java
index fdc60c7a..d07ee041 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java
@@ -53,9 +53,11 @@ import com.google.common.hash.Hashing;
 import com.google.common.io.BaseEncoding;
 import com.google.common.io.ByteStreams;
 import com.google.common.truth.Correspondence;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.protobuf.Any;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.TextFormat;
+import com.google.rpc.DebugInfo;
 import com.google.storage.v2.BidiReadHandle;
 import com.google.storage.v2.BidiReadObjectError;
 import com.google.storage.v2.BidiReadObjectRedirectedError;
@@ -77,8 +79,8 @@ import io.grpc.StatusRuntimeException;
 import io.grpc.protobuf.ProtoUtils;
 import io.grpc.stub.StreamObserver;
 import java.io.ByteArrayOutputStream;
+import java.io.IOException;
 import java.nio.ByteBuffer;
-import java.nio.channels.AsynchronousCloseException;
 import java.nio.channels.Channels;
 import java.nio.channels.ScatteringByteChannel;
 import java.nio.channels.WritableByteChannel;
@@ -87,13 +89,20 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -1045,11 +1054,13 @@ public final class ITObjectReadSessionFakeTest {
               assertAll(
                   () -> assertThat(se2).isNotSameInstanceAs(se3),
                   () ->
-                      assertThat(se2).hasCauseThat().isInstanceOf(AsynchronousCloseException.class),
+                      assertThat(se2)
+                          .hasCauseThat()
+                          .isInstanceOf(AsyncSessionClosedException.class),
                   () ->
                       assertThat(se3)
                           .hasCauseThat()
-                          .isInstanceOf(AsynchronousCloseException.class));
+                          .isInstanceOf(AsyncSessionClosedException.class));
             });
       }
     }
@@ -1375,9 +1386,6 @@ public final class ITObjectReadSessionFakeTest {
                     .build())
             .build();
 
-    System.out.println("req1 = " + TextFormat.printer().shortDebugString(req1));
-    System.out.println("req2 = " + TextFormat.printer().shortDebugString(req2));
-    System.out.println("req3 = " + TextFormat.printer().shortDebugString(req3));
     ImmutableMap<BidiReadObjectRequest, BidiReadObjectResponse> db =
         ImmutableMap.<BidiReadObjectRequest, BidiReadObjectResponse>builder()
             .put(req1, res1)
@@ -1537,6 +1545,196 @@ public final class ITObjectReadSessionFakeTest {
     }
   }
 
+  /**
+   * Define a test where multiple reads for the same session will be performed, and some of those
+   * reads cause OUT_OF_RANGE errors.
+   *
+   * <p>An OUT_OF_RANGE error is delivered as a stream level status, which means any reads which
+   * share a stream must be restarted while the read that caused the OUT_OF_RANGE should be failed.
+   *
+   * <p>Verify this behavior for both channel based and future byte[] based.
+   */
+  @Test
+  public void serverOutOfRangeIsNotRetried() throws Exception {
+    ChecksummedTestContent expected = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 10, 20);
+    Semaphore sem = new Semaphore(1);
+
+    BidiReadObjectResponse dataResp =
+        BidiReadObjectResponse.newBuilder()
+            .addObjectDataRanges(
+                ObjectRangeData.newBuilder()
+                    .setChecksummedData(expected.asChecksummedData())
+                    .setReadRange(getReadRange(0, 10, 20))
+                    .setRangeEnd(true)
+                    .build())
+            .build();
+
+    AtomicInteger bidiReadObjectCount = new AtomicInteger();
+    ExecutorService exec =
+        Executors.newCachedThreadPool(
+            new ThreadFactoryBuilder().setDaemon(true).setNameFormat("test-server-%d").build());
+
+    StorageImplBase fakeStorage =
+        new StorageImplBase() {
+          @Override
+          public StreamObserver<BidiReadObjectRequest> bidiReadObject(
+              StreamObserver<BidiReadObjectResponse> respond) {
+            bidiReadObjectCount.getAndIncrement();
+            return new StreamObserver<BidiReadObjectRequest>() {
+              @Override
+              public void onNext(BidiReadObjectRequest request) {
+                if (request.equals(REQ_OPEN)) {
+                  respond.onNext(RES_OPEN);
+                } else if (request.getReadRangesList().get(0).getReadOffset() == 10) {
+                  exec.submit(
+                      () -> {
+                        try {
+                          sem.acquire();
+                          BidiReadObjectResponse.Builder b = dataResp.toBuilder();
+                          ReadRange readRange = request.getReadRangesList().get(0);
+                          ObjectRangeData.Builder bb = dataResp.getObjectDataRanges(0).toBuilder();
+                          bb.getReadRangeBuilder().setReadId(readRange.getReadId());
+                          b.setObjectDataRanges(0, bb.build());
+                          respond.onNext(b.build());
+                        } catch (InterruptedException e) {
+                          respond.onError(
+                              TestUtils.apiException(Code.UNIMPLEMENTED, e.getMessage()));
+                        }
+                      });
+                } else if (bidiReadObjectCount.getAndIncrement() >= 1) {
+                  Optional<ReadRange> readRange = request.getReadRangesList().stream().findFirst();
+                  String message =
+                      String.format(
+                          Locale.US,
+                          "OUT_OF_RANGE read_offset = %d",
+                          readRange.map(ReadRange::getReadOffset).orElse(0L));
+                  long readId = readRange.map(ReadRange::getReadId).orElse(0L);
+
+                  BidiReadObjectError err2 =
+                      BidiReadObjectError.newBuilder()
+                          .addReadRangeErrors(
+                              ReadRangeError.newBuilder()
+                                  .setReadId(readId)
+                                  .setStatus(
+                                      com.google.rpc.Status.newBuilder()
+                                          .setCode(com.google.rpc.Code.OUT_OF_RANGE_VALUE)
+                                          .build())
+                                  .build())
+                          .build();
+
+                  com.google.rpc.Status grpcStatusDetails =
+                      com.google.rpc.Status.newBuilder()
+                          .setCode(com.google.rpc.Code.UNAVAILABLE_VALUE)
+                          .setMessage("fail read_id: " + readId)
+                          .addDetails(Any.pack(err2))
+                          .addDetails(
+                              Any.pack(
+                                  DebugInfo.newBuilder()
+                                      .setDetail(message)
+                                      .addStackEntries(
+                                          TextFormat.printer().shortDebugString(request))
+                                      .build()))
+                          .build();
+
+                  Metadata trailers = new Metadata();
+                  trailers.put(GRPC_STATUS_DETAILS_KEY, grpcStatusDetails);
+                  StatusRuntimeException statusRuntimeException =
+                      Status.OUT_OF_RANGE.withDescription(message).asRuntimeException(trailers);
+                  respond.onError(statusRuntimeException);
+                } else {
+                  respond.onError(
+                      apiException(
+                          Code.UNIMPLEMENTED,
+                          "Unexpected request { "
+                              + TextFormat.printer().shortDebugString(request)
+                              + " }"));
+                }
+              }
+
+              @Override
+              public void onError(Throwable t) {
+                respond.onError(t);
+              }
+
+              @Override
+              public void onCompleted() {
+                respond.onCompleted();
+              }
+            };
+          }
+        };
+    try (FakeServer fakeServer = FakeServer.of(fakeStorage);
+        Storage storage = fakeServer.getGrpcStorageOptions().getService()) {
+
+      BlobId id = BlobId.of("b", "o");
+
+      try (BlobReadSession session = storage.blobReadSession(id).get(5, TimeUnit.SECONDS)) {
+
+        ApiFuture<byte[]> shouldSucceedFuture =
+            session.readAs(
+                ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(10, 20)));
+
+        ApiFuture<byte[]> shouldFailFuture =
+            session.readAs(
+                ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.beginAt(37)));
+        ExecutionException exceptionFromFuture =
+            assertThrows(
+                ExecutionException.class, () -> shouldFailFuture.get(30, TimeUnit.SECONDS));
+        sem.release();
+
+        Exception exceptionFromChannel;
+        byte[] bytesFromFuture = shouldSucceedFuture.get(30, TimeUnit.SECONDS);
+
+        AtomicReference<byte[]> bytesFromChannel = new AtomicReference<>();
+        Future<Long> asyncShouldSucceedChannel =
+            exec.submit(
+                () -> {
+                  try (ScatteringByteChannel succeed =
+                      session.readAs(
+                          ReadProjectionConfigs.asChannel().withRangeSpec(RangeSpec.of(10, 20)))) {
+                    sem.acquire();
+                    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+                    long copy = ByteStreams.copy(succeed, Channels.newChannel(baos));
+                    bytesFromChannel.set(baos.toByteArray());
+                    return copy;
+                  } catch (IOException e) {
+                    throw new RuntimeException(e);
+                  }
+                });
+
+        try (ScatteringByteChannel fail =
+            session.readAs(
+                ReadProjectionConfigs.asChannel().withRangeSpec(RangeSpec.beginAt(39)))) {
+          exceptionFromChannel =
+              assertThrows(
+                  IOException.class,
+                  () -> {
+                    int read = 0;
+                    do {
+                      read = fail.read(ByteBuffer.allocate(1));
+                    } while (read == 0);
+                  });
+          sem.release();
+        }
+        asyncShouldSucceedChannel.get(30, TimeUnit.SECONDS);
+        Exception finalExceptionFromChannel = exceptionFromChannel;
+        assertAll(
+            () ->
+                assertThat(exceptionFromFuture)
+                    .hasCauseThat()
+                    .hasCauseThat()
+                    .isInstanceOf(OutOfRangeException.class),
+            () ->
+                assertThat(finalExceptionFromChannel)
+                    .hasCauseThat()
+                    .hasCauseThat()
+                    .isInstanceOf(OutOfRangeException.class),
+            () -> assertThat(xxd(bytesFromFuture)).isEqualTo(xxd(expected.getBytes())),
+            () -> assertThat(xxd(bytesFromChannel.get())).isEqualTo(xxd(expected.getBytes())));
+      }
+    }
+  }
+
   private static void runTestAgainstFakeServer(
       FakeStorage fakeStorage, RangeSpec range, ChecksummedTestContent expected) throws Exception {
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionTest.java
index 08691300..39ec13ff 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionTest.java
@@ -25,8 +25,10 @@ import static org.junit.Assert.assertThrows;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
+import com.google.api.gax.rpc.OutOfRangeException;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.ZeroCopySupport.DisposableByteString;
 import com.google.cloud.storage.it.ChecksummedTestContent;
@@ -101,6 +103,23 @@ public final class ITObjectReadSessionTest {
     }
   }
 
+  @Test
+  public void attemptingToGetFutureOutSizeSessionFails() throws Throwable {
+    ObjectAndContent obj512KiB = objectsFixture.getObj512KiB();
+    BlobId blobId = obj512KiB.getInfo().getBlobId();
+
+    ApiFuture<byte[]> future;
+    try (BlobReadSession session = storage.blobReadSession(blobId).get(30, TimeUnit.SECONDS)) {
+      future = session.readAs(ReadProjectionConfigs.asFutureBytes());
+    }
+
+    ExecutionException ee =
+        assertThrows(ExecutionException.class, () -> future.get(1, TimeUnit.SECONDS));
+
+    assertThat(ee).hasCauseThat().isInstanceOf(StorageException.class);
+    assertThat(ee).hasCauseThat().hasCauseThat().isInstanceOf(AsyncSessionClosedException.class);
+  }
+
   @Test
   public void lotsOfBytes() throws Exception {
     ChecksummedTestContent testContent =
@@ -300,6 +319,37 @@ public final class ITObjectReadSessionTest {
     }
   }
 
+  @Test
+  public void outOfRange()
+      throws ExecutionException, InterruptedException, TimeoutException, IOException {
+    BlobId blobId;
+    BlobWriteSession session =
+        storage.blobWriteSession(
+            BlobInfo.newBuilder(bucket, generator.randomObjectName()).build(),
+            BlobWriteOption.doesNotExist());
+    try (WritableByteChannel upload = session.open()) {
+      upload.write(DataGenerator.base64Characters().genByteBuffer(4));
+    }
+    blobId = session.getResult().get(5, TimeUnit.SECONDS).getBlobId();
+    try (BlobReadSession blobReadSession =
+        storage.blobReadSession(blobId).get(30, TimeUnit.SECONDS)) {
+
+      BlobInfo info1 = blobReadSession.getBlobInfo();
+      assertThat(info1).isNotNull();
+
+      ReadAsFutureBytes cfg = ReadProjectionConfigs.asFutureBytes();
+
+      ApiFuture<byte[]> f2 = blobReadSession.readAs(cfg.withRangeSpec(RangeSpec.beginAt(5)));
+      ExecutionException ee =
+          assertThrows(ExecutionException.class, () -> f2.get(30, TimeUnit.SECONDS));
+      assertThat(ee).hasCauseThat().hasCauseThat().isInstanceOf(OutOfRangeException.class);
+
+      ApiFuture<byte[]> f1 = blobReadSession.readAs(cfg.withRangeSpec(RangeSpec.all()));
+      byte[] bytes1 = f1.get(30, TimeUnit.SECONDS);
+      assertThat(bytes1.length).isEqualTo(4);
+    }
+  }
+
   private BlobInfo create(ChecksummedTestContent content) {
     return storage.create(
         BlobInfo.newBuilder(bucket, generator.randomObjectName()).build(),
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ObjectReadSessionStreamTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ObjectReadSessionStreamTest.java
index 38790341..75fec2cb 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ObjectReadSessionStreamTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ObjectReadSessionStreamTest.java
@@ -49,7 +49,6 @@ import com.google.storage.v2.BucketName;
 import com.google.storage.v2.Object;
 import java.io.IOException;
 import java.nio.ByteBuffer;
-import java.nio.channels.AsynchronousCloseException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -219,19 +218,19 @@ public final class ObjectReadSessionStreamTest {
           Throwable t1 = read1.fail.get(2, TimeUnit.SECONDS);
           // t1.printStackTrace(System.out);
           assertThat(t1).isInstanceOf(StorageException.class);
-          assertThat(t1).hasCauseThat().isInstanceOf(AsynchronousCloseException.class);
+          assertThat(t1).hasCauseThat().isInstanceOf(AsyncSessionClosedException.class);
         },
         () -> {
           Throwable t2 = read2.fail.get(2, TimeUnit.SECONDS);
           // t2.printStackTrace(System.err);
           assertThat(t2).isInstanceOf(StorageException.class);
-          assertThat(t2).hasCauseThat().isInstanceOf(AsynchronousCloseException.class);
+          assertThat(t2).hasCauseThat().isInstanceOf(AsyncSessionClosedException.class);
         },
         () -> {
           Throwable t3 = read3.fail.get(2, TimeUnit.SECONDS);
           // t3.printStackTrace(System.out);
           assertThat(t3).isInstanceOf(StorageException.class);
-          assertThat(t3).hasCauseThat().isInstanceOf(AsynchronousCloseException.class);
+          assertThat(t3).hasCauseThat().isInstanceOf(AsyncSessionClosedException.class);
         });
   }
 
@@ -312,7 +311,7 @@ public final class ObjectReadSessionStreamTest {
           () -> assertThat(bytes2Close.get()).isTrue(),
           () -> assertThat(read1.acceptingBytes()).isFalse(),
           () -> assertThat(se).hasMessageThat().contains("Parent stream shutdown"),
-          () -> assertThat(se).hasCauseThat().isInstanceOf(AsynchronousCloseException.class));
+          () -> assertThat(se).hasCauseThat().isInstanceOf(AsyncSessionClosedException.class));
     }
   }
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageExceptionGrpcCompatibilityTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageExceptionGrpcCompatibilityTest.java
index 2e2246ae..4dc14ccb 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageExceptionGrpcCompatibilityTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageExceptionGrpcCompatibilityTest.java
@@ -207,6 +207,45 @@ public final class StorageExceptionGrpcCompatibilityTest {
         () -> assertThat(message).contains("\t}"));
   }
 
+  @SuppressWarnings("ThrowableNotThrown")
+  @Test
+  public void apiExceptionErrorDetails_onlyAttachedOnce() throws Exception {
+    Help help =
+        Help.newBuilder()
+            .addLinks(
+                Link.newBuilder().setDescription("link1").setUrl("https://google.com").build())
+            .build();
+    List<Any> errors = ImmutableList.of(Any.pack(help));
+    ErrorDetails errorDetails = ErrorDetails.builder().setRawErrorMessages(errors).build();
+
+    ApiException ex =
+        ApiExceptionFactory.createException(
+            Code.OUT_OF_RANGE.toStatus().asRuntimeException(),
+            GrpcStatusCode.of(Code.OUT_OF_RANGE),
+            false,
+            errorDetails);
+
+    // apply a coalesce to the exception -- similar to what a retry algorithm might do to determine
+    // retryability. This is not ideal, as it is unpure but it is the way things are today with the
+    // structure of storage exception and ApiException.
+    BaseServiceException ignore1 = StorageException.coalesce(ex);
+    BaseServiceException se = StorageException.coalesce(ex);
+
+    String message = TestUtils.messagesToText(se);
+    Printer printer = TextFormat.printer();
+    assertAll(
+        () -> assertThat(message).contains("ErrorDetails {"),
+        () -> assertThat(message).contains(printer.shortDebugString(help)),
+        () -> assertThat(message).contains("\t}"),
+        () -> {
+          // make sure the error details are only attached to the exception once
+          String str = "ErrorDetails {";
+          int indexOf1 = message.indexOf(str);
+          int indexOf2 = message.indexOf(str, indexOf1 + str.length());
+          assertThat(indexOf2).isEqualTo(-1);
+        });
+  }
+
   private void doTestCoalesce(int expectedCode, Code code) {
     Status status = code.toStatus();
     GrpcStatusCode statusCode = GrpcStatusCode.of(code);
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java
index 093b5259..cd681208 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java
@@ -46,6 +46,7 @@ import com.google.cloud.storage.it.runner.annotations.Inject;
 import com.google.cloud.storage.it.runner.registry.Generator;
 import com.google.cloud.storage.it.runner.registry.ObjectsFixture;
 import com.google.cloud.storage.it.runner.registry.ObjectsFixture.ObjectAndContent;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.io.BaseEncoding;
 import com.google.common.io.ByteStreams;
 import java.io.ByteArrayInputStream;
@@ -528,6 +529,36 @@ public final class ITBlobReadChannelTest {
     }
   }
 
+  @Test
+  public void responseWith416AttemptingToReadStartingPastTheEndOfTheObjectIsTerminallyEOF()
+      throws IOException {
+    int length = 10;
+    byte[] bytes = DataGenerator.base64Characters().genBytes(length);
+
+    BlobInfo info1 =
+        BlobInfo.newBuilder(bucket, generator.randomObjectName())
+            .setMetadata(ImmutableMap.of("gen", "1"))
+            .build();
+    Blob gen1 = storage.create(info1, bytes, BlobTargetOption.doesNotExist());
+
+    try (ReadChannel reader = storage.reader(gen1.getBlobId())) {
+      reader.seek(length + 1);
+      ByteBuffer buf = ByteBuffer.allocate(1);
+      assertThat(reader.read(buf)).isEqualTo(-1);
+      assertThat(reader.read(buf)).isEqualTo(-1);
+
+      BlobInfo update = gen1.toBuilder().setMetadata(ImmutableMap.of("gen", "2")).build();
+      BlobInfo gen2 =
+          storage.create(
+              update,
+              DataGenerator.base64Characters().genBytes(length + 2),
+              BlobTargetOption.generationMatch());
+
+      assertThat(reader.read(buf)).isEqualTo(-1);
+      assertThat(reader.read(buf)).isEqualTo(-1);
+    }
+  }
+
   /** Read channel does not consider itself closed once it returns {@code -1} from read. */
   @Test
   public void readChannelIsAlwaysOpen_willReturnNegative1UntilExplicitlyClosed() throws Exception {