Repository: googleapis/java-storage
Pull request #1578 (2022-08-22 18:13:13)

chore: wire automatic gzip decompression into DownloadSessionBuilder and channels
Create new GzipReadableByteChannel which will be wired into the download session channel
if enabled.

By default, auto decompression IS NOT enabled, GrpcStorageImpl is responsible for translating
the meaning from `RETURN_RAW_INPUT` stream to `autoGzipDecompress`. Enabling means additional
startup effort and memory overhead due to loss of buffer control.

#### AutoClosableFixture
JUnit 4 Rule which can be used to bind some `AutoClosable` to a `@ClassRule` or `@Rule` lifecycle



Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBufferedReadableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBufferedReadableByteChannel.java
index 45359a22..4fae3718 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBufferedReadableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBufferedReadableByteChannel.java
@@ -133,7 +133,7 @@ final class DefaultBufferedReadableByteChannel implements BufferedReadableByteCh
 
   @Override
   public boolean isOpen() {
-    return channel.isOpen();
+    return !retEOF && channel.isOpen();
   }
 
   @Override
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 8177e4ea..e2a2dada 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
@@ -49,12 +49,14 @@ final class GapicDownloadSessionBuilder {
   public static final class ReadableByteChannelSessionBuilder {
 
     private final ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read;
+    private boolean autoGzipDecompression;
     private Hasher hasher;
 
     private ReadableByteChannelSessionBuilder(
         ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read) {
       this.read = read;
       this.hasher = Hasher.noop();
+      this.autoGzipDecompression = false;
     }
 
     public BufferedReadableByteChannelSessionBuilder buffered() {
@@ -66,25 +68,37 @@ final class GapicDownloadSessionBuilder {
       return this;
     }
 
+    public ReadableByteChannelSessionBuilder setAutoGzipDecompression(
+        boolean autoGzipDecompression) {
+      this.autoGzipDecompression = autoGzipDecompression;
+      return this;
+    }
+
     public BufferedReadableByteChannelSessionBuilder buffered(int capacity) {
-      return new BufferedReadableByteChannelSessionBuilder(
-          BufferHandle.allocate(capacity), getF(read, hasher));
+      return new BufferedReadableByteChannelSessionBuilder(BufferHandle.allocate(capacity), getF());
     }
 
     public BufferedReadableByteChannelSessionBuilder buffered(ByteBuffer buffer) {
-      return new BufferedReadableByteChannelSessionBuilder(
-          BufferHandle.handleOf(buffer), getF(read, hasher));
+      return new BufferedReadableByteChannelSessionBuilder(BufferHandle.handleOf(buffer), getF());
     }
 
     public UnbufferedReadableByteChannelSessionBuilder unbuffered() {
-      return new UnbufferedReadableByteChannelSessionBuilder(getF(read, hasher));
+      return new UnbufferedReadableByteChannelSessionBuilder(getF());
     }
 
-    private static BiFunction<
-            ReadObjectRequest, SettableApiFuture<Object>, UnbufferedReadableByteChannel>
-        getF(ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read, Hasher hasher) {
-      return (object, resultFuture) ->
-          new GapicUnbufferedReadableByteChannel(resultFuture, read, object, hasher);
+    private BiFunction<ReadObjectRequest, SettableApiFuture<Object>, UnbufferedReadableByteChannel>
+        getF() {
+      // for any non-final value, create a reference to the value at this point in time
+      Hasher hasher = this.hasher;
+      boolean autoGzipDecompression = this.autoGzipDecompression;
+      return (object, resultFuture) -> {
+        if (autoGzipDecompression) {
+          return new GzipReadableByteChannel(
+              new GapicUnbufferedReadableByteChannel(resultFuture, read, object, hasher));
+        } else {
+          return new GapicUnbufferedReadableByteChannel(resultFuture, read, object, hasher);
+        }
+      };
     }
 
     public static final class BufferedReadableByteChannelSessionBuilder {
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 c1079166..b1f2150a 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,6 +19,7 @@ package com.google.cloud.storage;
 import static com.google.common.base.Preconditions.checkArgument;
 
 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.rpc.ServerStreamingCallable;
@@ -182,6 +183,10 @@ final class GapicUnbufferedReadableByteChannel
     iter.close();
   }
 
+  ApiFuture<Object> getResult() {
+    return result;
+  }
+
   private void copy(ReadCursor c, ByteBuffer content, ByteBuffer[] dsts, int offset, int length) {
     long copiedBytes = Buffers.copy(content, dsts, offset, length);
     c.advance(copiedBytes);
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 83910610..411b5a7d 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
@@ -44,7 +44,8 @@ final class GrpcBlobReadChannel implements ReadChannel {
 
   GrpcBlobReadChannel(
       ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read,
-      ReadObjectRequest request) {
+      ReadObjectRequest request,
+      boolean autoGzipDecompression) {
     this.lazyReadChannel =
         new LazyReadChannel(
             Suppliers.memoize(
@@ -54,6 +55,7 @@ final class GrpcBlobReadChannel implements ReadChannel {
                       .read()
                       .byteChannel(read)
                       .setHasher(Hasher.noop())
+                      .setAutoGzipDecompression(autoGzipDecompression)
                       .buffered(Buffers.allocate(chunkSize))
                       .setReadObjectRequest(req)
                       .build();
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 53a28e17..d6e77ea9 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
@@ -565,19 +565,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) {
-    Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob);
-    ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts);
-    Set<StatusCode.Code> codes =
-        GrpcStorageImpl.resultRetryAlgorithmToCodes(
-            retryAlgorithmManager.getFor(readObjectRequest));
-    GrpcCallContext grpcCallContext = GrpcCallContext.createDefault().withRetryableCodes(codes);
-    UnbufferedReadableByteChannelSession<Object> session =
-        ResumableMedia.gapic()
-            .read()
-            .byteChannel(storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext))
-            .unbuffered()
-            .setReadObjectRequest(readObjectRequest)
-            .build();
+    UnbufferedReadableByteChannelSession<Object> session = unbufferedReadSession(blob, options);
 
     ByteArrayOutputStream baos = new ByteArrayOutputStream();
     try (UnbufferedReadableByteChannel r = session.open();
@@ -607,28 +595,15 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         GrpcStorageImpl.resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(request));
     GrpcCallContext grpcCallContext = GrpcCallContext.createDefault().withRetryableCodes(codes);
     return new GrpcBlobReadChannel(
-        storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext), request);
+        storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext),
+        request,
+        !opts.autoGzipDecompression());
   }
 
   @Override
   public void downloadTo(BlobId blob, Path path, BlobSourceOption... options) {
 
-    Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob);
-
-    // TODO: handle StorageRpc.Option.RETURN_RAW_INPUT_STREAM impacts
-
-    ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts);
-    Set<StatusCode.Code> codes =
-        GrpcStorageImpl.resultRetryAlgorithmToCodes(
-            retryAlgorithmManager.getFor(readObjectRequest));
-    GrpcCallContext grpcCallContext = GrpcCallContext.createDefault().withRetryableCodes(codes);
-    UnbufferedReadableByteChannelSession<Object> session =
-        ResumableMedia.gapic()
-            .read()
-            .byteChannel(storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext))
-            .unbuffered()
-            .setReadObjectRequest(readObjectRequest)
-            .build();
+    UnbufferedReadableByteChannelSession<Object> session = unbufferedReadSession(blob, options);
 
     try (UnbufferedReadableByteChannel r = session.open();
         WritableByteChannel w = Files.newByteChannel(path, WRITE_OPS)) {
@@ -641,22 +616,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   @Override
   public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption... options) {
 
-    Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob);
-
-    // TODO: handle StorageRpc.Option.RETURN_RAW_INPUT_STREAM impacts
-
-    ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts);
-    Set<StatusCode.Code> codes =
-        GrpcStorageImpl.resultRetryAlgorithmToCodes(
-            retryAlgorithmManager.getFor(readObjectRequest));
-    GrpcCallContext grpcCallContext = GrpcCallContext.createDefault().withRetryableCodes(codes);
-    UnbufferedReadableByteChannelSession<Object> session =
-        ResumableMedia.gapic()
-            .read()
-            .byteChannel(storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext))
-            .unbuffered()
-            .setReadObjectRequest(readObjectRequest)
-            .build();
+    UnbufferedReadableByteChannelSession<Object> session = unbufferedReadSession(blob, options);
 
     try (UnbufferedReadableByteChannel r = session.open();
         WritableByteChannel w = Channels.newChannel(outputStream)) {
@@ -1227,6 +1187,23 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     return opts.writeObjectRequest().apply(requestBuilder).build();
   }
 
