Repository: googleapis/java-storage
Pull request #2304 (2023-11-15 18:51:13)

fix: fix Storage#readAllBytes to allow reading compressed bytes
Storage#readAllBytes currently ignores BlobSourceOption.shouldReturnRawInputStream(true), fix this so that the option is respected.

Update grpc transport default compression handling for each of the following methods, so it is the same behavior as HTTP:
* GrpcStorage.Impl#readAllBytes
* GrpcStorage.Impl#downloadTo(BlobInfo, Path)
* GrpcStorage.Impl#downloadTo(BlobInfo, OutputStream)

Move the tow tests from ITDownloadToTest into new ITAutomaticGzipDecompressionTest which tests compression handling for all read paths.




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 a5dd1375..5f242517 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
@@ -101,7 +101,12 @@ class ApiaryUnbufferedReadableByteChannel implements UnbufferedReadableByteChann
     long totalRead = 0;
     do {
       if (sbc == null) {
-        sbc = Retrying.run(options, resultRetryAlgorithm, this::open, Function.identity());
+        try {
+          sbc = Retrying.run(options, resultRetryAlgorithm, this::open, Function.identity());
+        } catch (StorageException e) {
+          result.setException(e);
+          throw e;
+        }
       }
 
       long totalRemaining = Buffers.totalRemaining(dsts, offset, length);
@@ -124,13 +129,17 @@ class ApiaryUnbufferedReadableByteChannel implements UnbufferedReadableByteChann
           sbc = null;
         } else if (t instanceof IOException) {
           IOException ioE = (IOException) t;
-          if (resultRetryAlgorithm.shouldRetry(StorageException.translate(ioE), null)) {
+          StorageException translate = StorageException.translate(ioE);
+          if (resultRetryAlgorithm.shouldRetry(translate, null)) {
             sbc = null;
           } else {
+            result.setException(translate);
             throw ioE;
           }
         } else {
-          throw new IOException(StorageException.coalesce(t));
+          BaseServiceException coalesce = StorageException.coalesce(t);
+          result.setException(coalesce);
+          throw new IOException(coalesce);
         }
       } finally {
         long totalRemainingAfter = Buffers.totalRemaining(dsts, offset, length);
@@ -207,20 +216,17 @@ class ApiaryUnbufferedReadableByteChannel implements UnbufferedReadableByteChann
       if (xGoogGeneration != null) {
         int statusCode = e.getStatusCode();
         if (statusCode == 404) {
-          throw new StorageException(404, "Failure while trying to resume download", e);
+          StorageException storageException =
+              new StorageException(404, "Failure while trying to resume download", e);
+          result.setException(storageException);
+          throw storageException;
         }
       }
-      StorageException translate = StorageException.translate(e);
-      result.setException(translate);
-      throw translate;
+      throw StorageException.translate(e);
     } catch (IOException e) {
-      StorageException translate = StorageException.translate(e);
-      result.setException(translate);
-      throw translate;
+      throw StorageException.translate(e);
     } catch (Throwable t) {
-      BaseServiceException coalesce = StorageException.coalesce(t);
-      result.setException(coalesce);
-      throw coalesce;
+      throw StorageException.coalesce(t);
     }
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java
index e8143871..64bccd4a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java
@@ -44,10 +44,7 @@ final class BlobReadChannelV2 extends BaseStorageReadChannel<StorageObject> {
     this.opts = opts;
     this.blobReadChannelContext = blobReadChannelContext;
     this.autoGzipDecompression =
-        // RETURN_RAW_INPUT_STREAM means do not add GZIPInputStream to the pipeline. Meaning, if
-        // RETURN_RAW_INPUT_STREAM is false, automatically attempt to decompress if Content-Encoding
-        // gzip.
-        Boolean.FALSE.equals(opts.get(StorageRpc.Option.RETURN_RAW_INPUT_STREAM));
+        Utils.isAutoGzipDecompression(opts, /*defaultWhenUndefined=*/ false);
   }
 
   @Override
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 d4c80596..2a21ddcf 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
@@ -653,7 +653,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
 
   @Override
   public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) {
-    UnbufferedReadableByteChannelSession<Object> session = unbufferedReadSession(blob, options);
+    UnbufferedReadableByteChannelSession<Object> session =
+        unbufferedDefaultAutoGzipDecompressingReadSession(blob, options);
 
     ByteArrayOutputStream baos = new ByteArrayOutputStream();
     try (UnbufferedReadableByteChannel r = session.open();
@@ -681,16 +682,19 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     ReadObjectRequest request = getReadObjectRequest(blob, opts);
     Set<StatusCode.Code> codes = resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(request));
     GrpcCallContext grpcCallContext = Retrying.newCallContext().withRetryableCodes(codes);
+    boolean autoGzipDecompression =
+        Utils.isAutoGzipDecompression(opts, /*defaultWhenUndefined=*/ false);
     return new GrpcBlobReadChannel(
         storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext),
         request,
-        !opts.autoGzipDecompression());
+        autoGzipDecompression);
   }
 
   @Override
   public void downloadTo(BlobId blob, Path path, BlobSourceOption... options) {
 
-    UnbufferedReadableByteChannelSession<Object> session = unbufferedReadSession(blob, options);
+    UnbufferedReadableByteChannelSession<Object> session =
+        unbufferedDefaultAutoGzipDecompressingReadSession(blob, options);
 
     try (UnbufferedReadableByteChannel r = session.open();
         WritableByteChannel w = Files.newByteChannel(path, WRITE_OPS)) {
@@ -703,7 +707,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
   @Override
   public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption... options) {
 
-    UnbufferedReadableByteChannelSession<Object> session = unbufferedReadSession(blob, options);
+    UnbufferedReadableByteChannelSession<Object> session =
+        unbufferedDefaultAutoGzipDecompressingReadSession(blob, options);
 
     try (UnbufferedReadableByteChannel r = session.open();
         WritableByteChannel w = Channels.newChannel(outputStream)) {
@@ -1801,18 +1806,20 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     return opts.writeObjectRequest().apply(requestBuilder).build();
   }
 
-  private UnbufferedReadableByteChannelSession<Object> unbufferedReadSession(
-      BlobId blob, BlobSourceOption[] options) {
+  private UnbufferedReadableByteChannelSession<Object>
+      unbufferedDefaultAutoGzipDecompressingReadSession(BlobId blob, BlobSourceOption[] options) {
     Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob).prepend(defaultOpts);
     ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts);
     Set<StatusCode.Code> codes =
         resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(readObjectRequest));
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(Retrying.newCallContext().withRetryableCodes(codes));
+    boolean autoGzipDecompression =
+        Utils.isAutoGzipDecompression(opts, /*defaultWhenUndefined=*/ true);
     return ResumableMedia.gapic()
         .read()
         .byteChannel(storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext))
