fix: update 416 handling for ReadChannel (#3018)

If a 416 or OUT_OF_RANGE error are encountered the stream will always return -1, even if a new generation of object is created.

Author: BenWhitehead

google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedReadableByteChannel.java

@@ -222,6 +222,8 @@ private ScatteringByteChannel open() {
         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) {

google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannel.java

@@ -22,6 +22,7 @@
 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 @@ protected void onResponseImpl(ReadObjectResponse response) {
 
     @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);
       }

google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java

@@ -46,6 +46,7 @@
 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 void responseWith416ReturnsZeroAndLeavesTheChannelOpen() throws IOExcepti
     }
   }
 
+  @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 {