+  private UnbufferedReadableByteChannelSession<Object> unbufferedReadSession(
+      BlobId blob, BlobSourceOption[] options) {
+    Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob);
+    ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts);
+    Set<StatusCode.Code> codes =
+        GrpcStorageImpl.resultRetryAlgorithmToCodes(
+            retryAlgorithmManager.getFor(readObjectRequest));
+    GrpcCallContext grpcCallContext = GrpcCallContext.createDefault().withRetryableCodes(codes);
+    return ResumableMedia.gapic()
+        .read()
+        .byteChannel(storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext))
+        .setAutoGzipDecompression(!opts.autoGzipDecompression())
+        .unbuffered()
+        .setReadObjectRequest(readObjectRequest)
+        .build();
+  }
+
   private FieldMask fieldMaskGenerator(Message message) {
     return FieldMask.newBuilder()
         .addAllPaths(
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
new file mode 100644
index 00000000..6ef048b3
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GzipReadableByteChannel.java
@@ -0,0 +1,171 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
+import com.google.storage.v2.Object;
+import java.io.*;
+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 {
+  private final GapicUnbufferedReadableByteChannel source;
+
+  private boolean retEOF = false;
+  private ScatteringByteChannel delegate;
+
+  GzipReadableByteChannel(GapicUnbufferedReadableByteChannel source) {
+    this.source = source;
+  }
+
+  @Override
+  public int read(ByteBuffer dst) throws IOException {
+    return Math.toIntExact(read(new ByteBuffer[] {dst}));
+  }
+
+  @Override
+  public long read(ByteBuffer[] dsts) throws IOException {
+    return read(dsts, 0, dsts.length);
+  }
+
+  @Override
+  public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
+    if (retEOF) {
+      return -1;
+    }
+    long bytesRead = 0;
+    // if our delegate is null, that means this is the first read attempt
+    if (delegate == null) {
+      // try to determine if the underlying data coming out of `source` is gzip
+      byte[] first4 = new byte[4]; // 4 bytes = 32-bits
+      final ByteBuffer wrap = ByteBuffer.wrap(first4);
+      // Step 1: initiate a read of the first 4 bytes of the object
+      //   this will have minimal overhead as the messages coming from gcs are inherently windowed
+      //   if the object size is between 5 and 2MiB the remaining bytes will be held in the channel
+      //   for later read.
+      source.read(wrap);
+      try {
+        // Step 2: wait for the object metadata, this is populated in the first message from GCS
+        Object object = source.getResult().get();
+        // 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(object.getContentEncoding())) {
+          // to wire gzip decompression into the byte path:
+          // Create an input stream of the first4 bytes we already read
+          ByteArrayInputStream first4again = new ByteArrayInputStream(first4);
+          // Create an InputStream facade of source
+          InputStream sourceInputStream = Channels.newInputStream(source);
+          // create a new InputStream with the first4 bytes prepended to source
+          SequenceInputStream first4AndSource =
+              new SequenceInputStream(first4again, sourceInputStream);
+          // add gzip decompression
+          GZIPInputStream decompress =
+              new GZIPInputStream(new OptimisticAvailabilityInputStream(first4AndSource));
+          // create a channel from our GZIPInputStream
+          ReadableByteChannel decompressedChannel = Channels.newChannel(decompress);
+          // turn our ReadableByteChannel into a ScatteringByteChannel
+          delegate = StorageByteChannels.readable().asScatteringByteChannel(decompressedChannel);
+        } else {
+          // if content encoding isn't gzip, copy the bytes we read into the dsts and set delegate
+          // to source
+          wrap.flip();
+          bytesRead += Buffers.copy(wrap, dsts, offset, length);
+          delegate = source;
+        }
+      } catch (InterruptedException | ExecutionException e) {
+        throw new IOException(e);
+      }
+    }
+
+    // Because we're pre-reading a few bytes of the object in order to determine if we need to
+    // plumb in gzip decompress, there is the possibility we will reach EOF while probing.
+    // In order to maintain correctness of EOF propagation, determine if we will need to signal EOF
+    // upon the next read.
+    long read = delegate.read(dsts, offset, length);
+    if (read == -1 && bytesRead == 0) {
+      return -1;
+    } else if (read == -1) {
+      retEOF = true;
+    } else {
+      bytesRead += read;
+    }
+
+    return bytesRead;
+  }
+
+  @Override
+  public boolean isOpen() {
+    return !retEOF && source.isOpen();
+  }
+
+  @Override
+  public void close() throws IOException {
+    // leverage try-with-resource to handle the dance closing these two resources
+    try (AutoCloseable ignored1 = source;
+        AutoCloseable ignored2 = delegate) {
+      delegate = null;
+    } catch (Exception e) {
+      throw new IOException("Error while attempting to close channel.", e);
+    }
+  }
+
+  /**
+   * There is an edge-case in the JDK's {@link GZIPInputStream} where it will prematurely terminate
+   * reading from the underlying InputStream.
+   *
+   * <p>This class decorates an InputStream to be optimistic about the number of available bytes
+   * when reading data to encourage GzipInputStream to consume the entire stream.
+   *
+   * <p>For a more in-depth write up see <a target="_blank" rel="noopener noreferrer"
+   * href="https://github.com/googleapis/google-http-java-client/pull/1608">google-http-java-client/pull/1608</a>
+   *
+   * <p><i>NOTE</i> This class is a copy of the private class from {@code
+   * com.google.api.client.http.GzipSupport}. This class should not be made public, as it is not
+   * general purpose and so is reproduced here.
+   */
+  private static final class OptimisticAvailabilityInputStream extends FilterInputStream {
+    private int lastRead = 0;
+
+    OptimisticAvailabilityInputStream(InputStream delegate) {
+      super(delegate);
+    }
+
+    @Override
+    public int available() throws IOException {
+      return lastRead > -1 ? Integer.MAX_VALUE : 0;
+    }
+
+    @Override
+    public int read() throws IOException {
+      return lastRead = super.read();
+    }
+
+    @Override
+    public int read(byte[] b) throws IOException {
+      return lastRead = super.read(b);
+    }
+
+    @Override
+    public int read(byte[] b, int off, int len) throws IOException {
+      return lastRead = super.read(b, off, len);
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageByteChannels.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageByteChannels.java
index cd4410db..48857cf2 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageByteChannels.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageByteChannels.java
@@ -22,6 +22,9 @@ import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedR
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.ScatteringByteChannel;
 
 final class StorageByteChannels {
 
@@ -46,6 +49,10 @@ final class StorageByteChannels {
         UnbufferedReadableByteChannel delegate) {
       return new SynchronizedUnbufferedReadableByteChannel(delegate);
     }
+
+    public ScatteringByteChannel asScatteringByteChannel(ReadableByteChannel c) {
+      return new ScatteringByteChannelFacade(c);
+    }
   }
 
   static final class Writable {
@@ -187,4 +194,57 @@ final class StorageByteChannels {
       delegate.close();
     }
   }
+
+  private static final class ScatteringByteChannelFacade implements ScatteringByteChannel {
+    private final ReadableByteChannel c;
+
+    private ScatteringByteChannelFacade(final ReadableByteChannel c) {
+      this.c = c;
+    }
+
+    @Override
+    public int read(ByteBuffer dst) throws IOException {
+      return Math.toIntExact(read(new ByteBuffer[] {dst}, 0, 1));
+    }
+
+    @Override
+    public long read(ByteBuffer[] dsts) throws IOException {
+      return read(dsts, 0, dsts.length);
+    }
+
+    @Override
+    public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
+      if (!c.isOpen()) {
+        throw new ClosedChannelException();
+      }
+
+      long totalBytesRead = 0;
+      for (int i = offset; i < length; i++) {
+        ByteBuffer dst = dsts[i];
+        if (dst.hasRemaining()) {
+          int read = c.read(dst);
+          if (read == -1) {
+            if (totalBytesRead == 0) {
+              c.close();
+              return -1;
+            } else {
+              break;
+            }
+          }
+          totalBytesRead += read;
+        }
+      }
+      return totalBytesRead;
+    }
+
+    @Override
+    public boolean isOpen() {
+      return c.isOpen();
+    }
+
+    @Override
+    public void close() throws IOException {
+      c.close();
+    }
+  }
 }
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 57fc397c..ddfae852 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
@@ -1228,6 +1228,7 @@ final class UnifiedOpts {
     }
   }
 