-        .setAutoGzipDecompression(!opts.autoGzipDecompression())
+        .setAutoGzipDecompression(autoGzipDecompression)
         .unbuffered()
         .setReadObjectRequest(readObjectRequest)
         .build();
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GzipReadableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GzipReadableByteChannel.java
index 967577ed..19ab980f 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GzipReadableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GzipReadableByteChannel.java
@@ -17,6 +17,7 @@
 package com.google.cloud.storage;
 
 import com.google.api.core.ApiFuture;
+import com.google.api.gax.rpc.ApiExceptions;
 import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
 import java.io.ByteArrayInputStream;
 import java.io.FilterInputStream;
@@ -27,7 +28,6 @@ import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.ScatteringByteChannel;
-import java.util.concurrent.ExecutionException;
 import java.util.zip.GZIPInputStream;
 
 final class GzipReadableByteChannel implements UnbufferedReadableByteChannel {
@@ -60,7 +60,7 @@ final class GzipReadableByteChannel implements UnbufferedReadableByteChannel {
       source.read(wrap);
       try {
         // Step 2: wait for the object metadata, this is populated in the first message from GCS
-        String contentEncoding = this.contentEncoding.get();
+        String contentEncoding = ApiExceptions.callAndTranslateApiException(this.contentEncoding);
         // if the Content-Encoding is gzip, Step 3: wire gzip decompression into the byte path
         //   this will have a copy impact as we are no longer controlling all the buffers
         if ("gzip".equals(contentEncoding) || "x-gzip".equals(contentEncoding)) {
@@ -86,7 +86,9 @@ final class GzipReadableByteChannel implements UnbufferedReadableByteChannel {
           bytesRead += Buffers.copy(wrap, dsts, offset, length);
           delegate = source;
         }
-      } catch (InterruptedException | ExecutionException e) {
+      } catch (StorageException se) {
+        throw se;
+      } catch (Exception e) {
         throw new IOException(e);
       }
     }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
index 9e0be115..99a05342 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
@@ -2301,18 +2301,6 @@ final class UnifiedOpts {
       return fuseMappers(ObjectTargetOpt.class, ObjectTargetOpt::blobInfo);
     }
 
-    /**
-     * Here for compatibility. This should NOT be an "Opt" instead an attribute of the channel
-     * builder. When {@link ReturnRawInputStream} is removed, this method should be removed as well.
-     *
-     * @see
-     *     GapicDownloadSessionBuilder.ReadableByteChannelSessionBuilder#setAutoGzipDecompression(boolean)
-     */
-    @Deprecated
-    boolean autoGzipDecompression() {
-      return filterTo(ReturnRawInputStream.class).findFirst().map(r -> r.val).orElse(true);
-    }
-
     Decoder<BlobInfo, BlobInfo> clearBlobFields() {
       return filterTo(Fields.class)
           .findFirst()
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java
index c24a68d4..b51d5a39 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java
@@ -24,6 +24,8 @@ import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.rpc.ApiCallContext;
 import com.google.cloud.storage.Conversions.Codec;
 import com.google.cloud.storage.UnifiedOpts.NamedField;
+import com.google.cloud.storage.UnifiedOpts.Opts;
+import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.MapDifference;
 import com.google.common.collect.Maps;
@@ -310,4 +312,28 @@ final class Utils {
   static GrpcCallContext merge(@NonNull GrpcCallContext l, @NonNull GrpcCallContext r) {
     return (GrpcCallContext) l.merge(r);
   }
+
+  /**
+   * RETURN_RAW_INPUT_STREAM means do not add GZIPInputStream to the pipeline. Meaning, if
+   * RETURN_RAW_INPUT_STREAM is false, automatically attempt to decompress if Content-Encoding gzip.
+   */
+  static boolean isAutoGzipDecompression(Opts<?> opts, boolean defaultWhenUndefined) {
+    return isAutoGzipDecompression(opts.getRpcOptions(), defaultWhenUndefined);
+  }
+
+  /**
+   * RETURN_RAW_INPUT_STREAM means do not add GZIPInputStream to the pipeline. Meaning, if
+   * RETURN_RAW_INPUT_STREAM is false, automatically attempt to decompress if Content-Encoding gzip.
+   */
+  static boolean isAutoGzipDecompression(
+      Map<StorageRpc.Option, ?> opts, boolean defaultWhenUndefined) {
+    // Option.getBoolean is package private, and we don't want to open it.
+    // if non-null explicitly compare to a boolean value to coerce it to a boolean result
+    Object returnRawInputStream = opts.get(StorageRpc.Option.RETURN_RAW_INPUT_STREAM);
+    if (returnRawInputStream == null) {
+      return defaultWhenUndefined;
+    } else {
+      return Boolean.FALSE.equals(returnRawInputStream);
+    }
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
index a51369e1..1402a4b5 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
@@ -752,6 +752,12 @@ public class HttpStorageRpc implements StorageRpc {
               .setIfGenerationNotMatch(Option.IF_GENERATION_NOT_MATCH.getLong(options))
               .setUserProject(Option.USER_PROJECT.getString(options));
       setEncryptionHeaders(getRequest.getRequestHeaders(), ENCRYPTION_KEY_PREFIX, options);
+      Boolean shouldReturnRawInputStream = Option.RETURN_RAW_INPUT_STREAM.getBoolean(options);
+      if (shouldReturnRawInputStream != null) {
+        getRequest.setReturnRawInputStream(shouldReturnRawInputStream);
+      } else {
+        getRequest.setReturnRawInputStream(false);
+      }
       ByteArrayOutputStream out = new ByteArrayOutputStream();
       getRequest.executeMedia().download(out);
       return out.toByteArray();


Test patch:
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 2e8efa05..e5ef2d47 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
@@ -245,10 +245,10 @@ public class ITGzipReadableByteChannelTest {
     }
 
     @Test
-    public void storage_readAllBytes_defaultCompressed() {
+    public void storage_readAllBytes_defaultUncompressed() {
       Storage s = storageFixture.getInstance();
       byte[] actual = s.readAllBytes(BlobId.of("buck", "obj-compressed"));
-      assertThat(actual).isEqualTo(dataCompressed);
+      assertThat(actual).isEqualTo(dataUncompressed);
     }
 
     @Test
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/UtilsTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/UtilsTest.java
new file mode 100644
index 00000000..12a091bd
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/UtilsTest.java
@@ -0,0 +1,67 @@
+/*
+ * Copyright 2023 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.cloud.storage.TestUtils.assertAll;
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.storage.UnifiedOpts.Opt;
+import com.google.cloud.storage.UnifiedOpts.Opts;
+import org.junit.Test;
+
+public final class UtilsTest {
+  private static final Opts<Opt> autoGzipDecompress_undefined = Opts.empty();
+  private static final Opts<Opt> autoGzipDecompress_no =
+      Opts.from(UnifiedOpts.returnRawInputStream(true));
+  private static final Opts<Opt> autoGzipDecompress_yes =
+      Opts.from(UnifiedOpts.returnRawInputStream(false));
+
+  @Test
+  public void isAutoGzipDecompression() throws Exception {
+    assertAll(
+        () ->
+            assertThat(
+                    Utils.isAutoGzipDecompression(
+                        autoGzipDecompress_undefined, /*defaultWhenUndefined=*/ false))
+                .isFalse(),
+        () ->
+            assertThat(
+                    Utils.isAutoGzipDecompression(
+                        autoGzipDecompress_undefined, /*defaultWhenUndefined=*/ true))
+                .isTrue(),
+        () ->
+            assertThat(
+                    Utils.isAutoGzipDecompression(
+                        autoGzipDecompress_no, /*defaultWhenUndefined=*/ false))
+                .isFalse(),
+        () ->
+            assertThat(
+                    Utils.isAutoGzipDecompression(
+                        autoGzipDecompress_no, /*defaultWhenUndefined=*/ true))
+                .isFalse(),
+        () ->
+            assertThat(
+                    Utils.isAutoGzipDecompression(
+                        autoGzipDecompress_yes, /*defaultWhenUndefined=*/ false))
+                .isTrue(),
+        () ->
+            assertThat(
+                    Utils.isAutoGzipDecompression(
+                        autoGzipDecompress_yes, /*defaultWhenUndefined=*/ true))
+                .isTrue());
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITAutomaticGzipDecompressionTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITAutomaticGzipDecompressionTest.java
new file mode 100644
index 00000000..912faa9d
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITAutomaticGzipDecompressionTest.java
@@ -0,0 +1,183 @@
+/*
+ * Copyright 2023 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.it;
+
+import static com.google.cloud.storage.TestUtils.xxd;
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.ReadChannel;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.TestUtils;
+import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.cloud.storage.it.runner.StorageITRunner;
+import com.google.cloud.storage.it.runner.annotations.Backend;
+import com.google.cloud.storage.it.runner.annotations.CrossRun;
+import com.google.cloud.storage.it.runner.annotations.Inject;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.common.io.ByteStreams;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.nio.channels.Channels;
+import java.nio.channels.WritableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@CrossRun(
+    backends = {Backend.PROD},
+    transports = {Transport.HTTP, Transport.GRPC})
+public final class ITAutomaticGzipDecompressionTest {
+
+  private static final byte[] helloWorldTextBytes = "hello world".getBytes();
+  private static final byte[] helloWorldGzipBytes = TestUtils.gzipBytes(helloWorldTextBytes);
+
+  @Inject public Storage storage;
+  @Inject public BucketInfo bucket;
+  @Inject public Generator generator;
+
+  private BlobInfo info;
+  private BlobId blobId;
+
+  @Before
+  public void setUp() throws Exception {
+    BlobInfo tmp =
+        BlobInfo.newBuilder(bucket, generator.randomObjectName())
+            // define an object with explicit content type and encoding.
+            // JSON and gRPC have differing default behavior returning these values if they are
+            // either undefined, or match HTTP defaults.
+            .setContentType("text/plain")
+            .setContentEncoding("gzip")
+            .build();
+
+    Blob gen1 = storage.create(tmp, helloWorldGzipBytes, BlobTargetOption.doesNotExist());
+    info = gen1.asBlobInfo();
+    blobId = info.getBlobId();
+  }
+
+  @Test
+  public void readAllBytes_default_uncompressed() {
+    byte[] bytes = storage.readAllBytes(blobId);
+    assertThat(xxd(bytes)).isEqualTo(xxd(helloWorldTextBytes));
+  }
+
+  @Test
+  public void readAllBytes_returnRawInputStream_yes() {
+    byte[] bytes = storage.readAllBytes(blobId, BlobSourceOption.shouldReturnRawInputStream(true));
+    assertThat(xxd(bytes)).isEqualTo(xxd(helloWorldGzipBytes));
+  }
+
+  @Test
+  public void readAllBytes_returnRawInputStream_no() {
+    byte[] bytes = storage.readAllBytes(blobId, BlobSourceOption.shouldReturnRawInputStream(false));
+    assertThat(xxd(bytes)).isEqualTo(xxd(helloWorldTextBytes));
+  }
+
+  @Test
+  public void reader_default_compressed() throws IOException {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    try (ReadChannel r = storage.reader(blobId)) {
+      WritableByteChannel w = Channels.newChannel(baos);
+      ByteStreams.copy(r, w);
+    }
+
+    assertThat(xxd(baos.toByteArray())).isEqualTo(xxd(helloWorldGzipBytes));
+  }
+
+  @Test
+  public void reader_returnRawInputStream_yes() throws IOException {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    try (ReadChannel r =
+        storage.reader(blobId, BlobSourceOption.shouldReturnRawInputStream(true))) {
+      WritableByteChannel w = Channels.newChannel(baos);
+      ByteStreams.copy(r, w);
+    }
+
+    assertThat(xxd(baos.toByteArray())).isEqualTo(xxd(helloWorldGzipBytes));
+  }
+
+  @Test
+  public void reader_returnRawInputStream_no() throws IOException {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    try (ReadChannel r =
+        storage.reader(blobId, BlobSourceOption.shouldReturnRawInputStream(false))) {
+      WritableByteChannel w = Channels.newChannel(baos);
+      ByteStreams.copy(r, w);
+    }
+
+    assertThat(xxd(baos.toByteArray())).isEqualTo(xxd(helloWorldTextBytes));
+  }
+
+  @Test
+  public void downloadTo_path_default_uncompressed() throws IOException {
+    Path helloWorldTxtGz = File.createTempFile(blobId.getName(), ".txt.gz").toPath();
+    storage.downloadTo(blobId, helloWorldTxtGz);
+
+    byte[] actualTxtGzBytes = Files.readAllBytes(helloWorldTxtGz);
+    assertThat(xxd(actualTxtGzBytes)).isEqualTo(xxd(helloWorldTextBytes));
+  }
+
+  @Test
+  public void downloadTo_path_returnRawInputStream_yes() throws IOException {
+    Path helloWorldTxtGz = File.createTempFile(blobId.getName(), ".txt.gz").toPath();
+    storage.downloadTo(blobId, helloWorldTxtGz, BlobSourceOption.shouldReturnRawInputStream(true));
+
+    byte[] actualTxtGzBytes = Files.readAllBytes(helloWorldTxtGz);
+    assertThat(xxd(actualTxtGzBytes)).isEqualTo(xxd(helloWorldGzipBytes));
+  }
+
+  @Test
+  public void downloadTo_path_returnRawInputStream_no() throws IOException {
+    Path helloWorldTxt = File.createTempFile(blobId.getName(), ".txt").toPath();
+    storage.downloadTo(blobId, helloWorldTxt, BlobSourceOption.shouldReturnRawInputStream(false));
+    byte[] actualTxtBytes = Files.readAllBytes(helloWorldTxt);
+    assertThat(xxd(actualTxtBytes)).isEqualTo(xxd(helloWorldTextBytes));
+  }
+
+  @Test
+  public void downloadTo_outputStream_default_uncompressed() {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    storage.downloadTo(blobId, baos);
+    byte[] actual = baos.toByteArray();
+    assertThat(xxd(actual)).isEqualTo(xxd(helloWorldTextBytes));
+  }
+
+  @Test
+  public void downloadTo_outputStream_returnRawInputStream_yes() {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    storage.downloadTo(blobId, baos, BlobSourceOption.shouldReturnRawInputStream(true));
+    byte[] actual = baos.toByteArray();
+    assertThat(xxd(actual)).isEqualTo(xxd(helloWorldGzipBytes));
+  }
+
+  @Test
+  public void downloadTo_outputStream_returnRawInputStream_no() {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    storage.downloadTo(blobId, baos, BlobSourceOption.shouldReturnRawInputStream(false));
+    byte[] actual = baos.toByteArray();
+    assertThat(xxd(actual)).isEqualTo(xxd(helloWorldTextBytes));
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITDownloadToTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITDownloadToTest.java
deleted file mode 100644
index 097609a5..00000000
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITDownloadToTest.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * 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.it;
-
-import static com.google.common.truth.Truth.assertThat;
-import static org.junit.Assert.fail;
-
-import com.google.cloud.storage.BlobId;
-import com.google.cloud.storage.BlobInfo;
-import com.google.cloud.storage.BucketInfo;
-import com.google.cloud.storage.Storage;
-import com.google.cloud.storage.TestUtils;
-import com.google.cloud.storage.TransportCompatibility.Transport;
-import com.google.cloud.storage.it.runner.StorageITRunner;
-import com.google.cloud.storage.it.runner.annotations.Backend;
-import com.google.cloud.storage.it.runner.annotations.CrossRun;
-import com.google.cloud.storage.it.runner.annotations.Inject;
-import com.google.cloud.storage.it.runner.registry.Generator;
-import java.io.File;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.Arrays;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-
-@RunWith(StorageITRunner.class)
-@CrossRun(
-    transports = {Transport.HTTP, Transport.GRPC},
-    backends = {Backend.PROD})
-public final class ITDownloadToTest {
-
-  private static final byte[] helloWorldTextBytes = "hello world".getBytes();
-  private static final byte[] helloWorldGzipBytes = TestUtils.gzipBytes(helloWorldTextBytes);
-
-  @Inject public Storage storage;
-  @Inject public BucketInfo bucket;
-  @Inject public Generator generator;
-
-  private BlobId blobId;
-
-  @Before
-  public void before() {
-    String objectString = generator.randomObjectName();
-    blobId = BlobId.of(bucket.getName(), objectString);
-    BlobInfo blobInfo =
-        BlobInfo.newBuilder(blobId).setContentEncoding("gzip").setContentType("text/plain").build();
-    storage.create(blobInfo, helloWorldGzipBytes);
-  }
-
-  @Test
-  public void downloadTo_returnRawInputStream_yes() throws IOException {
-    Path helloWorldTxtGz = File.createTempFile(blobId.getName(), ".txt.gz").toPath();
-    storage.downloadTo(
-        blobId, helloWorldTxtGz, Storage.BlobSourceOption.shouldReturnRawInputStream(true));
-
-    byte[] actualTxtGzBytes = Files.readAllBytes(helloWorldTxtGz);
-    if (Arrays.equals(actualTxtGzBytes, helloWorldTextBytes)) {
-      fail("expected gzipped bytes, but got un-gzipped bytes");
-    }
-    assertThat(actualTxtGzBytes).isEqualTo(helloWorldGzipBytes);
-  }
-
-  @Test
-  public void downloadTo_returnRawInputStream_no() throws IOException {
-    Path helloWorldTxt = File.createTempFile(blobId.getName(), ".txt").toPath();
-    storage.downloadTo(
-        blobId, helloWorldTxt, Storage.BlobSourceOption.shouldReturnRawInputStream(false));
-    byte[] actualTxtBytes = Files.readAllBytes(helloWorldTxt);
-    assertThat(actualTxtBytes).isEqualTo(helloWorldTextBytes);
-  }
-}