+  @Deprecated
   static final class ReturnRawInputStream extends RpcOptVal<@NonNull Boolean>
       implements ObjectSourceOpt {
     private static final long serialVersionUID = 505293506385742781L;
@@ -1962,6 +1963,18 @@ 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(false);
+    }
+
     private Mapper<ImmutableMap.Builder<StorageRpc.Option, Object>> rpcOptionMapper() {
       return fuseMappers(RpcOptVal.class, RpcOptVal::mapper);
     }


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/AutoClosableFixture.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/AutoClosableFixture.java
new file mode 100644
index 00000000..01f0d425
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/AutoClosableFixture.java
@@ -0,0 +1,65 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import com.google.common.base.Preconditions;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.junit.rules.TestRule;
+import org.junit.runner.Description;
+import org.junit.runners.model.Statement;
+
+final class AutoClosableFixture<@NonNull T extends AutoCloseable> implements TestRule {
+  private final ThrowingSupplier<T> supplier;
+
+  @Nullable private T instance;
+
+  private AutoClosableFixture(ThrowingSupplier<T> supplier) {
+    this.supplier = supplier;
+  }
+
+  @NonNull
+  public T getInstance() {
+    Preconditions.checkState(instance != null, "getInstance() called outside active lifecycle.");
+    return instance;
+  }
+
+  @Override
+  public Statement apply(Statement base, Description description) {
+    return new Statement() {
+      @Override
+      public void evaluate() throws Throwable {
+        try (T t = supplier.get()) {
+          instance = t;
+          base.evaluate();
+        } finally {
+          instance = null;
+        }
+      }
+    };
+  }
+
+  static <@NonNull T extends AutoCloseable> AutoClosableFixture<T> of(
+      ThrowingSupplier<T> supplier) {
+    return new AutoClosableFixture<>(supplier);
+  }
+
+  @FunctionalInterface
+  interface ThrowingSupplier<@NonNull T> {
+    T get() throws Exception;
+  }
+}
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 506a013d..d2778342 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
@@ -35,6 +35,10 @@ final class FakeServer implements AutoCloseable {
     this.grpcStorageOptions = grpcStorageOptions;
   }
 
+  GrpcStorageOptions getGrpcStorageOptions() {
+    return grpcStorageOptions;
+  }
+
   StorageSettings storageSettings() throws IOException {
     return grpcStorageOptions.getStorageSettings();
   }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannelTest.java
index 5907edf0..092f5405 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannelTest.java
@@ -16,25 +16,18 @@
 
 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.grpc.GrpcCallContext;
-import com.google.api.gax.grpc.GrpcStatusCode;
-import com.google.api.gax.rpc.ApiException;
-import com.google.api.gax.rpc.ApiExceptionFactory;
-import com.google.api.gax.rpc.ErrorDetails;
 import com.google.api.gax.rpc.StatusCode;
 import com.google.cloud.storage.ChannelSession.UnbufferedReadSession;
 import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
 import com.google.common.hash.Hashing;
-import com.google.protobuf.Any;
 import com.google.protobuf.ByteString;
-import com.google.rpc.DebugInfo;
-import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.ContentRange;
 import com.google.storage.v2.Object;
 import com.google.storage.v2.ObjectChecksums;
@@ -43,7 +36,6 @@ import com.google.storage.v2.ReadObjectResponse;
 import com.google.storage.v2.StorageClient;
 import com.google.storage.v2.StorageGrpc;
 import io.grpc.Status.Code;
-import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -361,23 +353,4 @@ public final class GapicUnbufferedReadableByteChannelTest {
       assertThrows(ClosedChannelException.class, () -> c.read(buf));
     }
   }
-
-  private static ChecksummedData getChecksummedData(ByteString content) {
-    int crc32c = Hashing.crc32c().hashBytes(content.asReadOnlyByteBuffer()).asInt();
-    return ChecksummedData.newBuilder().setContent(content).setCrc32C(crc32c).build();
-  }
-
-  private static ApiException apiException(Code code) {
-    StatusRuntimeException statusRuntimeException = code.toStatus().asRuntimeException();
-    DebugInfo debugInfo =
-        DebugInfo.newBuilder().setDetail("forced failure |~| " + code.name()).build();
-    ErrorDetails errorDetails =
-        ErrorDetails.builder().setRawErrorMessages(ImmutableList.of(Any.pack(debugInfo))).build();
-    return ApiExceptionFactory.createException(
-        statusRuntimeException, GrpcStatusCode.of(code), true, errorDetails);
-  }
-
-  private static GrpcCallContext contextWithRetryForCodes(StatusCode.Code... code) {
-    return GrpcCallContext.createDefault().withRetryableCodes(ImmutableSet.copyOf(code));
-  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/GzipReadableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/GzipReadableByteChannelTest.java
new file mode 100644
index 00000000..27809c7b
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GzipReadableByteChannelTest.java
@@ -0,0 +1,317 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static com.google.cloud.storage.TestUtils.getChecksummedData;
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.ReadChannel;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
+import com.google.common.io.ByteStreams;
+import com.google.protobuf.ByteString;
+import com.google.storage.v2.Object;
+import com.google.storage.v2.ReadObjectRequest;
+import com.google.storage.v2.ReadObjectResponse;
+import com.google.storage.v2.StorageClient;
+import com.google.storage.v2.StorageGrpc;
+import io.grpc.Status;
+import io.grpc.stub.StreamObserver;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.security.SecureRandom;
+import java.util.concurrent.ExecutionException;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.runners.Enclosed;
+import org.junit.runner.RunWith;
+
+@RunWith(Enclosed.class)
+public class GzipReadableByteChannelTest {
+
+  @SuppressWarnings("PointlessArithmeticExpression")
+  private static final int _1KiB = 1 * 1024;
+
+  private static final int _2KiB = 2 * 1024;
+  private static final int _3KiB = _1KiB + _2KiB;
+  private static final SecureRandom rand = new SecureRandom();
+
+  private static final byte[] dataUncompressed = DataGenerator.rand(rand).genBytes(_3KiB);
+  private static final byte[] dataCompressed = TestUtils.gzipBytes(dataUncompressed);
+  private static final ByteString contentUncompressed1 =
+      ByteString.copyFrom(dataUncompressed, 0, _2KiB);
+  private static final ByteString contentUncompressed2 =
+      ByteString.copyFrom(dataUncompressed, _2KiB, _1KiB);
+  private static final ByteString contentCompressed1 =
+      ByteString.copyFrom(dataCompressed, 0, _2KiB);
+  private static final ByteString contentCompressed2 =
+      ByteString.copyFrom(dataCompressed, _2KiB, dataCompressed.length - _2KiB);
+  private static final ReadObjectRequest reqUncompressed =
+      ReadObjectRequest.newBuilder()
+          .setBucket("projects/_/buckets/buck")
+          .setObject("obj-uncompressed")
+          .build();
+  private static final ReadObjectRequest reqCompressed =
+      ReadObjectRequest.newBuilder()
+          .setBucket("projects/_/buckets/buck")
+          .setObject("obj-compressed")
+          .build();
+
+  private static final ReadObjectResponse respUncompressed1 =
+      ReadObjectResponse.newBuilder()
+          .setMetadata(Object.newBuilder().setContentEncoding("identity").build())
+          .setChecksummedData(getChecksummedData(contentUncompressed1))
+          .build();
+  private static final ReadObjectResponse respUncompressed2 =
+      ReadObjectResponse.newBuilder()
+          .setChecksummedData(getChecksummedData(contentUncompressed2))
+          .build();
+
+  private static final ReadObjectResponse respCompressed1 =
+      ReadObjectResponse.newBuilder()
+          .setMetadata(Object.newBuilder().setContentEncoding("gzip").build())
+          .setChecksummedData(getChecksummedData(contentCompressed1))
+          .build();
+  private static final ReadObjectResponse respCompressed2 =
+      ReadObjectResponse.newBuilder()
+          .setChecksummedData(getChecksummedData(contentCompressed2))
+          .build();
+
+  public static final class Uncompressed {
+    private static final StorageGrpc.StorageImplBase fakeStorage =
+        new StorageGrpc.StorageImplBase() {
+          @Override
+          public void readObject(
+              ReadObjectRequest request, StreamObserver<ReadObjectResponse> responseObserver) {
+            if (request.equals(reqUncompressed)) {
+              responseObserver.onNext(respUncompressed1);
+              responseObserver.onNext(respUncompressed2);
+              responseObserver.onCompleted();
+            } else {
+              responseObserver.onError(TestUtils.apiException(Status.Code.UNIMPLEMENTED));
+            }
+          }
+        };
+
+    @ClassRule(order = 1)
+    public static final AutoClosableFixture<FakeServer> fakeServer =
+        AutoClosableFixture.of(() -> FakeServer.of(fakeStorage));
+
+    @ClassRule(order = 2)
+    public static final AutoClosableFixture<StorageClient> storageClient =
+        AutoClosableFixture.of(
+            () -> StorageClient.create(fakeServer.getInstance().storageSettings()));
+
+    @Test
+    public void autoGzipDecompress_true() throws IOException {
+      UnbufferedReadableByteChannelSession<Object> session =
+          ResumableMedia.gapic()
+              .read()
+              .byteChannel(storageClient.getInstance().readObjectCallable())
+              .setHasher(Hasher.noop())
+              .setAutoGzipDecompression(true)
+              .unbuffered()
+              .setReadObjectRequest(reqUncompressed)
+              .build();
+
+      byte[] actualBytes = new byte[dataUncompressed.length];
+      try (UnbufferedReadableByteChannel c = session.open()) {
+        c.read(ByteBuffer.wrap(actualBytes));
+      }
+      assertThat(actualBytes).isEqualTo(dataUncompressed);
+    }
+
+    @Test
+    public void autoGzipDecompress_false() throws IOException {
+      UnbufferedReadableByteChannelSession<Object> session =
+          ResumableMedia.gapic()
+              .read()
+              .byteChannel(storageClient.getInstance().readObjectCallable())
+              .setHasher(Hasher.noop())
+              .setAutoGzipDecompression(false)
+              .unbuffered()
+              .setReadObjectRequest(reqUncompressed)
+              .build();
+
+      byte[] actualBytes = new byte[dataUncompressed.length];
+      try (UnbufferedReadableByteChannel c = session.open()) {
+        c.read(ByteBuffer.wrap(actualBytes));
+      }
+      assertThat(actualBytes).isEqualTo(dataUncompressed);
+    }
+  }
+
+  public static final class Compressed {
+
+    private static final StorageGrpc.StorageImplBase fakeStorage =
+        new StorageGrpc.StorageImplBase() {
+          @Override
+          public void readObject(
+              ReadObjectRequest request, StreamObserver<ReadObjectResponse> responseObserver) {
+            if (request.equals(reqCompressed)) {
+              responseObserver.onNext(respCompressed1);
+              responseObserver.onNext(respCompressed2);
+              responseObserver.onCompleted();
+            } else {
+              responseObserver.onError(TestUtils.apiException(Status.Code.UNIMPLEMENTED));
+            }
+          }
+        };
+
+    @ClassRule(order = 1)
+    public static final AutoClosableFixture<FakeServer> fakeServer =
+        AutoClosableFixture.of(() -> FakeServer.of(fakeStorage));
+
+    @ClassRule(order = 2)
+    public static final AutoClosableFixture<StorageClient> storageClient =
+        AutoClosableFixture.of(
+            () -> StorageClient.create(fakeServer.getInstance().storageSettings()));
+
+    @ClassRule(order = 3)
+    public static final StorageFixture storageFixture =
+        StorageFixture.of(() -> fakeServer.getInstance().getGrpcStorageOptions().getService());
+
+    @Test
+    public void autoGzipDecompress_true() throws IOException {
+      UnbufferedReadableByteChannelSession<Object> session =
+          ResumableMedia.gapic()
+              .read()
+              .byteChannel(storageClient.getInstance().readObjectCallable())
+              .setHasher(Hasher.noop())
+              .setAutoGzipDecompression(true)
+              .unbuffered()
+              .setReadObjectRequest(reqCompressed)
+              .build();
+
+      byte[] actualBytes = new byte[dataUncompressed.length];
+      try (UnbufferedReadableByteChannel c = session.open()) {
+        c.read(ByteBuffer.wrap(actualBytes));
+      }
+      assertThat(actualBytes).isEqualTo(dataUncompressed);
+    }
+
+    @Test
+    public void autoGzipDecompress_false() throws IOException {
+      UnbufferedReadableByteChannelSession<Object> session =
+          ResumableMedia.gapic()
+              .read()
+              .byteChannel(storageClient.getInstance().readObjectCallable())
+              .setHasher(Hasher.noop())
+              .setAutoGzipDecompression(false)
+              .unbuffered()
+              .setReadObjectRequest(reqCompressed)
+              .build();
+
+      byte[] actualBytes = new byte[dataCompressed.length];
+      try (UnbufferedReadableByteChannel c = session.open()) {
+        c.read(ByteBuffer.wrap(actualBytes));
+      }
+      assertThat(actualBytes).isEqualTo(dataCompressed);
+    }
+
+    @Test
+    public void storage_readAllBytes() {
+      Storage s = storageFixture.getInstance();
+      byte[] actual = s.readAllBytes(BlobId.of("buck", "obj-compressed"));
+      assertThat(actual).isEqualTo(dataUncompressed);
+    }
+
+    @Test
+    public void storage_readAllBytes_returnRawInputStream() {
+      Storage s = storageFixture.getInstance();
+      byte[] actual =
+          s.readAllBytes(
+              BlobId.of("buck", "obj-compressed"),
+              BlobSourceOption.shouldReturnRawInputStream(true));
+      assertThat(actual).isEqualTo(dataCompressed);
+    }
+
+    @Test
+    public void storage_reader() throws Exception {
+      Storage s = storageFixture.getInstance();
+      byte[] actual = new byte[dataUncompressed.length];
+      try (ReadChannel c = s.reader(BlobId.of("buck", "obj-compressed"))) {
+        c.read(ByteBuffer.wrap(actual));
+      }
+      assertThat(actual).isEqualTo(dataUncompressed);
+    }
+
+    @Test
+    public void storage_reader_returnRawInputStream() throws Exception {
+      Storage s = storageFixture.getInstance();
+      byte[] actual = new byte[dataCompressed.length];
+      try (ReadChannel c =
+          s.reader(
+              BlobId.of("buck", "obj-compressed"),
+              BlobSourceOption.shouldReturnRawInputStream(true))) {
+        c.read(ByteBuffer.wrap(actual));
+      }
+      assertThat(actual).isEqualTo(dataCompressed);
+    }
+  }
+
+  public static final class Behavior {
+
+    @Test
+    public void properlyTracksEOF() throws IOException, InterruptedException, ExecutionException {
+      final StorageGrpc.StorageImplBase fakeStorage =
+          new StorageGrpc.StorageImplBase() {
+            int count = 0;
+
+            @Override
+            public void readObject(
+                ReadObjectRequest request, StreamObserver<ReadObjectResponse> responseObserver) {
+              if (count++ == 0) {
+                responseObserver.onNext(
+                    ReadObjectResponse.newBuilder()
+                        .setMetadata(Object.newBuilder().setSize(1).build())
+                        .setChecksummedData(getChecksummedData(ByteString.copyFromUtf8("a")))
+                        .build());
+                responseObserver.onCompleted();
+              } else {
+                responseObserver.onError(TestUtils.apiException(Status.Code.UNIMPLEMENTED));
+              }
+            }
+          };
+
+      try (FakeServer fakeServer = FakeServer.of(fakeStorage);
+          StorageClient sc = StorageClient.create(fakeServer.storageSettings())) {
+        ReadableByteChannelSession<?, Object> session =
+            ResumableMedia.gapic()
+                .read()
+                .byteChannel(sc.readObjectCallable())
+                .setHasher(Hasher.noop())
+                .setAutoGzipDecompression(true)
+                .unbuffered()
+                .setReadObjectRequest(reqUncompressed)
+                .build();
+
+        byte[] expected = new byte[] {(byte) 'a'};
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        try (ReadableByteChannel c = session.open()) {
+          ByteStreams.copy(c, Channels.newChannel(baos));
+        }
+        byte[] actual = baos.toByteArray();
+        assertThat(actual).isEqualTo(expected);
+        assertThat(session.getResult().get()).isNotNull();
+      }
+    }
+  }
+}
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
new file mode 100644
index 00000000..28f0ae7d
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.api.gax.rpc.ApiException;
+import com.google.api.gax.rpc.ApiExceptionFactory;
+import com.google.api.gax.rpc.ErrorDetails;
+import com.google.api.gax.rpc.StatusCode;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.hash.Hashing;
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.rpc.DebugInfo;
+import com.google.storage.v2.ChecksummedData;
+import io.grpc.Status.Code;
+import io.grpc.StatusRuntimeException;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.zip.GZIPOutputStream;
+
+public final class TestUtils {
+
+  private TestUtils() {}
+
+  public static byte[] gzipBytes(byte[] bytes) {
+    ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+    try (OutputStream out = new GZIPOutputStream(byteArrayOutputStream)) {
+      out.write(bytes);
+    } catch (IOException ignore) {
+      // GZIPOutputStream will only throw if the underlying stream throws.
+      // ByteArrayOutputStream does not throw on write
+    }
+
+    return byteArrayOutputStream.toByteArray();
+  }
+
+  public static ChecksummedData getChecksummedData(ByteString content) {
+    int crc32c = Hashing.crc32c().hashBytes(content.asReadOnlyByteBuffer()).asInt();
+    return ChecksummedData.newBuilder().setContent(content).setCrc32C(crc32c).build();
+  }
+
+  public static ApiException apiException(Code code) {
+    StatusRuntimeException statusRuntimeException = code.toStatus().asRuntimeException();
+    DebugInfo debugInfo =
+        DebugInfo.newBuilder().setDetail("forced failure |~| " + code.name()).build();
+    ErrorDetails errorDetails =
+        ErrorDetails.builder().setRawErrorMessages(ImmutableList.of(Any.pack(debugInfo))).build();
+    return ApiExceptionFactory.createException(
+        statusRuntimeException, GrpcStatusCode.of(code), true, errorDetails);
+  }
+
+  public static GrpcCallContext contextWithRetryForCodes(StatusCode.Code... code) {
+    return GrpcCallContext.createDefault().withRetryableCodes(ImmutableSet.copyOf(code));
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITDownloadToTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITDownloadToTest.java
index 0a1a37c7..0dd743f6 100644
--- 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
@@ -24,14 +24,12 @@ import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.BucketFixture;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.StorageFixture;
-import java.io.ByteArrayOutputStream;
+import com.google.cloud.storage.TestUtils;
 import java.io.File;
 import java.io.IOException;
-import java.io.OutputStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Arrays;
-import java.util.zip.GZIPOutputStream;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Test;
@@ -46,7 +44,7 @@ public final class ITDownloadToTest {
       BucketFixture.newBuilder().setHandle(storageFixture::getInstance).build();
 
   private static final byte[] helloWorldTextBytes = "hello world".getBytes();
-  private static final byte[] helloWorldGzipBytes = gzipBytes(helloWorldTextBytes);
+  private static final byte[] helloWorldGzipBytes = TestUtils.gzipBytes(helloWorldTextBytes);
 
   private static Storage storage;
   private static BlobId blobId;
@@ -82,14 +80,4 @@ public final class ITDownloadToTest {
     byte[] actualTxtBytes = Files.readAllBytes(helloWorldTxt);
     assertThat(actualTxtBytes).isEqualTo(helloWorldTextBytes);
   }
-
-  private static byte[] gzipBytes(byte[] bytes) {
-    ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
-    try (OutputStream out = new GZIPOutputStream(byteArrayOutputStream)) {
-      out.write(bytes);
-    } catch (IOException ignore) {
-    }
-
-    return byteArrayOutputStream.toByteArray();
-  }
 }