Repository: googleapis/java-storage
Pull request #2302 (2023-11-14 17:24:50)

deps: update actions/github-script action to v7
[![Mend Renovate](https://app.renovatebot.com/images/banner.svg)](https://renovatebot.com)

This PR contains the following updates:

| Package | Type | Update | Change |
|---|---|---|---|
| [actions/github-script](https://togithub.com/actions/github-script) | action | major | `v6` -> `v7` |

---

### Release Notes

<details>
<summary>actions/github-script (actions/github-script)</summary>

### [`v7`](https://togithub.com/actions/github-script/compare/v6...v7)

[Compare Source](https://togithub.com/actions/github-script/compare/v6...v7)

</details>

---

### Configuration

📅 **Schedule**: Branch creation - At any time (no schedule defined), Automerge - At any time (no schedule defined).

🚦 **Automerge**: Disabled by config. Please merge this manually once you are satisfied.

♻ **Rebasing**: Whenever PR becomes conflicted, or you tick the rebase/retry checkbox.

🔕 **Ignore**: Close this PR and you won't be reminded about this update again.

---

 - [ ] <!-- rebase-check -->If you want to rebase/retry this PR, check this box

---

This PR has been generated by [Mend Renovate](https://www.mend.io/free-developer-tools/renovate/). View repository job log [here](https://developer.mend.io/github/googleapis/java-storage).
<!--renovate-debug:eyJjcmVhdGVkSW5WZXIiOiIzNy40Ni4wIiwidXBkYXRlZEluVmVyIjoiMzcuNDYuMCIsInRhcmdldEJyYW5jaCI6Im1haW4ifQ==-->


forking-renovate[bot]: ### Edited/Blocked Notification

Renovate will not automatically rebase this PR, because it does not recognize the last commit author and assumes somebody else may have edited the PR.

You can manually request rebase by checking the rebase/retry box above.

 ⚠ **Warning**: custom changes will be lost.

Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Acl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Acl.java
index 3c879d36..e81bf789 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Acl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Acl.java
@@ -172,7 +172,7 @@ public final class Acl implements Serializable {
 
     @Override
     public String toString() {
-      return Conversions.apiary().entity().encode(this);
+      return Conversions.json().entity().encode(this);
     }
   }
 
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 d5f9cd35..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
@@ -47,6 +47,7 @@ import java.io.StringReader;
 import java.math.BigInteger;
 import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
+import java.nio.channels.ClosedChannelException;
 import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.ScatteringByteChannel;
 import java.util.List;
@@ -68,6 +69,7 @@ class ApiaryUnbufferedReadableByteChannel implements UnbufferedReadableByteChann
   private long position;
   private ScatteringByteChannel sbc;
   private boolean open;
+  private boolean returnEOF;
 
   // returned X-Goog-Generation header value
   private Long xGoogGeneration;
@@ -84,16 +86,30 @@ class ApiaryUnbufferedReadableByteChannel implements UnbufferedReadableByteChann
     this.options = options;
     this.resultRetryAlgorithm = resultRetryAlgorithm;
     this.open = true;
+    this.returnEOF = false;
     this.position = apiaryReadRequest.getByteRangeSpec().beginOffset();
   }
 
   @Override
   public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
+    if (returnEOF) {
+      close();
+      return -1;
+    } else if (!open) {
+      throw new ClosedChannelException();
+    }
+    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);
       try {
         // According to the contract of Retrying#run it's possible for sbc to be null even after
         // invocation. However, the function we provide is guaranteed to return non-null or throw
@@ -101,11 +117,11 @@ class ApiaryUnbufferedReadableByteChannel implements UnbufferedReadableByteChann
         //noinspection ConstantConditions
         long read = sbc.read(dsts, offset, length);
         if (read == -1) {
-          open = false;
+          returnEOF = true;
         } else {
-          position += read;
+          totalRead += read;
         }
-        return read;
+        return totalRead;
       } catch (Exception t) {
         if (resultRetryAlgorithm.shouldRetry(t, null)) {
           // if our retry algorithm COULD allow a retry, continue the loop and allow trying to
@@ -113,13 +129,24 @@ 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);
+        long delta = totalRemaining - totalRemainingAfter;
+        if (delta > 0) {
+          position += delta;
+          totalRead += delta;
         }
       }
     } while (true);
@@ -189,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/BackwardCompatibilityUtils.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BackwardCompatibilityUtils.java
index 33f31756..4b63b95a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BackwardCompatibilityUtils.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BackwardCompatibilityUtils.java
@@ -93,7 +93,7 @@ final class BackwardCompatibilityUtils {
       BucketInfo.log.log(record);
 
       LifecycleCondition condition =
-          Conversions.apiary().lifecycleCondition().decode(rule.getCondition());
+          Conversions.json().lifecycleCondition().decode(rule.getCondition());
       return new LifecycleRule(LifecycleAction.newDeleteAction(), condition);
     }
     LifecycleCondition.Builder condition = LifecycleCondition.newBuilder();
@@ -135,7 +135,7 @@ final class BackwardCompatibilityUtils {
         return new BucketInfo.IsLiveDeleteRule(isLive);
       }
     }
-    return new RawDeleteRule(Conversions.apiary().lifecycleRule().encode(from));
+    return new RawDeleteRule(Conversions.json().lifecycleRule().encode(from));
   }
 
   @SuppressWarnings("deprecation")
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java
index f5f07632..06d26a13 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java
@@ -101,12 +101,10 @@ abstract class BaseStorageWriteChannel<T> implements StorageWriteChannel {
       }
       int write = tmp.write(src);
       return write;
-    } catch (StorageException e) {
-      throw new IOException(e);
     } catch (IOException e) {
       throw e;
     } catch (Exception e) {
-      throw new IOException(StorageException.coalesce(e));
+      throw StorageException.coalesce(e);
     }
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobInfo.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobInfo.java
index d33f3140..288de2b3 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobInfo.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobInfo.java
@@ -83,7 +83,7 @@ public class BlobInfo implements Serializable {
    * The getter for this property never returns null, however null awareness is critical for
    * encoding
    *
-   * @see ApiaryConversions#blobInfo() encoder
+   * @see JsonConversions#blobInfo() encoder
    */
   final Map<String, String> metadata;
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannel.java
index 3efb97b2..d54c79ce 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannel.java
@@ -101,7 +101,7 @@ class BlobReadChannel implements ReadChannel {
 
     @Override
     public ReadChannel restore() {
-      StorageObject encode = Conversions.apiary().blobId().encode(blob);
+      StorageObject encode = Conversions.json().blobId().encode(blob);
       BlobReadChannelV2 channel =
           new BlobReadChannelV2(
               encode, requestOptions, BlobReadChannelContext.from(serviceOptions));
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 82aba8b1..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
@@ -39,15 +39,12 @@ final class BlobReadChannelV2 extends BaseStorageReadChannel<StorageObject> {
       StorageObject storageObject,
       Map<StorageRpc.Option, ?> opts,
       BlobReadChannelContext blobReadChannelContext) {
-    super(Conversions.apiary().blobInfo());
+    super(Conversions.json().blobInfo());
     this.storageObject = 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/BlobWriteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannel.java
index c74c94f4..6494fc66 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannel.java
@@ -78,8 +78,7 @@ class BlobWriteChannel extends BaseWriteChannel<StorageOptions, BlobInfo>
     @Override
     public WriteChannel restore() {
       try {
-        StorageObject encode =
-            entity != null ? Conversions.apiary().blobInfo().encode(entity) : null;
+        StorageObject encode = entity != null ? Conversions.json().blobInfo().encode(entity) : null;
         return new BlobWriteChannelV2.BlobWriteChannelV2State(
                 (HttpStorageOptions) serviceOptions,
                 JsonResumableWrite.of(encode, ImmutableMap.of(), uploadId, position),
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannelV2.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannelV2.java
index 48bad4de..8b9de3f6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannelV2.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannelV2.java
@@ -32,7 +32,7 @@ final class BlobWriteChannelV2 extends BaseStorageWriteChannel<StorageObject> {
   private final JsonResumableWrite start;
 
   BlobWriteChannelV2(BlobReadChannelContext blobChannelContext, JsonResumableWrite start) {
-    super(Conversions.apiary().blobInfo());
+    super(Conversions.json().blobInfo());
     this.start = start;
     this.blobChannelContext = blobChannelContext;
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfig.java
index a67d5a20..b52e9bd4 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfig.java
@@ -17,11 +17,9 @@
 package com.google.cloud.storage;
 
 import com.google.api.core.InternalApi;
-import com.google.cloud.storage.Conversions.Decoder;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
-import com.google.storage.v2.WriteObjectResponse;
 import java.io.IOException;
 import java.io.Serializable;
 import java.time.Clock;
@@ -52,9 +50,15 @@ public abstract class BlobWriteSessionConfig implements Serializable {
   interface WriterFactory {
     @InternalApi
     WritableByteChannelSession<?, BlobInfo> writeSession(
-        StorageInternal s,
-        BlobInfo info,
-        Opts<ObjectTargetOpt> opts,
-        Decoder<WriteObjectResponse, BlobInfo> d);
+        StorageInternal s, BlobInfo info, Opts<ObjectTargetOpt> opts);
   }
+
+  /**
+   * Internal marker interface to signify an implementation of {@link BlobWriteSessionConfig} is
+   * compatible with {@link com.google.cloud.storage.TransportCompatibility.Transport#GRPC}
+   *
+   * <p>We could evaluate the annotations, but the code for that is more complicated and probably
+   * not worth the effort.
+   */
+  interface GrpcCompatible {}
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
index e3ca3390..0f11cb19 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
@@ -21,6 +21,7 @@ import com.google.cloud.storage.GrpcStorageOptions.GrpcStorageDefaults;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartCleanupStrategy;
 import com.google.cloud.storage.Storage.BlobTargetOption;
 import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.common.collect.ImmutableList;
 import java.io.IOException;
 import java.nio.channels.WritableByteChannel;
@@ -240,6 +241,7 @@ public final class BlobWriteSessionConfigs {
    * @since 2.26.0 This new api is in preview and is subject to breaking changes.
    */
   @BetaApi
+  @TransportCompatibility({Transport.GRPC})
   public static DefaultBlobWriteSessionConfig getDefault() {
     return new DefaultBlobWriteSessionConfig(ByteSizeConstants._16MiB);
   }
@@ -255,6 +257,7 @@ public final class BlobWriteSessionConfigs {
    * @since 2.26.0 This new api is in preview and is subject to breaking changes.
    */
   @BetaApi
+  @TransportCompatibility({Transport.GRPC})
   public static BlobWriteSessionConfig bufferToTempDirThenUpload() throws IOException {
     return bufferToDiskThenUpload(
         Paths.get(System.getProperty("java.io.tmpdir"), "google-cloud-storage"));
@@ -271,6 +274,7 @@ public final class BlobWriteSessionConfigs {
    * @since 2.26.0 This new api is in preview and is subject to breaking changes.
    */
   @BetaApi
+  @TransportCompatibility({Transport.GRPC})
   public static BufferToDiskThenUpload bufferToDiskThenUpload(Path path) throws IOException {
     return bufferToDiskThenUpload(ImmutableList.of(path));
   }
@@ -289,6 +293,7 @@ public final class BlobWriteSessionConfigs {
    * @since 2.26.0 This new api is in preview and is subject to breaking changes.
    */
   @BetaApi
+  @TransportCompatibility({Transport.GRPC})
   public static BufferToDiskThenUpload bufferToDiskThenUpload(Collection<Path> paths)
       throws IOException {
     return new BufferToDiskThenUpload(ImmutableList.copyOf(paths), false);
@@ -306,6 +311,7 @@ public final class BlobWriteSessionConfigs {
    * @since 2.27.0 This new api is in preview and is subject to breaking changes.
    */
   @BetaApi
+  @TransportCompatibility(Transport.GRPC)
   public static JournalingBlobWriteSessionConfig journaling(Collection<Path> paths) {
     return new JournalingBlobWriteSessionConfig(ImmutableList.copyOf(paths), false);
   }
@@ -321,6 +327,7 @@ public final class BlobWriteSessionConfigs {
    * @since 2.28.0 This new api is in preview and is subject to breaking changes.
    */
   @BetaApi
+  @TransportCompatibility({Transport.GRPC})
   public static ParallelCompositeUploadBlobWriteSessionConfig parallelCompositeUpload() {
     return ParallelCompositeUploadBlobWriteSessionConfig.withDefaults();
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketInfo.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketInfo.java
index a4d52629..5750b555 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketInfo.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketInfo.java
@@ -89,7 +89,7 @@ public class BucketInfo implements Serializable {
    * The getter for this property never returns null, however null awareness is critical for
    * encoding to properly determine how to process rules conversion.
    *
-   * @see ApiaryConversions#bucketInfo() encoder
+   * @see JsonConversions#bucketInfo() encoder
    */
   final List<LifecycleRule> lifecycleRules;
 
@@ -119,7 +119,7 @@ public class BucketInfo implements Serializable {
   /**
    * non-private for backward compatibility on message class. log messages are now emitted from
    *
-   * @see ApiaryConversions#lifecycleRule()
+   * @see JsonConversions#lifecycleRule()
    */
   static final Logger log = Logger.getLogger(BucketInfo.class.getName());
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BufferToDiskThenUpload.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BufferToDiskThenUpload.java
index 430f30bc..8c9cb7c1 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BufferToDiskThenUpload.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BufferToDiskThenUpload.java
@@ -21,15 +21,14 @@ import com.google.api.core.ApiFutures;
 import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
 import com.google.api.core.SettableApiFuture;
-import com.google.cloud.storage.Conversions.Decoder;
 import com.google.cloud.storage.RecoveryFileManager.RecoveryVolumeSinkFactory;
 import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.MoreExecutors;
-import com.google.storage.v2.WriteObjectResponse;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
@@ -60,7 +59,9 @@ import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
  */
 @Immutable
 @BetaApi
-public final class BufferToDiskThenUpload extends BlobWriteSessionConfig {
+@TransportCompatibility({Transport.GRPC})
+public final class BufferToDiskThenUpload extends BlobWriteSessionConfig
+    implements BlobWriteSessionConfig.GrpcCompatible {
   private static final long serialVersionUID = 9059242302276891867L;
 
   /**
@@ -151,10 +152,7 @@ public final class BufferToDiskThenUpload extends BlobWriteSessionConfig {
     @InternalApi
     @Override
     public WritableByteChannelSession<?, BlobInfo> writeSession(
-        StorageInternal storage,
-        BlobInfo info,
-        Opts<ObjectTargetOpt> opts,
-        Decoder<WriteObjectResponse, BlobInfo> d) {
+        StorageInternal storage, BlobInfo info, Opts<ObjectTargetOpt> opts) {
       return new Factory.WriteToFileThenUpload(
           storage, info, opts, recoveryFileManager.newRecoveryFile(info));
     }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java
index e72613fe..1ea80287 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java
@@ -20,6 +20,7 @@ import java.io.IOException;
 import java.nio.Buffer;
 import java.nio.ByteBuffer;
 import java.nio.channels.ReadableByteChannel;
+import java.util.Arrays;
 import java.util.function.Consumer;
 
 /**
@@ -159,4 +160,9 @@ final class Buffers {
     }
     return total;
   }
+
+  static long totalRemaining(ByteBuffer[] buffers, int offset, int length) {
+    ByteBuffer[] sub = Arrays.copyOfRange(buffers, offset, length);
+    return Arrays.stream(sub).mapToLong(Buffer::remaining).sum();
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Conversions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Conversions.java
index 3b81639a..b8e9c571 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Conversions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Conversions.java
@@ -18,23 +18,55 @@ package com.google.cloud.storage;
 
 import org.checkerframework.checker.nullness.qual.Nullable;
 
+/**
+ * Abstraction utilities for converting between two different types.
+ *
+ * <p>Primarily targeted at encapsulating the logic for conversion from our model classes and the
+ * respective transport specific models.
+ */
 final class Conversions {
 
   private Conversions() {}
 
-  static ApiaryConversions apiary() {
-    return ApiaryConversions.INSTANCE;
+  /** Entry point to the registry of Codecs for conversions with the JSON Api model */
+  static JsonConversions json() {
+    return JsonConversions.INSTANCE;
   }
 
+  /** Entry point to the registry of Codecs for conversions with the gRPC Api model */
   static GrpcConversions grpc() {
     return GrpcConversions.INSTANCE;
   }
 
+  /**
+   * Abstraction representing a conversion to a different model type.
+   *
+   * <p>This class is the inverse of {@link Decoder}
+   *
+   * <p>A symmetric {@link Encoder} {@link Decoder} pair can make a {@link Codec}
+   *
+   * @param <From>
+   * @param <To>
+   * @see Decoder
+   * @see Codec
+   */
   @FunctionalInterface
   interface Encoder<From, To> {
     To encode(From f);
   }
 
+  /**
+   * Abstraction representing a conversion from a different model type.
+   *
+   * <p>This class is the inverse of {@link Encoder}
+   *
+   * <p>A symmetric {@link Encoder} {@link Decoder} pair can make a {@link Codec}
+   *
+   * @param <From>
+   * @param <To>
+   * @see Encoder
+   * @see Codec
+   */
   @FunctionalInterface
   interface Decoder<From, To> {
     To decode(From f);
@@ -43,13 +75,13 @@ final class Conversions {
       return f -> d.decode(this.decode(f));
     }
 
-    static <X> Decoder<X, X> identity() {
-      return (x) -> x;
-    }
-
     default <In> Decoder<In, To> compose(Decoder<In, From> before) {
       return in -> this.decode(before.decode(in));
     }
+
+    static <X> Decoder<X, X> identity() {
+      return (x) -> x;
+    }
   }
 
   interface Codec<A, B> extends Encoder<A, B>, Decoder<B, A> {
@@ -94,6 +126,10 @@ final class Conversions {
     }
   }
 
+  /**
+   * Internal implementation detail, not to be opened if the containing class and interfaces are
+   * ever opened up for access.
+   */
   private static final class SimpleCodec<A, B> implements Codec<A, B> {
     private final Encoder<A, B> e;
     private final Decoder<B, A> d;
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
index 79e3759e..090f929e 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
@@ -20,12 +20,15 @@ import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
 import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
+import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel;
 import com.google.cloud.storage.Conversions.Decoder;
+import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.MoreExecutors;
+import com.google.storage.v2.WriteObjectRequest;
 import com.google.storage.v2.WriteObjectResponse;
 import java.nio.channels.WritableByteChannel;
 import java.time.Clock;
@@ -52,7 +55,9 @@ import javax.annotation.concurrent.Immutable;
  */
 @Immutable
 @BetaApi
-public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig {
+@TransportCompatibility({Transport.GRPC})
+public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig
+    implements BlobWriteSessionConfig.GrpcCompatible {
   private static final long serialVersionUID = -6873740918589930633L;
 
   private final int chunkSize;
@@ -102,6 +107,9 @@ public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig
 
   @InternalApi
   private static final class Factory implements WriterFactory {
+    private static final Decoder<WriteObjectResponse, BlobInfo>
+        WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER =
+            Conversions.grpc().blobInfo().compose(WriteObjectResponse::getResource);
 
     private final int chunkSize;
 
@@ -112,21 +120,36 @@ public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig
     @InternalApi
     @Override
     public WritableByteChannelSession<?, BlobInfo> writeSession(
-        StorageInternal s,
-        BlobInfo info,
-        Opts<ObjectTargetOpt> opts,
-        Decoder<WriteObjectResponse, BlobInfo> d) {
-      // todo: invert this
-      //   make GrpcBlobWriteChannel use this factory to produce its WriteSession
+        StorageInternal s, BlobInfo info, Opts<ObjectTargetOpt> opts) {
       if (s instanceof GrpcStorageImpl) {
-        GrpcStorageImpl g = (GrpcStorageImpl) s;
-        GrpcBlobWriteChannel writer = g.internalWriter(info, opts);
-        writer.setChunkSize(chunkSize);
-        WritableByteChannelSession<BufferedWritableByteChannel, WriteObjectResponse> session =
-            writer.newLazyWriteChannel().getSession();
-        return new DecoratedWritableByteChannelSession<>(session, d);
+        return new DecoratedWritableByteChannelSession<>(
+            new LazySession<>(
+                new LazyWriteChannel<>(
+                    () -> {
+                      GrpcStorageImpl grpc = (GrpcStorageImpl) s;
+                      GrpcCallContext grpcCallContext =
+                          opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
+                      WriteObjectRequest req = grpc.getWriteObjectRequest(info, opts);
+
+                      ApiFuture<ResumableWrite> startResumableWrite =
+                          grpc.startResumableWrite(grpcCallContext, req);
+                      return ResumableMedia.gapic()
+                          .write()
+                          .byteChannel(grpc.storageClient.writeObjectCallable())
+                          .setHasher(Hasher.noop())
+                          .setByteStringStrategy(ByteStringStrategy.copy())
+                          .resumable()
+                          .withRetryConfig(
+                              grpc.getOptions(), grpc.retryAlgorithmManager.idempotent())
+                          .buffered(BufferHandle.allocate(chunkSize))
+                          .setStartAsync(startResumableWrite)
+                          .build();
+                    })),
+            WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER);
+      } else {
+        throw new IllegalStateException(
+            "Unknown Storage implementation: " + s.getClass().getName());
       }
-      return CrossTransportUtils.throwGrpcOnly(DefaultBlobWriteSessionConfig.class, "");
     }
   }
 
@@ -162,4 +185,23 @@ public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig
           delegate.getResult(), decoder::decode, MoreExecutors.directExecutor());
     }
   }
+
+  private static final class LazySession<R>
+      implements WritableByteChannelSession<BufferedWritableByteChannel, R> {
+    private final LazyWriteChannel<R> lazy;
+
+    private LazySession(LazyWriteChannel<R> lazy) {
+      this.lazy = lazy;
+    }
+
+    @Override
+    public ApiFuture<BufferedWritableByteChannel> openAsync() {
+      return lazy.getSession().openAsync();
+    }
+
+    @Override
+    public ApiFuture<R> getResult() {
+      return lazy.getSession().getResult();
+    }
+  }
 }
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 16963e4d..516748dc 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
@@ -176,7 +176,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
   final GrpcConversions codecs;
   final GrpcRetryAlgorithmManager retryAlgorithmManager;
   final SyntaxDecoders syntaxDecoders;
-  private final Decoder<WriteObjectResponse, BlobInfo> writeObjectResponseBlobInfoDecoder;
 
   // workaround for https://github.com/googleapis/java-storage/issues/1736
   private final Opts<UserProject> defaultOpts;
@@ -194,8 +193,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     this.codecs = Conversions.grpc();
     this.retryAlgorithmManager = options.getRetryAlgorithmManager();
     this.syntaxDecoders = new SyntaxDecoders();
-    this.writeObjectResponseBlobInfoDecoder =
-        codecs.blobInfo().compose(WriteObjectResponse::getResource);
     this.defaultProjectId = UnifiedOpts.projectId(options.getProjectId());
   }
 
@@ -253,11 +250,34 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
 
   @Override
   public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... options) {
-    try {
-      return createFrom(blobInfo, content, options);
-    } catch (IOException e) {
+    requireNonNull(blobInfo, "blobInfo must be non null");
+
+    Opts<ObjectTargetOpt> opts = Opts.unwrap(options).resolveFrom(blobInfo).prepend(defaultOpts);
+    GrpcCallContext grpcCallContext =
+        opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
+    WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
+
+    UnbufferedWritableByteChannelSession<WriteObjectResponse> session =
+        ResumableMedia.gapic()
+            .write()
+            .byteChannel(
+                storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
+            .setHasher(Hasher.enabled())
+            .setByteStringStrategy(ByteStringStrategy.noCopy())
+            .direct()
+            .unbuffered()
+            .setRequest(req)
+            .build();
+
+    // Specifically not in the try-with, so we don't close the provided stream
+    ReadableByteChannel src =
+        Channels.newChannel(firstNonNull(content, new ByteArrayInputStream(ZERO_BYTES)));
+    try (UnbufferedWritableByteChannel dst = session.open()) {
+      ByteStreams.copy(src, dst);
+    } catch (Exception e) {
       throw StorageException.coalesce(e);
     }
+    return getBlob(session.getResult());
   }
 
   @Override
@@ -312,7 +332,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
       }
       return codecs.blobInfo().decode(object).asBlob(this);
     } catch (InterruptedException | ExecutionException e) {
-      throw StorageException.coalesce(e);
+      throw StorageException.coalesce(e.getCause());
     }
   }
 
@@ -362,7 +382,14 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
   @Override
   public Bucket get(String bucket, BucketGetOption... options) {
     Opts<BucketSourceOpt> unwrap = Opts.unwrap(options);
-    return internalBucketGet(bucket, unwrap);
+    try {
+      return internalBucketGet(bucket, unwrap);
+    } catch (StorageException e) {
+      if (e.getCode() == 404) {
+        return null;
+      }
+      throw e;
+    }
   }
 
   @Override
@@ -656,7 +683,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();
@@ -684,16 +712,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)) {
@@ -706,7 +737,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)) {
@@ -719,14 +751,9 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
   @Override
   public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options) {
     Opts<ObjectTargetOpt> opts = Opts.unwrap(options).resolveFrom(blobInfo).prepend(defaultOpts);
-    return internalWriter(blobInfo, opts);
-  }
-
-  @Override
-  public GrpcBlobWriteChannel internalWriter(BlobInfo info, Opts<ObjectTargetOpt> opts) {
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
-    WriteObjectRequest req = getWriteObjectRequest(info, opts);
+    WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
     Hasher hasher = Hasher.noop();
     return new GrpcBlobWriteChannel(
         storageClient.writeObjectCallable(),
@@ -1534,7 +1561,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
   public BlobWriteSession blobWriteSession(BlobInfo info, BlobWriteOption... options) {
     Opts<ObjectTargetOpt> opts = Opts.unwrap(options).resolveFrom(info);
     WritableByteChannelSession<?, BlobInfo> writableByteChannelSession =
-        writerFactory.writeSession(this, info, opts, writeObjectResponseBlobInfoDecoder);
+        writerFactory.writeSession(this, info, opts);
     return BlobWriteSessions.of(writableByteChannelSession);
   }
 
@@ -1809,18 +1836,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/GrpcStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
index bee342f9..7f41bc96 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
@@ -16,6 +16,7 @@
 
 package com.google.cloud.storage;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static java.util.Objects.requireNonNull;
 
 import com.google.api.core.ApiClock;
@@ -533,6 +534,9 @@ public final class GrpcStorageOptions extends StorageOptions
     public GrpcStorageOptions.Builder setBlobWriteSessionConfig(
         @NonNull BlobWriteSessionConfig blobWriteSessionConfig) {
       requireNonNull(blobWriteSessionConfig, "blobWriteSessionConfig must be non null");
+      checkArgument(
+          blobWriteSessionConfig instanceof BlobWriteSessionConfig.GrpcCompatible,
+          "The provided instance of BlobWriteSessionConfig is not compatible with gRPC transport.");
       this.blobWriteSessionConfig = blobWriteSessionConfig;
       return this;
     }
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/HttpCopyWriter.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java
index b3f6db3d..7f4eacd9 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java
@@ -43,7 +43,7 @@ public class HttpCopyWriter extends CopyWriter {
     while (!isDone()) {
       copyChunk();
     }
-    BlobInfo info = Conversions.apiary().blobInfo().decode(rewriteResponse.result);
+    BlobInfo info = Conversions.json().blobInfo().decode(rewriteResponse.result);
     return info.asBlob(serviceOptions.getService());
   }
 
@@ -79,14 +79,14 @@ public class HttpCopyWriter extends CopyWriter {
   public RestorableState<CopyWriter> capture() {
     return StateImpl.newBuilder(
             serviceOptions,
-            Conversions.apiary().blobId().decode(rewriteResponse.rewriteRequest.source),
+            Conversions.json().blobId().decode(rewriteResponse.rewriteRequest.source),
             rewriteResponse.rewriteRequest.sourceOptions,
             rewriteResponse.rewriteRequest.overrideInfo,
-            Conversions.apiary().blobInfo().decode(rewriteResponse.rewriteRequest.target),
+            Conversions.json().blobInfo().decode(rewriteResponse.rewriteRequest.target),
             rewriteResponse.rewriteRequest.targetOptions)
         .setResult(
             rewriteResponse.result != null
-                ? Conversions.apiary().blobInfo().decode(rewriteResponse.result)
+                ? Conversions.json().blobInfo().decode(rewriteResponse.result)
                 : null)
         .setBlobSize(getBlobSize())
         .setIsDone(isDone())
@@ -207,16 +207,16 @@ public class HttpCopyWriter extends CopyWriter {
     public CopyWriter restore() {
       RewriteRequest rewriteRequest =
           new RewriteRequest(
-              Conversions.apiary().blobId().encode(source),
+              Conversions.json().blobId().encode(source),
               sourceOptions,
               overrideInfo,
-              Conversions.apiary().blobInfo().encode(target),
+              Conversions.json().blobInfo().encode(target),
               targetOptions,
               megabytesCopiedPerChunk);
       RewriteResponse rewriteResponse =
           new RewriteResponse(
               rewriteRequest,
-              result != null ? Conversions.apiary().blobInfo().encode(result) : null,
+              result != null ? Conversions.json().blobInfo().encode(result) : null,
               blobSize,
               isDone,
               rewriteToken,
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java
index c8cc0e3d..7978dd2b 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java
@@ -68,7 +68,8 @@ import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 @Immutable
 @BetaApi
 @TransportCompatibility(Transport.GRPC)
-public final class JournalingBlobWriteSessionConfig extends BlobWriteSessionConfig {
+public final class JournalingBlobWriteSessionConfig extends BlobWriteSessionConfig
+    implements BlobWriteSessionConfig.GrpcCompatible {
   private static final long serialVersionUID = 9059242302276891867L;
 
   /**
@@ -159,10 +160,7 @@ public final class JournalingBlobWriteSessionConfig extends BlobWriteSessionConf
     @InternalApi
     @Override
     public WritableByteChannelSession<?, BlobInfo> writeSession(
-        StorageInternal storage,
-        BlobInfo info,
-        Opts<ObjectTargetOpt> opts,
-        Decoder<WriteObjectResponse, BlobInfo> d) {
+        StorageInternal storage, BlobInfo info, Opts<ObjectTargetOpt> opts) {
       if (storage instanceof GrpcStorageImpl) {
         GrpcStorageImpl grpcStorage = (GrpcStorageImpl) storage;
         RecoveryFile recoveryFile = recoveryFileManager.newRecoveryFile(info);
@@ -189,7 +187,7 @@ public final class JournalingBlobWriteSessionConfig extends BlobWriteSessionConf
                 .setStartAsync(start)
                 .build();
 
-        return new JournalingUpload<>(session, start, d);
+        return new JournalingUpload<>(session, start);
       } else {
         return CrossTransportUtils.throwHttpJsonOnly(BlobWriteSessionConfigs.class, "journaling");
       }
@@ -204,11 +202,10 @@ public final class JournalingBlobWriteSessionConfig extends BlobWriteSessionConf
 
       public JournalingUpload(
           WritableByteChannelSession<WBC, WriteObjectResponse> session,
-          ApiFuture<WriteCtx<ResumableWrite>> start,
-          Decoder<WriteObjectResponse, BlobInfo> decoder) {
+          ApiFuture<WriteCtx<ResumableWrite>> start) {
         this.session = session;
         this.start = start;
-        this.decoder = decoder;
+        this.decoder = Conversions.grpc().blobInfo().compose(WriteObjectResponse::getResource);
       }
 
       @Override
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryConversions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonConversions.java
similarity index 99%
rename from google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryConversions.java
rename to google-cloud-storage/src/main/java/com/google/cloud/storage/JsonConversions.java
index 7226267e..8ba271c9 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryConversions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonConversions.java
@@ -83,8 +83,8 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 
 @InternalApi
-final class ApiaryConversions {
-  static final ApiaryConversions INSTANCE = new ApiaryConversions();
+final class JsonConversions {
+  static final JsonConversions INSTANCE = new JsonConversions();
   // gRPC has a Bucket.project property that apiary doesn't have yet.
   // when converting from gRPC to apiary or vice-versa we want to preserve this property. Until
   // such a time as the apiary model has a project field, we manually apply it with this name.
@@ -141,7 +141,7 @@ final class ApiaryConversions {
   private final Codec<com.google.cloud.Condition, com.google.api.services.storage.model.Expr>
       iamConditionCodec = Codec.of(this::conditionEncode, this::conditionDecode);
 
-  private ApiaryConversions() {}
+  private JsonConversions() {}
 
   Codec<Entity, String> entity() {
     return entityCodec;
@@ -433,7 +433,7 @@ final class ApiaryConversions {
         lift(Lifecycle::getRule).andThen(toListOf(lifecycleRule()::decode)),
         to::setLifecycleRules);
     ifNonNull(from.getDefaultEventBasedHold(), to::setDefaultEventBasedHold);
-    ifNonNull(from.getLabels(), ApiaryConversions::replaceDataNullValuesWithNull, to::setLabels);
+    ifNonNull(from.getLabels(), JsonConversions::replaceDataNullValuesWithNull, to::setLabels);
     ifNonNull(from.getBilling(), Billing::getRequesterPays, to::setRequesterPays);
     Encryption encryption = from.getEncryption();
     if (encryption != null
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadBlobWriteSessionConfig.java
index 7b9c78f5..d0240669 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadBlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadBlobWriteSessionConfig.java
@@ -26,9 +26,9 @@ import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
 import com.google.api.core.SettableApiFuture;
 import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel;
-import com.google.cloud.storage.Conversions.Decoder;
 import com.google.cloud.storage.MetadataField.PartRange;
 import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.common.annotations.VisibleForTesting;
@@ -36,7 +36,6 @@ import com.google.common.hash.HashCode;
 import com.google.common.hash.HashFunction;
 import com.google.common.hash.Hashing;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import com.google.storage.v2.WriteObjectResponse;
 import java.io.IOException;
 import java.io.ObjectOutputStream;
 import java.io.Serializable;
@@ -116,7 +115,9 @@ import org.checkerframework.checker.nullness.qual.NonNull;
  */
 @Immutable
 @BetaApi
-public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessionConfig {
+@TransportCompatibility({Transport.GRPC})
+public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessionConfig
+    implements BlobWriteSessionConfig.GrpcCompatible {
 
   private static final int MAX_PARTS_PER_COMPOSE = 32;
   private final int maxPartsPerCompose;
@@ -669,10 +670,7 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
 
     @Override
     public WritableByteChannelSession<?, BlobInfo> writeSession(
-        StorageInternal s,
-        BlobInfo info,
-        Opts<ObjectTargetOpt> opts,
-        Decoder<WriteObjectResponse, BlobInfo> d) {
+        StorageInternal s, BlobInfo info, Opts<ObjectTargetOpt> opts) {
       return new PCUSession(s, info, opts);
     }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableMedia.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableMedia.java
index 2d3fbf93..80448c3a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableMedia.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableMedia.java
@@ -37,7 +37,7 @@ final class ResumableMedia {
             () ->
                 storageOptions
                     .getStorageRpcV1()
-                    .open(Conversions.apiary().blobInfo().encode(blob), optionsMap),
+                    .open(Conversions.json().blobInfo().encode(blob), optionsMap),
             Function.identity());
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageBatch.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageBatch.java
index 2b89d442..9a1f125d 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageBatch.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageBatch.java
@@ -23,6 +23,7 @@ import com.google.api.services.storage.model.StorageObject;
 import com.google.cloud.storage.Storage.BlobGetOption;
 import com.google.cloud.storage.Storage.BlobSourceOption;
 import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.cloud.storage.spi.v1.RpcBatch;
 import com.google.cloud.storage.spi.v1.StorageRpc;
@@ -53,6 +54,7 @@ import java.util.Map;
  * Blob blob = result.get(); // returns get result or throws StorageException
  * }</pre>
  */
+@TransportCompatibility(Transport.HTTP)
 public class StorageBatch {
 
   private final RpcBatch batch;
@@ -86,6 +88,7 @@ public class StorageBatch {
    * {@code false} if the blob was not found, or throws a {@link StorageException} if the operation
    * failed.
    */
+  @TransportCompatibility(Transport.HTTP)
   public StorageBatchResult<Boolean> delete(
       String bucket, String blob, BlobSourceOption... options) {
     return delete(BlobId.of(bucket, blob), options);
@@ -97,11 +100,12 @@ public class StorageBatch {
    * {@code false} if the blob was not found, or throws a {@link StorageException} if the operation
    * failed.
    */
+  @TransportCompatibility(Transport.HTTP)
   public StorageBatchResult<Boolean> delete(BlobId blob, BlobSourceOption... options) {
     StorageBatchResult<Boolean> result = new StorageBatchResult<>();
     RpcBatch.Callback<Void> callback = createDeleteCallback(result);
     Map<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
-    batch.addDelete(Conversions.apiary().blobId().encode(blob), callback, optionsMap);
+    batch.addDelete(Conversions.json().blobId().encode(blob), callback, optionsMap);
     return result;
   }
 
@@ -111,12 +115,13 @@ public class StorageBatch {
    * {@link StorageBatchResult#get()} on the return value yields the updated {@link Blob} if
    * successful, or throws a {@link StorageException} if the operation failed.
    */
+  @TransportCompatibility(Transport.HTTP)
   public StorageBatchResult<Blob> update(BlobInfo blobInfo, BlobTargetOption... options) {
     StorageBatchResult<Blob> result = new StorageBatchResult<>();
     RpcBatch.Callback<StorageObject> callback = createUpdateCallback(this.options, result);
     Map<StorageRpc.Option, ?> optionMap =
         Opts.unwrap(options).resolveFrom(blobInfo).getRpcOptions();
-    batch.addPatch(Conversions.apiary().blobInfo().encode(blobInfo), callback, optionMap);
+    batch.addPatch(Conversions.json().blobInfo().encode(blobInfo), callback, optionMap);
     return result;
   }
 
@@ -127,6 +132,7 @@ public class StorageBatch {
    * {@code null} if no such blob exists, or throws a {@link StorageException} if the operation
    * failed.
    */
+  @TransportCompatibility(Transport.HTTP)
   public StorageBatchResult<Blob> get(String bucket, String blob, BlobGetOption... options) {
     return get(BlobId.of(bucket, blob), options);
   }
@@ -138,15 +144,17 @@ public class StorageBatch {
    * {@code null} if no such blob exists, or throws a {@link StorageException} if the operation
    * failed.
    */
+  @TransportCompatibility(Transport.HTTP)
   public StorageBatchResult<Blob> get(BlobId blob, BlobGetOption... options) {
     StorageBatchResult<Blob> result = new StorageBatchResult<>();
     RpcBatch.Callback<StorageObject> callback = createGetCallback(this.options, result);
     Map<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
-    batch.addGet(Conversions.apiary().blobId().encode(blob), callback, optionsMap);
+    batch.addGet(Conversions.json().blobId().encode(blob), callback, optionsMap);
     return result;
   }
 
   /** Submits this batch for processing using a single RPC request. */
+  @TransportCompatibility(Transport.HTTP)
   public void submit() {
     batch.submit();
   }
@@ -175,7 +183,7 @@ public class StorageBatch {
     return new RpcBatch.Callback<StorageObject>() {
       @Override
       public void onSuccess(StorageObject response) {
-        BlobInfo info = Conversions.apiary().blobInfo().decode(response);
+        BlobInfo info = Conversions.json().blobInfo().decode(response);
         result.success(response == null ? null : info.asBlob(serviceOptions.getService()));
       }
 
@@ -196,7 +204,7 @@ public class StorageBatch {
     return new RpcBatch.Callback<StorageObject>() {
       @Override
       public void onSuccess(StorageObject response) {
-        BlobInfo info = Conversions.apiary().blobInfo().decode(response);
+        BlobInfo info = Conversions.json().blobInfo().decode(response);
         result.success(response == null ? null : info.asBlob(serviceOptions.getService()));
       }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
index d4590c36..48167b5d 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
@@ -21,7 +21,6 @@ import static com.google.common.base.MoreObjects.firstNonNull;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 import static java.nio.charset.StandardCharsets.UTF_8;
-import static java.util.concurrent.Executors.callable;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.gax.paging.Page;
@@ -37,12 +36,14 @@ import com.google.cloud.PageImpl.NextPageFetcher;
 import com.google.cloud.Policy;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Acl.Entity;
+import com.google.cloud.storage.ApiaryUnbufferedReadableByteChannel.ApiaryReadRequest;
 import com.google.cloud.storage.BlobReadChannelV2.BlobReadChannelContext;
 import com.google.cloud.storage.HmacKey.HmacKeyMetadata;
 import com.google.cloud.storage.PostPolicyV4.ConditionV4Type;
 import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
 import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
 import com.google.cloud.storage.PostPolicyV4.PostPolicyV4Document;
+import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
 import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
@@ -59,9 +60,10 @@ import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.hash.Hashing;
 import com.google.common.io.BaseEncoding;
-import com.google.common.io.CountingOutputStream;
+import com.google.common.io.ByteStreams;
 import com.google.common.primitives.Ints;
 import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -73,6 +75,7 @@ import java.net.URLEncoder;
 import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.text.SimpleDateFormat;
@@ -106,7 +109,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   private static final int DEFAULT_BUFFER_SIZE = 15 * 1024 * 1024;
   private static final int MIN_BUFFER_SIZE = 256 * 1024;
 
-  private static final ApiaryConversions codecs = Conversions.apiary();
+  private static final JsonConversions codecs = Conversions.json();
 
   final HttpRetryAlgorithmManager retryAlgorithmManager;
   final StorageRpc storageRpc;
@@ -128,7 +131,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
     return run(
         algorithm,
         () -> storageRpc.create(bucketPb, optionsMap),
-        (b) -> Conversions.apiary().bucketInfo().decode(b).asBucket(this));
+        (b) -> Conversions.json().bucketInfo().decode(b).asBucket(this));
   }
 
   @Override
@@ -187,7 +190,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
         firstNonNull(content, new ByteArrayInputStream(EMPTY_BYTE_ARRAY));
     // retries are not safe when the input is an InputStream, so we can't retry.
     BlobInfo info =
-        Conversions.apiary()
+        Conversions.json()
             .blobInfo()
             .decode(storageRpc.create(blobPb, inputStreamParam, optionsMap));
     return info.asBlob(this);
@@ -213,7 +216,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
             storageRpc.create(
                 blobPb, new ByteArrayInputStream(content, offset, length), optionsMap),
         (x) -> {
-          BlobInfo info1 = Conversions.apiary().blobInfo().decode(x);
+          BlobInfo info1 = Conversions.json().blobInfo().decode(x);
           return info1.asBlob(this);
         });
   }
@@ -426,7 +429,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
                   : Iterables.transform(
                       result.y(),
                       bucketPb ->
-                          Conversions.apiary()
+                          Conversions.json()
                               .bucketInfo()
                               .decode(bucketPb)
                               .asBucket(serviceOptions.getService()));
@@ -453,7 +456,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
                   : Iterables.transform(
                       result.y(),
                       storageObject -> {
-                        BlobInfo info = Conversions.apiary().blobInfo().decode(storageObject);
+                        BlobInfo info = Conversions.json().blobInfo().decode(storageObject);
                         return info.asBlob(serviceOptions.getService());
                       });
           return new PageImpl<>(
@@ -475,7 +478,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
       return run(
           algorithm,
           () -> storageRpc.patch(bucketPb, optionsMap),
-          (x) -> Conversions.apiary().bucketInfo().decode(x).asBucket(this));
+          (x) -> Conversions.json().bucketInfo().decode(x).asBucket(this));
     }
   }
 
@@ -495,7 +498,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
           algorithm,
           () -> storageRpc.patch(pb, optionsMap),
           (x) -> {
-            BlobInfo info = Conversions.apiary().blobInfo().decode(x);
+            BlobInfo info = Conversions.json().blobInfo().decode(x);
             return info.asBlob(this);
           });
     }
@@ -560,7 +563,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
         algorithm,
         () -> storageRpc.compose(sources, targetPb, targetOptions),
         (x) -> {
-          BlobInfo info = Conversions.apiary().blobInfo().decode(x);
+          BlobInfo info = Conversions.json().blobInfo().decode(x);
           return info.asBlob(this);
         });
   }
@@ -604,9 +607,25 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
     Opts<ObjectSourceOpt> unwrap = Opts.unwrap(options);
     Opts<ObjectSourceOpt> resolve = unwrap.resolveFrom(blob);
     ImmutableMap<StorageRpc.Option, ?> optionsMap = resolve.getRpcOptions();
-    ResultRetryAlgorithm<?> algorithm =
-        retryAlgorithmManager.getForObjectsGet(storageObject, optionsMap);
-    return run(algorithm, () -> storageRpc.load(storageObject, optionsMap), Function.identity());
+    boolean autoGzipDecompression =
+        Utils.isAutoGzipDecompression(resolve, /*defaultWhenUndefined=*/ true);
+    UnbufferedReadableByteChannelSession<StorageObject> session =
+        ResumableMedia.http()
+            .read()
+            .byteChannel(BlobReadChannelContext.from(this))
+            .setAutoGzipDecompression(autoGzipDecompression)
+            .unbuffered()
+            .setApiaryReadRequest(
+                new ApiaryReadRequest(storageObject, optionsMap, ByteRangeSpec.nullRange()))
+            .build();
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    try (UnbufferedReadableByteChannel r = session.open();
+        WritableByteChannel w = Channels.newChannel(baos)) {
+      ByteStreams.copy(r, w);
+    } catch (IOException e) {
+      throw StorageException.translate(e);
+    }
+    return baos.toByteArray();
   }
 
   @Override
@@ -622,7 +641,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   @Override
   public StorageReadChannel reader(BlobId blob, BlobSourceOption... options) {
     Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob);
-    StorageObject storageObject = Conversions.apiary().blobId().encode(blob);
+    StorageObject storageObject = Conversions.json().blobId().encode(blob);
     ImmutableMap<StorageRpc.Option, ?> optionsMap = opts.getRpcOptions();
     return new BlobReadChannelV2(storageObject, optionsMap, BlobReadChannelContext.from(this));
   }
@@ -638,19 +657,26 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   @Override
   public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption... options) {
-    final CountingOutputStream countingOutputStream = new CountingOutputStream(outputStream);
     final StorageObject pb = codecs.blobId().encode(blob);
-    ImmutableMap<StorageRpc.Option, ?> optionsMap =
-        Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
-    ResultRetryAlgorithm<?> algorithm = retryAlgorithmManager.getForObjectsGet(pb, optionsMap);
-    run(
-        algorithm,
-        callable(
-            () -> {
-              storageRpc.read(
-                  pb, optionsMap, countingOutputStream.getCount(), countingOutputStream);
-            }),
-        Function.identity());
+    Opts<ObjectSourceOpt> resolve = Opts.unwrap(options).resolveFrom(blob);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = resolve.getRpcOptions();
+    boolean autoGzipDecompression =
+        Utils.isAutoGzipDecompression(resolve, /*defaultWhenUndefined=*/ true);
+    UnbufferedReadableByteChannelSession<StorageObject> session =
+        ResumableMedia.http()
+            .read()
+            .byteChannel(BlobReadChannelContext.from(this))
+            .setAutoGzipDecompression(autoGzipDecompression)
+            .unbuffered()
+            .setApiaryReadRequest(new ApiaryReadRequest(pb, optionsMap, ByteRangeSpec.nullRange()))
+            .build();
+    // don't close the provided stream
+    WritableByteChannel w = Channels.newChannel(outputStream);
+    try (UnbufferedReadableByteChannel r = session.open()) {
+      ByteStreams.copy(r, w);
+    } catch (IOException e) {
+      throw StorageException.translate(e);
+    }
   }
 
   @Override
@@ -1452,21 +1478,21 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
     return run(
         algorithm,
         () -> storageRpc.getIamPolicy(bucket, optionsMap),
-        apiPolicy -> Conversions.apiary().policyCodec().decode(apiPolicy));
+        apiPolicy -> Conversions.json().policyCodec().decode(apiPolicy));
   }
 
   @Override
   public Policy setIamPolicy(
       final String bucket, final Policy policy, BucketSourceOption... options) {
     com.google.api.services.storage.model.Policy pb =
-        Conversions.apiary().policyCodec().encode(policy);
+        Conversions.json().policyCodec().encode(policy);
     ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForBucketsSetIamPolicy(bucket, pb, optionsMap);
     return run(
         algorithm,
         () -> storageRpc.setIamPolicy(bucket, pb, optionsMap),
-        apiPolicy -> Conversions.apiary().policyCodec().decode(apiPolicy));
+        apiPolicy -> Conversions.json().policyCodec().decode(apiPolicy));
   }
 
   @Override
@@ -1500,7 +1526,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
     return run(
         algorithm,
         () -> storageRpc.lockRetentionPolicy(bucketPb, optionsMap),
-        (x) -> Conversions.apiary().bucketInfo().decode(x).asBucket(this));
+        (x) -> Conversions.json().bucketInfo().decode(x).asBucket(this));
   }
 
   @Override
@@ -1574,7 +1600,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
         algorithm,
         () -> storageRpc.get(storedObject, optionsMap),
         (x) -> {
-          BlobInfo info = Conversions.apiary().blobInfo().decode(x);
+          BlobInfo info = Conversions.json().blobInfo().decode(x);
           return info.asBlob(this);
         });
   }
@@ -1587,6 +1613,6 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
     return run(
         algorithm,
         () -> storageRpc.get(bucketPb, optionsMap),
-        (b) -> Conversions.apiary().bucketInfo().decode(b).asBucket(this));
+        (b) -> Conversions.json().bucketInfo().decode(b).asBucket(this));
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java
index 3f632aab..0d700c46 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java
@@ -31,10 +31,6 @@ interface StorageInternal {
     throw new UnsupportedOperationException("not implemented");
   }
 
-  default StorageWriteChannel internalWriter(BlobInfo info, Opts<ObjectTargetOpt> opts) {
-    throw new UnsupportedOperationException("not implemented");
-  }
-
   default BlobInfo internalDirectUpload(BlobInfo info, Opts<ObjectTargetOpt> opts, ByteBuffer buf) {
     throw new UnsupportedOperationException("not implemented");
   }
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/AclTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/AclTest.java
index c011c285..a4d6253c 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/AclTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/AclTest.java
@@ -39,11 +39,11 @@ public class AclTest {
   private static final String ETAG = "etag";
   private static final String ID = "id";
   private static final Acl ACL = Acl.newBuilder(ENTITY, ROLE).setEtag(ETAG).setId(ID).build();
-  private static final Codec<Entity, String> CODEC_ENTITY = Conversions.apiary().entity();
+  private static final Codec<Entity, String> CODEC_ENTITY = Conversions.json().entity();
   private static final Codec<Acl, ObjectAccessControl> CODEC_ACL_OBJECT =
-      Conversions.apiary().objectAcl();
+      Conversions.json().objectAcl();
   private static final Codec<Acl, BucketAccessControl> CODEC_ACL_BUCKET =
-      Conversions.apiary().bucketAcl();
+      Conversions.json().bucketAcl();
 
   static {
   }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BackwardCompatibilityUtilsTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BackwardCompatibilityUtilsTest.java
index 46cec8e0..b19c7a28 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BackwardCompatibilityUtilsTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BackwardCompatibilityUtilsTest.java
@@ -34,7 +34,7 @@ import org.junit.Test;
 
 public final class BackwardCompatibilityUtilsTest {
   private static final Codec<DeleteRule, Rule> codec =
-      BackwardCompatibilityUtils.deleteRuleCodec.andThen(Conversions.apiary().lifecycleRule());
+      BackwardCompatibilityUtils.deleteRuleCodec.andThen(Conversions.json().lifecycleRule());
 
   @Test
   public void testDeleteRules_conversionRoundTrip_age() {
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BaseConvertablePropertyTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BaseConvertablePropertyTest.java
index f8a8ffc6..549c62a4 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BaseConvertablePropertyTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BaseConvertablePropertyTest.java
@@ -55,7 +55,7 @@ abstract class BaseConvertablePropertyTest<
 
   /** Provide the codec instance used to convert between {@code ModelT} and {@code ProtoT} */
   Codec<ModelT, ApiaryT> apiaryCodec() {
-    ApiaryConversions instance = Conversions.apiary();
+    JsonConversions instance = Conversions.json();
     return resolveCodec(instance, apiaryTType);
   }
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobIdPropertyTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobIdPropertyTest.java
index f38f76e5..52927317 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobIdPropertyTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobIdPropertyTest.java
@@ -44,7 +44,7 @@ final class BlobIdPropertyTest {
   @Property
   void codecCompatibilityRoundTrip(@ForAll(supplier = ObjectRefs.class) Object p) {
     Codec<BlobId, Object> codecG = Conversions.grpc().blobId();
-    Codec<BlobId, StorageObject> codecA = Conversions.apiary().blobId();
+    Codec<BlobId, StorageObject> codecA = Conversions.json().blobId();
 
     BlobId model = codecG.decode(p);
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobIdTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobIdTest.java
index f1c37ce1..37cb5a14 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobIdTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobIdTest.java
@@ -64,6 +64,6 @@ public class BlobIdTest {
   @Test
   public void testToPbAndFromPb() {
     compareBlobIds(
-        BLOB, Conversions.apiary().blobId().decode(Conversions.apiary().blobId().encode(BLOB)));
+        BLOB, Conversions.json().blobId().decode(Conversions.json().blobId().encode(BLOB)));
   }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobInfoTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobInfoTest.java
index d17cb865..c563c9e8 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobInfoTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobInfoTest.java
@@ -285,23 +285,23 @@ public class BlobInfoTest {
   public void testToPbAndFromPb() {
     compareCustomerEncryptions(
         CUSTOMER_ENCRYPTION,
-        Conversions.apiary()
+        Conversions.json()
             .customerEncryption()
-            .decode(Conversions.apiary().customerEncryption().encode(CUSTOMER_ENCRYPTION)));
+            .decode(Conversions.json().customerEncryption().encode(CUSTOMER_ENCRYPTION)));
     compareBlobs(
         BLOB_INFO,
-        Conversions.apiary().blobInfo().decode(Conversions.apiary().blobInfo().encode(BLOB_INFO)));
+        Conversions.json().blobInfo().decode(Conversions.json().blobInfo().encode(BLOB_INFO)));
     BlobInfo blobInfo = BlobInfo.newBuilder(BlobId.of("b", "n")).build();
     compareBlobs(
         blobInfo,
-        Conversions.apiary().blobInfo().decode(Conversions.apiary().blobInfo().encode(blobInfo)));
+        Conversions.json().blobInfo().decode(Conversions.json().blobInfo().encode(blobInfo)));
     StorageObject object =
         new StorageObject()
             .setName("n/")
             .setBucket("b")
             .setSize(BigInteger.ZERO)
             .set("isDirectory", true);
-    blobInfo = Conversions.apiary().blobInfo().decode(object);
+    blobInfo = Conversions.json().blobInfo().decode(object);
     assertEquals("b", blobInfo.getBucket());
     assertEquals("n/", blobInfo.getName());
     assertNull(blobInfo.getAcl());
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketInfoTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketInfoTest.java
index a11d989b..2e3cb4bd 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketInfoTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketInfoTest.java
@@ -236,7 +236,7 @@ public class BucketInfoTest {
   @Test
   @SuppressWarnings({"unchecked", "deprecation"})
   public void testToPbAndFromPb() throws Exception {
-    Codec<BucketInfo, Bucket> codec = Conversions.apiary().bucketInfo();
+    Codec<BucketInfo, Bucket> codec = Conversions.json().bucketInfo();
 
     Bucket encode1 = codec.encode(BUCKET_INFO);
     BucketInfo decode1 = codec.decode(encode1);
@@ -296,7 +296,7 @@ public class BucketInfoTest {
   @Test
   public void testLifecycleRules() {
     Rule deleteLifecycleRule =
-        Conversions.apiary()
+        Conversions.json()
             .lifecycleRule()
             .encode(
                 new LifecycleRule(
@@ -316,7 +316,7 @@ public class BucketInfoTest {
     assertEquals(1, deleteLifecycleRule.getCondition().getMatchesSuffix().size());
     assertEquals("xyz", deleteLifecycleRule.getCondition().getMatchesSuffix().get(0));
 
-    LifecycleRule lcr = Conversions.apiary().lifecycleRule().decode(deleteLifecycleRule);
+    LifecycleRule lcr = Conversions.json().lifecycleRule().decode(deleteLifecycleRule);
     assertEquals(LifecycleRule.DeleteLifecycleAction.TYPE, lcr.getAction().getActionType());
     assertEquals(10, lcr.getCondition().getAge().intValue());
     assertEquals(2, lcr.getCondition().getMatchesPrefix().size());
@@ -328,7 +328,7 @@ public class BucketInfoTest {
     assertTrue(lcr.getAction() instanceof DeleteLifecycleAction);
 
     Rule setStorageClassLifecycleRule =
-        Conversions.apiary()
+        Conversions.json()
             .lifecycleRule()
             .encode(
                 new LifecycleRule(
@@ -344,11 +344,11 @@ public class BucketInfoTest {
     assertTrue(setStorageClassLifecycleRule.getCondition().getIsLive());
     assertEquals(10, setStorageClassLifecycleRule.getCondition().getNumNewerVersions().intValue());
     assertTrue(
-        Conversions.apiary().lifecycleRule().decode(setStorageClassLifecycleRule).getAction()
+        Conversions.json().lifecycleRule().decode(setStorageClassLifecycleRule).getAction()
             instanceof SetStorageClassLifecycleAction);
 
     Rule lifecycleRule =
-        Conversions.apiary()
+        Conversions.json()
             .lifecycleRule()
             .encode(
                 new LifecycleRule(
@@ -374,11 +374,11 @@ public class BucketInfoTest {
     assertEquals("prefix-", lifecycleRule.getCondition().getMatchesPrefix().get(0));
     assertEquals("-suffix", lifecycleRule.getCondition().getMatchesSuffix().get(0));
     assertTrue(
-        Conversions.apiary().lifecycleRule().decode(lifecycleRule).getAction()
+        Conversions.json().lifecycleRule().decode(lifecycleRule).getAction()
             instanceof SetStorageClassLifecycleAction);
 
     Rule abortMpuLifecycleRule =
-        Conversions.apiary()
+        Conversions.json()
             .lifecycleRule()
             .encode(
                 new LifecycleRule(
@@ -386,11 +386,11 @@ public class BucketInfoTest {
                     LifecycleCondition.newBuilder().setAge(10).build()));
     assertEquals(AbortIncompleteMPUAction.TYPE, abortMpuLifecycleRule.getAction().getType());
     assertEquals(10, abortMpuLifecycleRule.getCondition().getAge().intValue());
-    LifecycleRule decode = Conversions.apiary().lifecycleRule().decode(abortMpuLifecycleRule);
+    LifecycleRule decode = Conversions.json().lifecycleRule().decode(abortMpuLifecycleRule);
     assertThat(decode.getAction()).isInstanceOf(AbortIncompleteMPUAction.class);
 
     Rule unsupportedRule =
-        Conversions.apiary()
+        Conversions.json()
             .lifecycleRule()
             .encode(
                 new LifecycleRule(
@@ -399,7 +399,7 @@ public class BucketInfoTest {
     unsupportedRule.setAction(
         unsupportedRule.getAction().setType("This action type also doesn't exist"));
 
-    Conversions.apiary()
+    Conversions.json()
         .lifecycleRule()
         .decode(
             unsupportedRule); // If this doesn't throw an exception, unsupported rules are working
@@ -408,7 +408,7 @@ public class BucketInfoTest {
   @Test
   public void testIamConfiguration() {
     Bucket.IamConfiguration iamConfiguration =
-        Conversions.apiary()
+        Conversions.json()
             .iamConfiguration()
             .encode(
                 IamConfiguration.newBuilder()
@@ -431,7 +431,7 @@ public class BucketInfoTest {
         JacksonFactory.getDefaultInstance().createJsonGenerator(stringWriter);
 
     jsonGenerator.serialize(
-        Conversions.apiary()
+        Conversions.json()
             .iamConfiguration()
             .encode(
                 IamConfiguration.newBuilder()
@@ -451,7 +451,7 @@ public class BucketInfoTest {
         new Bucket.IamConfiguration.UniformBucketLevelAccess();
     iamConfiguration.setUniformBucketLevelAccess(uniformBucketLevelAccess);
     iamConfiguration.setPublicAccessPrevention("random-string");
-    IamConfiguration fromPb = Conversions.apiary().iamConfiguration().decode(iamConfiguration);
+    IamConfiguration fromPb = Conversions.json().iamConfiguration().decode(iamConfiguration);
 
     assertEquals(PublicAccessPrevention.UNKNOWN, fromPb.getPublicAccessPrevention());
   }
@@ -459,7 +459,7 @@ public class BucketInfoTest {
   @Test
   public void testLogging() {
     Bucket.Logging logging =
-        Conversions.apiary()
+        Conversions.json()
             .logging()
             .encode(
                 BucketInfo.Logging.newBuilder()
@@ -472,27 +472,27 @@ public class BucketInfoTest {
 
   @Test
   public void testRuleMappingIsCorrect_noMutations() {
-    Bucket bucket = Conversions.apiary().bucketInfo().encode(bi().build());
+    Bucket bucket = Conversions.json().bucketInfo().encode(bi().build());
     assertNull(bucket.getLifecycle());
   }
 
   @Test
   public void testRuleMappingIsCorrect_deleteLifecycleRules() {
-    Bucket bucket = Conversions.apiary().bucketInfo().encode(bi().deleteLifecycleRules().build());
+    Bucket bucket = Conversions.json().bucketInfo().encode(bi().deleteLifecycleRules().build());
     assertEquals(EMPTY_LIFECYCLE, bucket.getLifecycle());
   }
 
   @Test
   @SuppressWarnings({"deprecation"})
   public void testRuleMappingIsCorrect_setDeleteRules_null() {
-    Bucket bucket = Conversions.apiary().bucketInfo().encode(bi().setDeleteRules(null).build());
+    Bucket bucket = Conversions.json().bucketInfo().encode(bi().setDeleteRules(null).build());
     assertNull(bucket.getLifecycle());
   }
 
   @Test
   @SuppressWarnings({"deprecation"})
   public void testRuleMappingIsCorrect_setDeleteRules_empty() {
-    Codec<BucketInfo, Bucket> codec = Conversions.apiary().bucketInfo();
+    Codec<BucketInfo, Bucket> codec = Conversions.json().bucketInfo();
     BucketInfo bucketInfo = bi().setDeleteRules(Collections.emptyList()).build();
     Bucket bucket = codec.encode(bucketInfo);
     Lifecycle actual = bucket.getLifecycle();
@@ -502,7 +502,7 @@ public class BucketInfoTest {
   @Test
   public void testRuleMappingIsCorrect_setLifecycleRules_empty() {
     Bucket bucket =
-        Conversions.apiary()
+        Conversions.json()
             .bucketInfo()
             .encode(bi().setLifecycleRules(Collections.<LifecycleRule>emptyList()).build());
     assertEquals(EMPTY_LIFECYCLE, bucket.getLifecycle());
@@ -513,9 +513,9 @@ public class BucketInfoTest {
     LifecycleRule lifecycleRule =
         new LifecycleRule(
             LifecycleAction.newDeleteAction(), LifecycleCondition.newBuilder().setAge(10).build());
-    Rule lifecycleDeleteAfter10 = Conversions.apiary().lifecycleRule().encode(lifecycleRule);
+    Rule lifecycleDeleteAfter10 = Conversions.json().lifecycleRule().encode(lifecycleRule);
     Bucket bucket =
-        Conversions.apiary()
+        Conversions.json()
             .bucketInfo()
             .encode(bi().setLifecycleRules(ImmutableList.of(lifecycleRule)).build());
     assertEquals(lifecycle(lifecycleDeleteAfter10), bucket.getLifecycle());
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BufferToDiskThenUploadTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BufferToDiskThenUploadTest.java
index 60272fec..1c963522 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BufferToDiskThenUploadTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BufferToDiskThenUploadTest.java
@@ -20,10 +20,8 @@ import static com.google.cloud.storage.TestUtils.xxd;
 import static com.google.common.truth.Truth.assertThat;
 
 import com.google.cloud.storage.BlobWriteSessionConfig.WriterFactory;
-import com.google.cloud.storage.Conversions.Decoder;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
-import com.google.storage.v2.WriteObjectResponse;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.WritableByteChannel;
@@ -39,9 +37,6 @@ import org.junit.rules.TestName;
 
 public final class BufferToDiskThenUploadTest {
 
-  private static final Decoder<WriteObjectResponse, BlobInfo>
-      WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER =
-          Conversions.grpc().blobInfo().compose(WriteObjectResponse::getResource);
   @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
   @Rule public final TestName testName = new TestName();
 
@@ -68,8 +63,7 @@ public final class BufferToDiskThenUploadTest {
               }
             },
             blobInfo,
-            Opts.empty(),
-            WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER);
+            Opts.empty());
 
     byte[] bytes = DataGenerator.base64Characters().genBytes(128);
     try (WritableByteChannel open = writeSession.open()) {
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java
index cd51a1e9..72ff6691 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java
@@ -53,18 +53,18 @@ public class CopyWriterTest {
   private static final Map<StorageRpc.Option, ?> EMPTY_OPTIONS = ImmutableMap.of();
   private static final RewriteRequest REQUEST_WITH_OBJECT =
       new StorageRpc.RewriteRequest(
-          Conversions.apiary().blobId().encode(BLOB_ID),
+          Conversions.json().blobId().encode(BLOB_ID),
           EMPTY_OPTIONS,
           true,
-          Conversions.apiary().blobInfo().encode(BLOB_INFO),
+          Conversions.json().blobInfo().encode(BLOB_INFO),
           EMPTY_OPTIONS,
           null);
   private static final RewriteRequest REQUEST_WITHOUT_OBJECT =
       new StorageRpc.RewriteRequest(
-          Conversions.apiary().blobId().encode(BLOB_ID),
+          Conversions.json().blobId().encode(BLOB_ID),
           EMPTY_OPTIONS,
           false,
-          Conversions.apiary().blobInfo().encode(BLOB_INFO),
+          Conversions.json().blobInfo().encode(BLOB_INFO),
           EMPTY_OPTIONS,
           null);
   private static final RewriteResponse RESPONSE_WITH_OBJECT =
@@ -74,7 +74,7 @@ public class CopyWriterTest {
   private static final RewriteResponse RESPONSE_WITH_OBJECT_DONE =
       new RewriteResponse(
           REQUEST_WITH_OBJECT,
-          Conversions.apiary().blobInfo().encode(RESULT_INFO),
+          Conversions.json().blobInfo().encode(RESULT_INFO),
           42L,
           true,
           "token",
@@ -82,7 +82,7 @@ public class CopyWriterTest {
   private static final RewriteResponse RESPONSE_WITHOUT_OBJECT_DONE =
       new RewriteResponse(
           REQUEST_WITHOUT_OBJECT,
-          Conversions.apiary().blobInfo().encode(RESULT_INFO),
+          Conversions.json().blobInfo().encode(RESULT_INFO),
           42L,
           true,
           "token",
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGrpcStorageImplUploadRetryTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGrpcStorageImplUploadRetryTest.java
index 12f1bd6b..2538b10f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGrpcStorageImplUploadRetryTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGrpcStorageImplUploadRetryTest.java
@@ -18,6 +18,7 @@ package com.google.cloud.storage;
 
 import static com.google.cloud.storage.ByteSizeConstants._2MiB;
 import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.gax.grpc.GrpcCallContext;
@@ -81,16 +82,19 @@ public final class ITGrpcStorageImplUploadRetryTest {
 
   @Test
   public void create_inputStream() throws Exception {
-    Resumable.FakeService service = Resumable.FakeService.create();
+    Direct.FakeService service = Direct.FakeService.create();
     try (TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
         FakeServer server = FakeServer.of(service);
         Storage s = server.getGrpcStorageOptions().getService();
         InputStream in = Channels.newInputStream(tmpFile.reader())) {
       BlobInfo info = BlobInfo.newBuilder("buck", "obj").build();
-      s.create(info, in, BlobWriteOption.doesNotExist());
+      // create uses a direct upload, once the stream is consumed there is no means for us to retry
+      // if an error happens it should be surfaced
+      StorageException se =
+          assertThrows(
+              StorageException.class, () -> s.create(info, in, BlobWriteOption.doesNotExist()));
+      assertThat(se.getCode()).isEqualTo(500);
     }
-
-    assertThat(service.returnError.get()).isFalse();
   }
 
   @Test
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/NotificationInfoTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/NotificationInfoTest.java
index c3a000fd..d60bd076 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/NotificationInfoTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/NotificationInfoTest.java
@@ -91,16 +91,16 @@ public class NotificationInfoTest {
   public void testToPbAndFromPb() {
     compareBucketsNotification(
         NOTIFICATION_INFO,
-        Conversions.apiary()
+        Conversions.json()
             .notificationInfo()
-            .decode(Conversions.apiary().notificationInfo().encode(NOTIFICATION_INFO)));
+            .decode(Conversions.json().notificationInfo().encode(NOTIFICATION_INFO)));
     NotificationInfo notificationInfo =
         NotificationInfo.of(TOPIC).toBuilder().setPayloadFormat(PayloadFormat.NONE).build();
     compareBucketsNotification(
         notificationInfo,
-        Conversions.apiary()
+        Conversions.json()
             .notificationInfo()
-            .decode(Conversions.apiary().notificationInfo().encode(notificationInfo)));
+            .decode(Conversions.json().notificationInfo().encode(notificationInfo)));
   }
 
   private void compareBucketsNotification(NotificationInfo expected, NotificationInfo actual) {
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/NotificationTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/NotificationTest.java
index 85121fbe..df8537d6 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/NotificationTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/NotificationTest.java
@@ -109,9 +109,9 @@ public class NotificationTest {
     replay(storage);
     compareBucketNotification(
         NOTIFICATION_INFO,
-        Conversions.apiary()
+        Conversions.json()
             .notificationInfo()
-            .decode(Conversions.apiary().notificationInfo().encode(NOTIFICATION_INFO))
+            .decode(Conversions.json().notificationInfo().encode(NOTIFICATION_INFO))
             .asNotification(storage));
   }
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/PackagePrivateMethodWorkarounds.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/PackagePrivateMethodWorkarounds.java
index d6c5ad0a..9b577014 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/PackagePrivateMethodWorkarounds.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/PackagePrivateMethodWorkarounds.java
@@ -42,9 +42,9 @@ public final class PackagePrivateMethodWorkarounds {
   public static Bucket bucketCopyWithStorage(Bucket b, Storage s) {
     BucketInfo.BuilderImpl builder =
         (BuilderImpl)
-            Conversions.apiary()
+            Conversions.json()
                 .bucketInfo()
-                .decode(Conversions.apiary().bucketInfo().encode(b))
+                .decode(Conversions.json().bucketInfo().encode(b))
                 .toBuilder();
     return new Bucket(s, builder);
   }
@@ -52,9 +52,9 @@ public final class PackagePrivateMethodWorkarounds {
   public static Blob blobCopyWithStorage(Blob b, Storage s) {
     BlobInfo.BuilderImpl builder =
         (BlobInfo.BuilderImpl)
-            Conversions.apiary()
+            Conversions.json()
                 .blobInfo()
-                .decode(Conversions.apiary().blobInfo().encode(b))
+                .decode(Conversions.json().blobInfo().encode(b))
                 .toBuilder();
     return new Blob(s, builder);
   }
@@ -64,7 +64,7 @@ public final class PackagePrivateMethodWorkarounds {
       if (w instanceof BlobWriteChannelV2) {
         BlobWriteChannelV2 blobWriteChannel = (BlobWriteChannelV2) w;
         return Optional.ofNullable(blobWriteChannel.getResolvedObject())
-            .map(Conversions.apiary().blobInfo()::decode);
+            .map(Conversions.json().blobInfo()::decode);
       } else if (w instanceof GrpcBlobWriteChannel) {
         GrpcBlobWriteChannel grpcBlobWriteChannel = (GrpcBlobWriteChannel) w;
         return Optional.of(grpcBlobWriteChannel.getObject())
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/PolicyHelperTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/PolicyHelperTest.java
index 1dd70d38..649de103 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/PolicyHelperTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/PolicyHelperTest.java
@@ -57,9 +57,9 @@ public class PolicyHelperTest {
             .setEtag(ETAG)
             .setVersion(1);
 
-    Policy actualLibPolicy = Conversions.apiary().policyCodec().decode(apiPolicy);
+    Policy actualLibPolicy = Conversions.json().policyCodec().decode(apiPolicy);
     com.google.api.services.storage.model.Policy actualApiPolicy =
-        Conversions.apiary().policyCodec().encode(libPolicy);
+        Conversions.json().policyCodec().encode(libPolicy);
 
     assertEquals(libPolicy, actualLibPolicy);
     assertTrue(new ApiPolicyMatcher(apiPolicy).matches(actualApiPolicy));
@@ -73,7 +73,7 @@ public class PolicyHelperTest {
             .setBindings(bindings)
             .setEtag(ETAG)
             .setVersion(1);
-    Policy policy = Conversions.apiary().policyCodec().decode(apiPolicy);
+    Policy policy = Conversions.json().policyCodec().decode(apiPolicy);
     assertEquals(policy.getBindings().size(), 0);
   }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
index 48d3e4ba..eab0e78e 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
@@ -261,7 +261,7 @@ public class SerializationTest extends BaseSerializationTest {
         new BlobWriteChannelV2(
             BlobReadChannelContext.from(options),
             JsonResumableWrite.of(
-                Conversions.apiary().blobInfo().encode(BlobInfo.newBuilder("b", "n").build()),
+                Conversions.json().blobInfo().encode(BlobInfo.newBuilder("b", "n").build()),
                 ImmutableMap.of(),
                 "upload-id",
                 0));
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ServiceAccountTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ServiceAccountTest.java
index 225ea26e..321e64d6 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ServiceAccountTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ServiceAccountTest.java
@@ -33,9 +33,9 @@ public class ServiceAccountTest {
   public void testToAndFromPb() {
     compareServiceAccount(
         SERVICE_ACCOUNT,
-        Conversions.apiary()
+        Conversions.json()
             .serviceAccount()
-            .decode(Conversions.apiary().serviceAccount().encode(SERVICE_ACCOUNT)));
+            .decode(Conversions.json().serviceAccount().encode(SERVICE_ACCOUNT)));
   }
 
   public void compareServiceAccount(ServiceAccount expected, ServiceAccount value) {
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java
index d91b1e74..46f2461f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java
@@ -93,7 +93,7 @@ public class StorageBatchTest {
     EasyMock.reset(batchMock);
     Capture<RpcBatch.Callback<Void>> callback = Capture.newInstance();
     batchMock.addDelete(
-        EasyMock.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO)),
+        EasyMock.eq(Conversions.json().blobInfo().encode(BLOB_INFO)),
         EasyMock.capture(callback),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     EasyMock.replay(batchMock);
@@ -123,7 +123,7 @@ public class StorageBatchTest {
     Capture<RpcBatch.Callback<Void>> callback = Capture.newInstance();
     Capture<Map<StorageRpc.Option, Object>> capturedOptions = Capture.newInstance();
     batchMock.addDelete(
-        EasyMock.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO)),
+        EasyMock.eq(Conversions.json().blobInfo().encode(BLOB_INFO)),
         EasyMock.capture(callback),
         EasyMock.capture(capturedOptions));
     EasyMock.replay(batchMock);
@@ -143,7 +143,7 @@ public class StorageBatchTest {
     EasyMock.reset(batchMock);
     Capture<RpcBatch.Callback<StorageObject>> callback = Capture.newInstance();
     batchMock.addPatch(
-        EasyMock.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO)),
+        EasyMock.eq(Conversions.json().blobInfo().encode(BLOB_INFO)),
         EasyMock.capture(callback),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     EasyMock.replay(batchMock);
@@ -177,7 +177,7 @@ public class StorageBatchTest {
     Capture<RpcBatch.Callback<StorageObject>> callback = Capture.newInstance();
     Capture<Map<StorageRpc.Option, Object>> capturedOptions = Capture.newInstance();
     batchMock.addPatch(
-        EasyMock.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO_COMPLETE)),
+        EasyMock.eq(Conversions.json().blobInfo().encode(BLOB_INFO_COMPLETE)),
         EasyMock.capture(callback),
         EasyMock.capture(capturedOptions));
     EasyMock.replay(batchMock, storage, optionsMock);
@@ -191,7 +191,7 @@ public class StorageBatchTest {
     assertEquals(42L, capturedOptions.getValue().get(StorageRpc.Option.IF_GENERATION_MATCH));
     assertEquals(42L, capturedOptions.getValue().get(StorageRpc.Option.IF_METAGENERATION_MATCH));
     RpcBatch.Callback<StorageObject> capturedCallback = callback.getValue();
-    capturedCallback.onSuccess(Conversions.apiary().blobInfo().encode(BLOB_INFO));
+    capturedCallback.onSuccess(Conversions.json().blobInfo().encode(BLOB_INFO));
     assertEquals(new Blob(storage, new Blob.BuilderImpl(BLOB_INFO)), batchResult.get());
   }
 
@@ -200,7 +200,7 @@ public class StorageBatchTest {
     EasyMock.reset(batchMock);
     Capture<RpcBatch.Callback<StorageObject>> callback = Capture.newInstance();
     batchMock.addGet(
-        EasyMock.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO)),
+        EasyMock.eq(Conversions.json().blobInfo().encode(BLOB_INFO)),
         EasyMock.capture(callback),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     EasyMock.replay(batchMock);
@@ -234,7 +234,7 @@ public class StorageBatchTest {
     Capture<RpcBatch.Callback<StorageObject>> callback = Capture.newInstance();
     Capture<Map<StorageRpc.Option, Object>> capturedOptions = Capture.newInstance();
     batchMock.addGet(
-        EasyMock.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO)),
+        EasyMock.eq(Conversions.json().blobInfo().encode(BLOB_INFO)),
         EasyMock.capture(callback),
         EasyMock.capture(capturedOptions));
     EasyMock.replay(storage, batchMock, optionsMock);
@@ -245,7 +245,7 @@ public class StorageBatchTest {
     assertThat(options).containsEntry(StorageRpc.Option.IF_GENERATION_MATCH, 42L);
     assertThat(options).containsEntry(StorageRpc.Option.IF_METAGENERATION_MATCH, 42L);
     RpcBatch.Callback<StorageObject> capturedCallback = callback.getValue();
-    capturedCallback.onSuccess(Conversions.apiary().blobInfo().encode(BLOB_INFO));
+    capturedCallback.onSuccess(Conversions.json().blobInfo().encode(BLOB_INFO));
     assertEquals(new Blob(storage, new Blob.BuilderImpl(BLOB_INFO)), batchResult.get());
   }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java
index 1d145340..65602f67 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java
@@ -369,10 +369,10 @@ public class StorageImplMockitoTest {
 
   @Test
   public void testCreateBucket() {
-    doReturn(Conversions.apiary().bucketInfo().encode(BUCKET_INFO1))
+    doReturn(Conversions.json().bucketInfo().encode(BUCKET_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
-        .create(Conversions.apiary().bucketInfo().encode(BUCKET_INFO1), EMPTY_RPC_OPTIONS);
+        .create(Conversions.json().bucketInfo().encode(BUCKET_INFO1), EMPTY_RPC_OPTIONS);
     initializeService();
     Bucket bucket = storage.create(BUCKET_INFO1);
     assertEquals(expectedBucket1, bucket);
@@ -380,10 +380,10 @@ public class StorageImplMockitoTest {
 
   @Test
   public void testCreateBucketWithOptions() {
-    doReturn(Conversions.apiary().bucketInfo().encode(BUCKET_INFO1))
+    doReturn(Conversions.json().bucketInfo().encode(BUCKET_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
-        .create(Conversions.apiary().bucketInfo().encode(BUCKET_INFO1), BUCKET_TARGET_OPTIONS);
+        .create(Conversions.json().bucketInfo().encode(BUCKET_INFO1), BUCKET_TARGET_OPTIONS);
     initializeService();
     Bucket bucket =
         storage.create(BUCKET_INFO1, BUCKET_TARGET_METAGENERATION, BUCKET_TARGET_PREDEFINED_ACL);
@@ -394,7 +394,7 @@ public class StorageImplMockitoTest {
   public void testCreateBucketFailure() {
     doThrow(STORAGE_FAILURE)
         .when(storageRpcMock)
-        .create(Conversions.apiary().bucketInfo().encode(BUCKET_INFO1), EMPTY_RPC_OPTIONS);
+        .create(Conversions.json().bucketInfo().encode(BUCKET_INFO1), EMPTY_RPC_OPTIONS);
     initializeService();
     try {
       storage.create(BUCKET_INFO1);
@@ -406,12 +406,11 @@ public class StorageImplMockitoTest {
 
   @Test
   public void testGetBucket() {
-    doReturn(Conversions.apiary().bucketInfo().encode(BUCKET_INFO1))
+    doReturn(Conversions.json().bucketInfo().encode(BUCKET_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .get(
-            Conversions.apiary().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1)),
-            EMPTY_RPC_OPTIONS);
+            Conversions.json().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1)), EMPTY_RPC_OPTIONS);
     initializeService();
     Bucket bucket = storage.get(BUCKET_NAME1);
     assertEquals(expectedBucket1, bucket);
@@ -419,11 +418,11 @@ public class StorageImplMockitoTest {
 
   @Test
   public void testGetBucketWithOptions() {
-    doReturn(Conversions.apiary().bucketInfo().encode(BUCKET_INFO1))
+    doReturn(Conversions.json().bucketInfo().encode(BUCKET_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .get(
-            Conversions.apiary().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1)),
+            Conversions.json().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1)),
             BUCKET_GET_OPTIONS);
     initializeService();
     Bucket bucket = storage.get(BUCKET_NAME1, BUCKET_GET_METAGENERATION);
@@ -434,11 +433,11 @@ public class StorageImplMockitoTest {
   public void testGetBucketWithSelectedFields() {
     ArgumentCaptor<Map<StorageRpc.Option, Object>> capturedOptions =
         ArgumentCaptor.forClass(Map.class);
-    doReturn(Conversions.apiary().bucketInfo().encode(BUCKET_INFO1))
+    doReturn(Conversions.json().bucketInfo().encode(BUCKET_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .get(
-            Mockito.eq(Conversions.apiary().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1))),
+            Mockito.eq(Conversions.json().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1))),
             capturedOptions.capture());
     initializeService();
     Bucket bucket = storage.get(BUCKET_NAME1, BUCKET_GET_METAGENERATION, BUCKET_GET_FIELDS);
@@ -457,11 +456,11 @@ public class StorageImplMockitoTest {
   public void testGetBucketWithEmptyFields() {
     ArgumentCaptor<Map<StorageRpc.Option, Object>> capturedOptions =
         ArgumentCaptor.forClass(Map.class);
-    doReturn(Conversions.apiary().bucketInfo().encode(BUCKET_INFO1))
+    doReturn(Conversions.json().bucketInfo().encode(BUCKET_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .get(
-            Mockito.eq(Conversions.apiary().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1))),
+            Mockito.eq(Conversions.json().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1))),
             capturedOptions.capture());
     initializeService();
     Bucket bucket = storage.get(BUCKET_NAME1, BUCKET_GET_METAGENERATION, BUCKET_GET_EMPTY_FIELDS);
@@ -479,8 +478,7 @@ public class StorageImplMockitoTest {
     doThrow(STORAGE_FAILURE)
         .when(storageRpcMock)
         .get(
-            Conversions.apiary().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1)),
-            EMPTY_RPC_OPTIONS);
+            Conversions.json().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1)), EMPTY_RPC_OPTIONS);
     initializeService();
     try {
       storage.get(BUCKET_NAME1);
@@ -492,11 +490,11 @@ public class StorageImplMockitoTest {
 
   @Test
   public void testGetBlob() {
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .get(
-            Conversions.apiary().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1)),
+            Conversions.json().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1)),
             EMPTY_RPC_OPTIONS);
     initializeService();
     Blob blob = storage.get(BUCKET_NAME1, BLOB_NAME1);
@@ -505,11 +503,11 @@ public class StorageImplMockitoTest {
 
   @Test
   public void testGetBlobWithOptions() {
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .get(
-            Conversions.apiary().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1)),
+            Conversions.json().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1)),
             BLOB_GET_OPTIONS);
     initializeService();
     Blob blob = storage.get(BUCKET_NAME1, BLOB_NAME1, BLOB_GET_METAGENERATION, BLOB_GET_GENERATION);
@@ -518,10 +516,10 @@ public class StorageImplMockitoTest {
 
   @Test
   public void testGetBlobWithOptionsFromBlobId() {
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
-        .get(Conversions.apiary().blobId().encode(BLOB_INFO1.getBlobId()), BLOB_GET_OPTIONS);
+        .get(Conversions.json().blobId().encode(BLOB_INFO1.getBlobId()), BLOB_GET_OPTIONS);
     initializeService();
     Blob blob =
         storage.get(
@@ -533,11 +531,11 @@ public class StorageImplMockitoTest {
   public void testGetBlobWithSelectedFields() {
     ArgumentCaptor<Map<StorageRpc.Option, Object>> capturedOptions =
         ArgumentCaptor.forClass(Map.class);
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .get(
-            Mockito.eq(Conversions.apiary().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1))),
+            Mockito.eq(Conversions.json().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1))),
             capturedOptions.capture());
     initializeService();
     Blob blob =
@@ -566,11 +564,11 @@ public class StorageImplMockitoTest {
   public void testGetBlobWithEmptyFields() {
     ArgumentCaptor<Map<StorageRpc.Option, Object>> capturedOptions =
         ArgumentCaptor.forClass(Map.class);
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .get(
-            Mockito.eq(Conversions.apiary().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1))),
+            Mockito.eq(Conversions.json().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1))),
             capturedOptions.capture());
     initializeService();
     Blob blob =
@@ -598,7 +596,7 @@ public class StorageImplMockitoTest {
     doThrow(STORAGE_FAILURE)
         .when(storageRpcMock)
         .get(
-            Conversions.apiary().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1)),
+            Conversions.json().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1)),
             EMPTY_RPC_OPTIONS);
     initializeService();
     try {
@@ -622,11 +620,11 @@ public class StorageImplMockitoTest {
   public void testCreateBlob() throws IOException {
     ArgumentCaptor<ByteArrayInputStream> capturedStream =
         ArgumentCaptor.forClass(ByteArrayInputStream.class);
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .create(
-            Mockito.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO_WITH_HASHES)),
+            Mockito.eq(Conversions.json().blobInfo().encode(BLOB_INFO_WITH_HASHES)),
             capturedStream.capture(),
             Mockito.eq(EMPTY_RPC_OPTIONS));
     initializeService();
@@ -641,12 +639,12 @@ public class StorageImplMockitoTest {
   public void testCreateBlobWithSubArrayFromByteArray() throws IOException {
     ArgumentCaptor<ByteArrayInputStream> capturedStream =
         ArgumentCaptor.forClass(ByteArrayInputStream.class);
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .create(
             Mockito.eq(
-                Conversions.apiary()
+                Conversions.json()
                     .blobInfo()
                     .encode(
                         BLOB_INFO1
@@ -673,10 +671,10 @@ public class StorageImplMockitoTest {
     ArgumentCaptor<ByteArrayInputStream> capturedStream =
         ArgumentCaptor.forClass(ByteArrayInputStream.class);
 
-    StorageObject storageObject = Conversions.apiary().blobInfo().encode(BLOB_INFO_WITH_HASHES);
+    StorageObject storageObject = Conversions.json().blobInfo().encode(BLOB_INFO_WITH_HASHES);
 
     doThrow(new StorageException(500, "internalError"))
-        .doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+        .doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .create(
@@ -709,12 +707,12 @@ public class StorageImplMockitoTest {
     ArgumentCaptor<ByteArrayInputStream> capturedStream =
         ArgumentCaptor.forClass(ByteArrayInputStream.class);
 
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .create(
             Mockito.eq(
-                Conversions.apiary()
+                Conversions.json()
                     .blobInfo()
                     .encode(
                         BLOB_INFO1
@@ -738,11 +736,11 @@ public class StorageImplMockitoTest {
     ArgumentCaptor<ByteArrayInputStream> capturedStream =
         ArgumentCaptor.forClass(ByteArrayInputStream.class);
 
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .create(
-            Mockito.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO_WITH_HASHES)),
+            Mockito.eq(Conversions.json().blobInfo().encode(BLOB_INFO_WITH_HASHES)),
             capturedStream.capture(),
             Mockito.eq(BLOB_TARGET_OPTIONS_CREATE));
     initializeService();
@@ -763,11 +761,11 @@ public class StorageImplMockitoTest {
     ArgumentCaptor<ByteArrayInputStream> capturedStream =
         ArgumentCaptor.forClass(ByteArrayInputStream.class);
 
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .create(
-            Mockito.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO_WITH_HASHES)),
+            Mockito.eq(Conversions.json().blobInfo().encode(BLOB_INFO_WITH_HASHES)),
             capturedStream.capture(),
             Mockito.eq(BLOB_TARGET_OPTIONS_CREATE_DISABLE_GZIP_CONTENT));
     initializeService();
@@ -782,12 +780,12 @@ public class StorageImplMockitoTest {
     ArgumentCaptor<ByteArrayInputStream> capturedStream =
         ArgumentCaptor.forClass(ByteArrayInputStream.class);
 
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
-        .doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
+        .doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .create(
-            Mockito.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO_WITH_HASHES)),
+            Mockito.eq(Conversions.json().blobInfo().encode(BLOB_INFO_WITH_HASHES)),
             capturedStream.capture(),
             Mockito.eq(ENCRYPTION_KEY_OPTIONS));
     initializeService();
@@ -808,12 +806,12 @@ public class StorageImplMockitoTest {
     ArgumentCaptor<ByteArrayInputStream> capturedStream =
         ArgumentCaptor.forClass(ByteArrayInputStream.class);
 
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
-        .doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
+        .doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .create(
-            Mockito.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO_WITH_HASHES)),
+            Mockito.eq(Conversions.json().blobInfo().encode(BLOB_INFO_WITH_HASHES)),
             capturedStream.capture(),
             Mockito.eq(KMS_KEY_NAME_OPTIONS));
     initializeService();
@@ -837,11 +835,11 @@ public class StorageImplMockitoTest {
     ByteArrayInputStream fileStream = new ByteArrayInputStream(BLOB_CONTENT);
 
     // verify that md5 and crc32c are cleared if present when calling create
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .create(
-            Mockito.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO_WITHOUT_HASHES)),
+            Mockito.eq(Conversions.json().blobInfo().encode(BLOB_INFO_WITHOUT_HASHES)),
             capturedStream.capture(),
             Mockito.eq(EMPTY_RPC_OPTIONS));
     initializeService();
@@ -860,11 +858,11 @@ public class StorageImplMockitoTest {
 
     // verify that md5 and crc32c are cleared if present when calling create
     ByteArrayInputStream fileStream = new ByteArrayInputStream(BLOB_CONTENT);
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .create(
-            Mockito.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO_WITHOUT_HASHES)),
+            Mockito.eq(Conversions.json().blobInfo().encode(BLOB_INFO_WITHOUT_HASHES)),
             capturedStream.capture(),
             Mockito.eq(BLOB_TARGET_OPTIONS_CREATE_DISABLE_GZIP_CONTENT));
     initializeService();
@@ -883,12 +881,12 @@ public class StorageImplMockitoTest {
     ByteArrayInputStream fileStream = new ByteArrayInputStream(BLOB_CONTENT);
 
     // verify that md5 and crc32c are cleared if present when calling create
-    doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
-        .doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
+    doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
+        .doReturn(Conversions.json().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
         .create(
-            Conversions.apiary().blobInfo().encode(BLOB_INFO_WITHOUT_HASHES),
+            Conversions.json().blobInfo().encode(BLOB_INFO_WITHOUT_HASHES),
             fileStream,
             ENCRYPTION_KEY_OPTIONS);
     initializeService();
@@ -913,7 +911,7 @@ public class StorageImplMockitoTest {
     doThrow(internalErrorException)
         .when(storageRpcMock)
         .create(
-            Conversions.apiary().blobInfo().encode(BLOB_INFO_WITHOUT_HASHES),
+            Conversions.json().blobInfo().encode(BLOB_INFO_WITHOUT_HASHES),
             fileStream,
             EMPTY_RPC_OPTIONS);
 
@@ -967,7 +965,7 @@ public class StorageImplMockitoTest {
     doReturn(uploadId)
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
-        .open(Conversions.apiary().blobInfo().encode(blobInfo), rpcOptions);
+        .open(Conversions.json().blobInfo().encode(blobInfo), rpcOptions);
 
     doReturn(storageObject)
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
@@ -975,7 +973,7 @@ public class StorageImplMockitoTest {
         .writeWithResponse(uploadId, buffer, 0, 0L, bytes.length, true);
 
     initializeService();
-    BlobInfo info = Conversions.apiary().blobInfo().decode(storageObject);
+    BlobInfo info = Conversions.json().blobInfo().decode(storageObject);
     expectedUpdated = info.asBlob(storage);
     return blobInfo;
   }
@@ -986,7 +984,7 @@ public class StorageImplMockitoTest {
     ImmutableList<BucketInfo> bucketInfoList = ImmutableList.of(BUCKET_INFO1, BUCKET_INFO2);
     Tuple<String, Iterable<com.google.api.services.storage.model.Bucket>> result =
         Tuple.of(
-            cursor, Iterables.transform(bucketInfoList, Conversions.apiary().bucketInfo()::encode));
+            cursor, Iterables.transform(bucketInfoList, Conversions.json().bucketInfo()::encode));
 
     doReturn(result)
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
@@ -1020,7 +1018,7 @@ public class StorageImplMockitoTest {
     ImmutableList<BucketInfo> bucketInfoList = ImmutableList.of(BUCKET_INFO1, BUCKET_INFO2);
     Tuple<String, Iterable<com.google.api.services.storage.model.Bucket>> result =
         Tuple.of(
-            cursor, Iterables.transform(bucketInfoList, Conversions.apiary().bucketInfo()::encode));
+            cursor, Iterables.transform(bucketInfoList, Conversions.json().bucketInfo()::encode));
 
     doReturn(result)
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
@@ -1051,8 +1049,7 @@ public class StorageImplMockitoTest {
     String cursor = "cursor";
     ImmutableList<BlobInfo> blobInfoList = ImmutableList.of(BLOB_INFO1, BLOB_INFO2);
     Tuple<String, Iterable<com.google.api.services.storage.model.StorageObject>> result =
-        Tuple.of(
-            cursor, Iterables.transform(blobInfoList, Conversions.apiary().blobInfo()::encode));
+        Tuple.of(cursor, Iterables.transform(blobInfoList, Conversions.json().blobInfo()::encode));
 
     doReturn(result)
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
@@ -1087,8 +1084,7 @@ public class StorageImplMockitoTest {
     String cursor = "cursor";
     ImmutableList<BlobInfo> blobInfoList = ImmutableList.of(BLOB_INFO1, BLOB_INFO2);
     Tuple<String, Iterable<com.google.api.services.storage.model.StorageObject>> result =
-        Tuple.of(
-            cursor, Iterables.transform(blobInfoList, Conversions.apiary().blobInfo()::encode));
+        Tuple.of(cursor, Iterables.transform(blobInfoList, Conversions.json().blobInfo()::encode));
     doReturn(result)
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
@@ -1107,8 +1103,7 @@ public class StorageImplMockitoTest {
     Map<StorageRpc.Option, ?> options = ImmutableMap.of(StorageRpc.Option.DELIMITER, "/");
     ImmutableList<BlobInfo> blobInfoList = ImmutableList.of(BLOB_INFO1, BLOB_INFO2);
     Tuple<String, Iterable<com.google.api.services.storage.model.StorageObject>> result =
-        Tuple.of(
-            cursor, Iterables.transform(blobInfoList, Conversions.apiary().blobInfo()::encode));
+        Tuple.of(cursor, Iterables.transform(blobInfoList, Conversions.json().blobInfo()::encode));
     doReturn(result)
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
@@ -1128,8 +1123,7 @@ public class StorageImplMockitoTest {
     Map<StorageRpc.Option, ?> options = ImmutableMap.of(StorageRpc.Option.DELIMITER, delimiter);
     ImmutableList<BlobInfo> blobInfoList = ImmutableList.of(BLOB_INFO1, BLOB_INFO2);
     Tuple<String, Iterable<com.google.api.services.storage.model.StorageObject>> result =
-        Tuple.of(
-            cursor, Iterables.transform(blobInfoList, Conversions.apiary().blobInfo()::encode));
+        Tuple.of(cursor, Iterables.transform(blobInfoList, Conversions.json().blobInfo()::encode));
     doReturn(result)
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
@@ -1152,8 +1146,7 @@ public class StorageImplMockitoTest {
             StorageRpc.Option.START_OFF_SET, startOffset, StorageRpc.Option.END_OFF_SET, endOffset);
     ImmutableList<BlobInfo> blobInfoList = ImmutableList.of(BLOB_INFO1, BLOB_INFO2);
     Tuple<String, Iterable<com.google.api.services.storage.model.StorageObject>> result =
-        Tuple.of(
-            cursor, Iterables.transform(blobInfoList, Conversions.apiary().blobInfo()::encode));
+        Tuple.of(cursor, Iterables.transform(blobInfoList, Conversions.json().blobInfo()::encode));
     doReturn(result)
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
@@ -1177,8 +1170,7 @@ public class StorageImplMockitoTest {
     Map<StorageRpc.Option, ?> options = ImmutableMap.of(StorageRpc.Option.MATCH_GLOB, matchGlob);
     ImmutableList<BlobInfo> blobInfoList = ImmutableList.of(BLOB_INFO1, BLOB_INFO2);
     Tuple<String, Iterable<com.google.api.services.storage.model.StorageObject>> result =
-        Tuple.of(
-            cursor, Iterables.transform(blobInfoList, Conversions.apiary().blobInfo()::encode));
+        Tuple.of(cursor, Iterables.transform(blobInfoList, Conversions.json().blobInfo()::encode));
     doReturn(result)
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
@@ -1205,10 +1197,10 @@ public class StorageImplMockitoTest {
 
   @Test
   public void testCreateNotification() {
-    doReturn(Conversions.apiary().notificationInfo().encode(NOTIFICATION_INFO_01))
+    doReturn(Conversions.json().notificationInfo().encode(NOTIFICATION_INFO_01))
         .when(storageRpcMock)
         .createNotification(
-            BUCKET_NAME1, Conversions.apiary().notificationInfo().encode(NOTIFICATION_INFO_01));
+            BUCKET_NAME1, Conversions.json().notificationInfo().encode(NOTIFICATION_INFO_01));
     initializeService();
     Notification notification = storage.createNotification(BUCKET_NAME1, NOTIFICATION_INFO_01);
     verifyBucketNotification(notification);
@@ -1216,7 +1208,7 @@ public class StorageImplMockitoTest {
 
   @Test
   public void testGetNotification() {
-    doReturn(Conversions.apiary().notificationInfo().encode(NOTIFICATION_INFO_01))
+    doReturn(Conversions.json().notificationInfo().encode(NOTIFICATION_INFO_01))
         .when(storageRpcMock)
         .getNotification(BUCKET_NAME1, GENERATED_ID);
     initializeService();
@@ -1228,8 +1220,8 @@ public class StorageImplMockitoTest {
   public void testListNotification() {
     doReturn(
             Arrays.asList(
-                Conversions.apiary().notificationInfo().encode(NOTIFICATION_INFO_01),
-                Conversions.apiary().notificationInfo().encode(NOTIFICATION_INFO_02)))
+                Conversions.json().notificationInfo().encode(NOTIFICATION_INFO_01),
+                Conversions.json().notificationInfo().encode(NOTIFICATION_INFO_02)))
         .when(storageRpcMock)
         .listNotifications(BUCKET_NAME1);
     initializeService();
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java
index f4e497be..2fd5501d 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java
@@ -19,12 +19,6 @@ package com.google.cloud.storage;
 import static com.google.cloud.storage.SignedUrlEncodingHelper.Rfc3986UriEncode;
 import static com.google.cloud.storage.testing.ApiPolicyMatcher.eqApiPolicy;
 import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.easymock.EasyMock.anyObject;
-import static org.easymock.EasyMock.eq;
-import static org.easymock.EasyMock.expect;
-import static org.easymock.EasyMock.getCurrentArguments;
-import static org.easymock.EasyMock.replay;
-import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
@@ -32,12 +26,10 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
 
 import com.google.api.client.googleapis.json.GoogleJsonError;
 import com.google.api.client.util.DateTime;
 import com.google.api.core.ApiClock;
-import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.services.storage.model.Policy.Bindings;
 import com.google.api.services.storage.model.StorageObject;
 import com.google.api.services.storage.model.TestIamPermissionsResponse;
@@ -59,12 +51,9 @@ import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.io.BaseEncoding;
-import java.io.File;
-import java.io.OutputStream;
 import java.io.UnsupportedEncodingException;
 import java.net.URL;
 import java.net.URLDecoder;
-import java.nio.file.Files;
 import java.security.InvalidKeyException;
 import java.security.Key;
 import java.security.KeyFactory;
@@ -86,7 +75,6 @@ import java.util.regex.Pattern;
 import javax.crypto.spec.SecretKeySpec;
 import org.easymock.Capture;
 import org.easymock.EasyMock;
-import org.easymock.IAnswer;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -354,8 +342,8 @@ public class StorageImplTest {
     BucketInfo updatedBucketInfo = BUCKET_INFO1.toBuilder().setIndexPage("some-page").build();
     EasyMock.expect(
             storageRpcMock.patch(
-                Conversions.apiary().bucketInfo().encode(updatedBucketInfo), EMPTY_RPC_OPTIONS))
-        .andReturn(Conversions.apiary().bucketInfo().encode(updatedBucketInfo));
+                Conversions.json().bucketInfo().encode(updatedBucketInfo), EMPTY_RPC_OPTIONS))
+        .andReturn(Conversions.json().bucketInfo().encode(updatedBucketInfo));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Bucket bucket = storage.update(updatedBucketInfo);
@@ -367,8 +355,8 @@ public class StorageImplTest {
     BucketInfo updatedBucketInfo = BUCKET_INFO1.toBuilder().setIndexPage("some-page").build();
     EasyMock.expect(
             storageRpcMock.patch(
-                Conversions.apiary().bucketInfo().encode(updatedBucketInfo), BUCKET_TARGET_OPTIONS))
-        .andReturn(Conversions.apiary().bucketInfo().encode(updatedBucketInfo));
+                Conversions.json().bucketInfo().encode(updatedBucketInfo), BUCKET_TARGET_OPTIONS))
+        .andReturn(Conversions.json().bucketInfo().encode(updatedBucketInfo));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Bucket bucket =
@@ -382,8 +370,8 @@ public class StorageImplTest {
     BlobInfo updatedBlobInfo = BLOB_INFO1.toBuilder().setContentType("some-content-type").build();
     EasyMock.expect(
             storageRpcMock.patch(
-                Conversions.apiary().blobInfo().encode(updatedBlobInfo), EMPTY_RPC_OPTIONS))
-        .andReturn(Conversions.apiary().blobInfo().encode(updatedBlobInfo));
+                Conversions.json().blobInfo().encode(updatedBlobInfo), EMPTY_RPC_OPTIONS))
+        .andReturn(Conversions.json().blobInfo().encode(updatedBlobInfo));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Blob blob = storage.update(updatedBlobInfo);
@@ -395,9 +383,8 @@ public class StorageImplTest {
     BlobInfo updatedBlobInfo = BLOB_INFO1.toBuilder().setContentType("some-content-type").build();
     EasyMock.expect(
             storageRpcMock.patch(
-                Conversions.apiary().blobInfo().encode(updatedBlobInfo),
-                BLOB_TARGET_OPTIONS_UPDATE))
-        .andReturn(Conversions.apiary().blobInfo().encode(updatedBlobInfo));
+                Conversions.json().blobInfo().encode(updatedBlobInfo), BLOB_TARGET_OPTIONS_UPDATE))
+        .andReturn(Conversions.json().blobInfo().encode(updatedBlobInfo));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Blob blob =
@@ -409,7 +396,7 @@ public class StorageImplTest {
   public void testDeleteBucket() {
     EasyMock.expect(
             storageRpcMock.delete(
-                Conversions.apiary().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1)),
+                Conversions.json().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1)),
                 EMPTY_RPC_OPTIONS))
         .andReturn(true);
     EasyMock.replay(storageRpcMock);
@@ -421,7 +408,7 @@ public class StorageImplTest {
   public void testDeleteBucketWithOptions() {
     EasyMock.expect(
             storageRpcMock.delete(
-                Conversions.apiary().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1)),
+                Conversions.json().bucketInfo().encode(BucketInfo.of(BUCKET_NAME1)),
                 BUCKET_SOURCE_OPTIONS))
         .andReturn(true);
     EasyMock.replay(storageRpcMock);
@@ -433,7 +420,7 @@ public class StorageImplTest {
   public void testDeleteBlob() {
     EasyMock.expect(
             storageRpcMock.delete(
-                Conversions.apiary().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1)),
+                Conversions.json().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1)),
                 EMPTY_RPC_OPTIONS))
         .andReturn(true);
     EasyMock.replay(storageRpcMock);
@@ -445,7 +432,7 @@ public class StorageImplTest {
   public void testDeleteBlobWithOptions() {
     EasyMock.expect(
             storageRpcMock.delete(
-                Conversions.apiary().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1)),
+                Conversions.json().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1)),
                 BLOB_SOURCE_OPTIONS))
         .andReturn(true);
     EasyMock.replay(storageRpcMock);
@@ -459,7 +446,7 @@ public class StorageImplTest {
   public void testDeleteBlobWithOptionsFromBlobId() {
     EasyMock.expect(
             storageRpcMock.delete(
-                Conversions.apiary().blobId().encode(BLOB_INFO1.getBlobId()), BLOB_SOURCE_OPTIONS))
+                Conversions.json().blobId().encode(BLOB_INFO1.getBlobId()), BLOB_SOURCE_OPTIONS))
         .andReturn(true);
     EasyMock.replay(storageRpcMock);
     initializeService();
@@ -480,11 +467,11 @@ public class StorageImplTest {
     EasyMock.expect(
             storageRpcMock.compose(
                 ImmutableList.of(
-                    Conversions.apiary().blobInfo().encode(BLOB_INFO2),
-                    Conversions.apiary().blobInfo().encode(BLOB_INFO3)),
-                Conversions.apiary().blobInfo().encode(BLOB_INFO1),
+                    Conversions.json().blobInfo().encode(BLOB_INFO2),
+                    Conversions.json().blobInfo().encode(BLOB_INFO3)),
+                Conversions.json().blobInfo().encode(BLOB_INFO1),
                 EMPTY_RPC_OPTIONS))
-        .andReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1));
+        .andReturn(Conversions.json().blobInfo().encode(BLOB_INFO1));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Blob blob = storage.compose(req);
@@ -502,11 +489,11 @@ public class StorageImplTest {
     EasyMock.expect(
             storageRpcMock.compose(
                 ImmutableList.of(
-                    Conversions.apiary().blobInfo().encode(BLOB_INFO2),
-                    Conversions.apiary().blobInfo().encode(BLOB_INFO3)),
-                Conversions.apiary().blobInfo().encode(BLOB_INFO1),
+                    Conversions.json().blobInfo().encode(BLOB_INFO2),
+                    Conversions.json().blobInfo().encode(BLOB_INFO3)),
+                Conversions.json().blobInfo().encode(BLOB_INFO1),
                 BLOB_TARGET_OPTIONS_COMPOSE))
-        .andReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1));
+        .andReturn(Conversions.json().blobInfo().encode(BLOB_INFO1));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Blob blob = storage.compose(req);
@@ -518,10 +505,10 @@ public class StorageImplTest {
     CopyRequest request = Storage.CopyRequest.of(BLOB_INFO1.getBlobId(), BLOB_INFO2.getBlobId());
     StorageRpc.RewriteRequest rpcRequest =
         new StorageRpc.RewriteRequest(
-            Conversions.apiary().blobId().encode(request.getSource()),
+            Conversions.json().blobId().encode(request.getSource()),
             EMPTY_RPC_OPTIONS,
             false,
-            Conversions.apiary().blobInfo().encode(BLOB_INFO2),
+            Conversions.json().blobInfo().encode(BLOB_INFO2),
             EMPTY_RPC_OPTIONS,
             null);
     StorageRpc.RewriteResponse rpcResponse =
@@ -545,10 +532,10 @@ public class StorageImplTest {
             .build();
     StorageRpc.RewriteRequest rpcRequest =
         new StorageRpc.RewriteRequest(
-            Conversions.apiary().blobId().encode(request.getSource()),
+            Conversions.json().blobId().encode(request.getSource()),
             BLOB_SOURCE_OPTIONS_COPY,
             true,
-            Conversions.apiary().blobInfo().encode(request.getTarget()),
+            Conversions.json().blobInfo().encode(request.getTarget()),
             BLOB_TARGET_OPTIONS_COMPOSE,
             null);
     StorageRpc.RewriteResponse rpcResponse =
@@ -572,10 +559,10 @@ public class StorageImplTest {
             .build();
     StorageRpc.RewriteRequest rpcRequest =
         new StorageRpc.RewriteRequest(
-            Conversions.apiary().blobId().encode(request.getSource()),
+            Conversions.json().blobId().encode(request.getSource()),
             ENCRYPTION_KEY_OPTIONS,
             true,
-            Conversions.apiary().blobInfo().encode(request.getTarget()),
+            Conversions.json().blobInfo().encode(request.getTarget()),
             ENCRYPTION_KEY_OPTIONS,
             null);
     StorageRpc.RewriteResponse rpcResponse =
@@ -609,10 +596,10 @@ public class StorageImplTest {
             .build();
     StorageRpc.RewriteRequest rpcRequest =
         new StorageRpc.RewriteRequest(
-            Conversions.apiary().blobId().encode(request.getSource()),
+            Conversions.json().blobId().encode(request.getSource()),
             ENCRYPTION_KEY_OPTIONS,
             true,
-            Conversions.apiary().blobInfo().encode(request.getTarget()),
+            Conversions.json().blobInfo().encode(request.getTarget()),
             KMS_KEY_NAME_OPTIONS,
             null);
     StorageRpc.RewriteResponse rpcResponse =
@@ -646,10 +633,10 @@ public class StorageImplTest {
             .build();
     StorageRpc.RewriteRequest rpcRequest =
         new StorageRpc.RewriteRequest(
-            Conversions.apiary().blobId().encode(request.getSource()),
+            Conversions.json().blobId().encode(request.getSource()),
             BLOB_SOURCE_OPTIONS_COPY,
             true,
-            Conversions.apiary().blobInfo().encode(request.getTarget()),
+            Conversions.json().blobInfo().encode(request.getTarget()),
             BLOB_TARGET_OPTIONS_COMPOSE,
             null);
     StorageRpc.RewriteResponse rpcResponse =
@@ -668,22 +655,17 @@ public class StorageImplTest {
     CopyRequest request = Storage.CopyRequest.of(BLOB_INFO1.getBlobId(), BLOB_INFO2.getBlobId());
     StorageRpc.RewriteRequest rpcRequest =
         new StorageRpc.RewriteRequest(
-            Conversions.apiary().blobId().encode(request.getSource()),
+            Conversions.json().blobId().encode(request.getSource()),
             EMPTY_RPC_OPTIONS,
             false,
-            Conversions.apiary().blobInfo().encode(BLOB_INFO2),
+            Conversions.json().blobInfo().encode(BLOB_INFO2),
             EMPTY_RPC_OPTIONS,
             null);
     StorageRpc.RewriteResponse rpcResponse1 =
         new StorageRpc.RewriteResponse(rpcRequest, null, 42L, false, "token", 21L);
     StorageRpc.RewriteResponse rpcResponse2 =
         new StorageRpc.RewriteResponse(
-            rpcRequest,
-            Conversions.apiary().blobInfo().encode(BLOB_INFO1),
-            42L,
-            true,
-            "token",
-            42L);
+            rpcRequest, Conversions.json().blobInfo().encode(BLOB_INFO1), 42L, true, "token", 42L);
     EasyMock.expect(storageRpcMock.openRewrite(rpcRequest)).andReturn(rpcResponse1);
     EasyMock.expect(storageRpcMock.continueRewrite(rpcResponse1)).andReturn(rpcResponse2);
     EasyMock.replay(storageRpcMock);
@@ -698,86 +680,6 @@ public class StorageImplTest {
     assertEquals(42L, writer.getBlobSize());
   }
 
-  @Test
-  public void testReadAllBytes() {
-    EasyMock.expect(
-            storageRpcMock.load(
-                Conversions.apiary().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1)),
-                EMPTY_RPC_OPTIONS))
-        .andReturn(BLOB_CONTENT);
-    EasyMock.replay(storageRpcMock);
-    initializeService();
-    byte[] readBytes = storage.readAllBytes(BUCKET_NAME1, BLOB_NAME1);
-    assertArrayEquals(BLOB_CONTENT, readBytes);
-  }
-
-  @Test
-  public void testReadAllBytesWithOptions() {
-    EasyMock.expect(
-            storageRpcMock.load(
-                Conversions.apiary().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1)),
-                BLOB_SOURCE_OPTIONS))
-        .andReturn(BLOB_CONTENT);
-    EasyMock.replay(storageRpcMock);
-    initializeService();
-    byte[] readBytes =
-        storage.readAllBytes(
-            BUCKET_NAME1, BLOB_NAME1, BLOB_SOURCE_GENERATION, BLOB_SOURCE_METAGENERATION);
-    assertArrayEquals(BLOB_CONTENT, readBytes);
-  }
-
-  @Test
-  public void testReadAllBytesWithDecriptionKey() {
-    EasyMock.expect(
-            storageRpcMock.load(
-                Conversions.apiary().blobId().encode(BlobId.of(BUCKET_NAME1, BLOB_NAME1)),
-                ENCRYPTION_KEY_OPTIONS))
-        .andReturn(BLOB_CONTENT)
-        .times(2);
-    EasyMock.replay(storageRpcMock);
-    initializeService();
-    byte[] readBytes =
-        storage.readAllBytes(BUCKET_NAME1, BLOB_NAME1, BlobSourceOption.decryptionKey(KEY));
-    assertArrayEquals(BLOB_CONTENT, readBytes);
-    readBytes =
-        storage.readAllBytes(BUCKET_NAME1, BLOB_NAME1, BlobSourceOption.decryptionKey(BASE64_KEY));
-    assertArrayEquals(BLOB_CONTENT, readBytes);
-  }
-
-  @Test
-  public void testReadAllBytesFromBlobIdWithOptions() {
-    EasyMock.expect(
-            storageRpcMock.load(
-                Conversions.apiary().blobId().encode(BLOB_INFO1.getBlobId()), BLOB_SOURCE_OPTIONS))
-        .andReturn(BLOB_CONTENT);
-    EasyMock.replay(storageRpcMock);
-    initializeService();
-    byte[] readBytes =
-        storage.readAllBytes(
-            BLOB_INFO1.getBlobId(),
-            BLOB_SOURCE_GENERATION_FROM_BLOB_ID,
-            BLOB_SOURCE_METAGENERATION);
-    assertArrayEquals(BLOB_CONTENT, readBytes);
-  }
-
-  @Test
-  public void testReadAllBytesFromBlobIdWithDecriptionKey() {
-    EasyMock.expect(
-            storageRpcMock.load(
-                Conversions.apiary().blobId().encode(BLOB_INFO1.getBlobId()),
-                ENCRYPTION_KEY_OPTIONS))
-        .andReturn(BLOB_CONTENT)
-        .times(2);
-    EasyMock.replay(storageRpcMock);
-    initializeService();
-    byte[] readBytes =
-        storage.readAllBytes(BLOB_INFO1.getBlobId(), BlobSourceOption.decryptionKey(KEY));
-    assertArrayEquals(BLOB_CONTENT, readBytes);
-    readBytes =
-        storage.readAllBytes(BLOB_INFO1.getBlobId(), BlobSourceOption.decryptionKey(BASE64_KEY));
-    assertArrayEquals(BLOB_CONTENT, readBytes);
-  }
-
   @Test
   public void testBatch() {
     RpcBatch batchMock = EasyMock.mock(RpcBatch.class);
@@ -1560,11 +1462,11 @@ public class StorageImplTest {
     Capture<RpcBatch.Callback<StorageObject>> callback1 = Capture.newInstance();
     Capture<RpcBatch.Callback<StorageObject>> callback2 = Capture.newInstance();
     batchMock.addGet(
-        EasyMock.eq(Conversions.apiary().blobId().encode(blobId1)),
+        EasyMock.eq(Conversions.json().blobId().encode(blobId1)),
         EasyMock.capture(callback1),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     batchMock.addGet(
-        EasyMock.eq(Conversions.apiary().blobId().encode(blobId2)),
+        EasyMock.eq(Conversions.json().blobId().encode(blobId2)),
         EasyMock.capture(callback2),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
@@ -1572,7 +1474,7 @@ public class StorageImplTest {
     EasyMock.replay(storageRpcMock, batchMock);
     initializeService();
     List<Blob> resultBlobs = storage.get(blobId1, blobId2);
-    callback1.getValue().onSuccess(Conversions.apiary().blobInfo().encode(BLOB_INFO1));
+    callback1.getValue().onSuccess(Conversions.json().blobInfo().encode(BLOB_INFO1));
     callback2.getValue().onFailure(new GoogleJsonError());
     assertEquals(2, resultBlobs.size());
     assertEquals(new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO1)), resultBlobs.get(0));
@@ -1588,11 +1490,11 @@ public class StorageImplTest {
     Capture<RpcBatch.Callback<StorageObject>> callback1 = Capture.newInstance();
     Capture<RpcBatch.Callback<StorageObject>> callback2 = Capture.newInstance();
     batchMock.addGet(
-        EasyMock.eq(Conversions.apiary().blobId().encode(blobId1)),
+        EasyMock.eq(Conversions.json().blobId().encode(blobId1)),
         EasyMock.capture(callback1),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     batchMock.addGet(
-        EasyMock.eq(Conversions.apiary().blobId().encode(blobId2)),
+        EasyMock.eq(Conversions.json().blobId().encode(blobId2)),
         EasyMock.capture(callback2),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
@@ -1600,7 +1502,7 @@ public class StorageImplTest {
     EasyMock.replay(storageRpcMock, batchMock);
     initializeService();
     List<Blob> resultBlobs = storage.get(ImmutableList.of(blobId1, blobId2));
-    callback1.getValue().onSuccess(Conversions.apiary().blobInfo().encode(BLOB_INFO1));
+    callback1.getValue().onSuccess(Conversions.json().blobInfo().encode(BLOB_INFO1));
     callback2.getValue().onFailure(new GoogleJsonError());
     assertEquals(2, resultBlobs.size());
     assertEquals(new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO1)), resultBlobs.get(0));
@@ -1616,11 +1518,11 @@ public class StorageImplTest {
     Capture<RpcBatch.Callback<Void>> callback1 = Capture.newInstance();
     Capture<RpcBatch.Callback<Void>> callback2 = Capture.newInstance();
     batchMock.addDelete(
-        EasyMock.eq(Conversions.apiary().blobId().encode(blobId1)),
+        EasyMock.eq(Conversions.json().blobId().encode(blobId1)),
         EasyMock.capture(callback1),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     batchMock.addDelete(
-        EasyMock.eq(Conversions.apiary().blobId().encode(blobId2)),
+        EasyMock.eq(Conversions.json().blobId().encode(blobId2)),
         EasyMock.capture(callback2),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
@@ -1644,11 +1546,11 @@ public class StorageImplTest {
     Capture<RpcBatch.Callback<Void>> callback1 = Capture.newInstance();
     Capture<RpcBatch.Callback<Void>> callback2 = Capture.newInstance();
     batchMock.addDelete(
-        EasyMock.eq(Conversions.apiary().blobId().encode(blobId1)),
+        EasyMock.eq(Conversions.json().blobId().encode(blobId1)),
         EasyMock.capture(callback1),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     batchMock.addDelete(
-        EasyMock.eq(Conversions.apiary().blobId().encode(blobId2)),
+        EasyMock.eq(Conversions.json().blobId().encode(blobId2)),
         EasyMock.capture(callback2),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
@@ -1670,11 +1572,11 @@ public class StorageImplTest {
     Capture<RpcBatch.Callback<StorageObject>> callback1 = Capture.newInstance();
     Capture<RpcBatch.Callback<StorageObject>> callback2 = Capture.newInstance();
     batchMock.addPatch(
-        EasyMock.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO1)),
+        EasyMock.eq(Conversions.json().blobInfo().encode(BLOB_INFO1)),
         EasyMock.capture(callback1),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     batchMock.addPatch(
-        EasyMock.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO2)),
+        EasyMock.eq(Conversions.json().blobInfo().encode(BLOB_INFO2)),
         EasyMock.capture(callback2),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
@@ -1682,7 +1584,7 @@ public class StorageImplTest {
     EasyMock.replay(storageRpcMock, batchMock);
     initializeService();
     List<Blob> resultBlobs = storage.update(BLOB_INFO1, BLOB_INFO2);
-    callback1.getValue().onSuccess(Conversions.apiary().blobInfo().encode(BLOB_INFO1));
+    callback1.getValue().onSuccess(Conversions.json().blobInfo().encode(BLOB_INFO1));
     callback2.getValue().onFailure(new GoogleJsonError());
     assertEquals(2, resultBlobs.size());
     assertEquals(new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO1)), resultBlobs.get(0));
@@ -1696,11 +1598,11 @@ public class StorageImplTest {
     Capture<RpcBatch.Callback<StorageObject>> callback1 = Capture.newInstance();
     Capture<RpcBatch.Callback<StorageObject>> callback2 = Capture.newInstance();
     batchMock.addPatch(
-        EasyMock.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO1)),
+        EasyMock.eq(Conversions.json().blobInfo().encode(BLOB_INFO1)),
         EasyMock.capture(callback1),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     batchMock.addPatch(
-        EasyMock.eq(Conversions.apiary().blobInfo().encode(BLOB_INFO2)),
+        EasyMock.eq(Conversions.json().blobInfo().encode(BLOB_INFO2)),
         EasyMock.capture(callback2),
         EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
     EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
@@ -1708,7 +1610,7 @@ public class StorageImplTest {
     EasyMock.replay(storageRpcMock, batchMock);
     initializeService();
     List<Blob> resultBlobs = storage.update(ImmutableList.of(BLOB_INFO1, BLOB_INFO2));
-    callback1.getValue().onSuccess(Conversions.apiary().blobInfo().encode(BLOB_INFO1));
+    callback1.getValue().onSuccess(Conversions.json().blobInfo().encode(BLOB_INFO1));
     callback2.getValue().onFailure(new GoogleJsonError());
     assertEquals(2, resultBlobs.size());
     assertEquals(new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO1)), resultBlobs.get(0));
@@ -1721,7 +1623,7 @@ public class StorageImplTest {
     EasyMock.expect(
             storageRpcMock.getAcl(
                 BUCKET_NAME1, "allAuthenticatedUsers", new HashMap<StorageRpc.Option, Object>()))
-        .andReturn(Conversions.apiary().bucketAcl().encode(ACL));
+        .andReturn(Conversions.json().bucketAcl().encode(ACL));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Acl acl = storage.getAcl(BUCKET_NAME1, User.ofAllAuthenticatedUsers());
@@ -1755,9 +1657,9 @@ public class StorageImplTest {
     Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
     EasyMock.expect(
             storageRpcMock.createAcl(
-                Conversions.apiary().bucketAcl().encode(ACL).setBucket(BUCKET_NAME1),
+                Conversions.json().bucketAcl().encode(ACL).setBucket(BUCKET_NAME1),
                 new HashMap<StorageRpc.Option, Object>()))
-        .andReturn(Conversions.apiary().bucketAcl().encode(returnedAcl));
+        .andReturn(Conversions.json().bucketAcl().encode(returnedAcl));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Acl acl = storage.createAcl(BUCKET_NAME1, ACL);
@@ -1769,9 +1671,9 @@ public class StorageImplTest {
     Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
     EasyMock.expect(
             storageRpcMock.patchAcl(
-                Conversions.apiary().bucketAcl().encode(ACL).setBucket(BUCKET_NAME1),
+                Conversions.json().bucketAcl().encode(ACL).setBucket(BUCKET_NAME1),
                 new HashMap<StorageRpc.Option, Object>()))
-        .andReturn(Conversions.apiary().bucketAcl().encode(returnedAcl));
+        .andReturn(Conversions.json().bucketAcl().encode(returnedAcl));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Acl acl = storage.updateAcl(BUCKET_NAME1, ACL);
@@ -1783,8 +1685,8 @@ public class StorageImplTest {
     EasyMock.expect(storageRpcMock.listAcls(BUCKET_NAME1, new HashMap<StorageRpc.Option, Object>()))
         .andReturn(
             ImmutableList.of(
-                Conversions.apiary().bucketAcl().encode(ACL),
-                Conversions.apiary().bucketAcl().encode(OTHER_ACL)));
+                Conversions.json().bucketAcl().encode(ACL),
+                Conversions.json().bucketAcl().encode(OTHER_ACL)));
     EasyMock.replay(storageRpcMock);
     initializeService();
     List<Acl> acls = storage.listAcls(BUCKET_NAME1);
@@ -1794,7 +1696,7 @@ public class StorageImplTest {
   @Test
   public void testGetDefaultBucketAcl() {
     EasyMock.expect(storageRpcMock.getDefaultAcl(BUCKET_NAME1, "allAuthenticatedUsers"))
-        .andReturn(Conversions.apiary().objectAcl().encode(ACL));
+        .andReturn(Conversions.json().objectAcl().encode(ACL));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Acl acl = storage.getDefaultAcl(BUCKET_NAME1, User.ofAllAuthenticatedUsers());
@@ -1824,8 +1726,8 @@ public class StorageImplTest {
     Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
     EasyMock.expect(
             storageRpcMock.createDefaultAcl(
-                Conversions.apiary().objectAcl().encode(ACL).setBucket(BUCKET_NAME1)))
-        .andReturn(Conversions.apiary().objectAcl().encode(returnedAcl));
+                Conversions.json().objectAcl().encode(ACL).setBucket(BUCKET_NAME1)))
+        .andReturn(Conversions.json().objectAcl().encode(returnedAcl));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Acl acl = storage.createDefaultAcl(BUCKET_NAME1, ACL);
@@ -1837,8 +1739,8 @@ public class StorageImplTest {
     Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
     EasyMock.expect(
             storageRpcMock.patchDefaultAcl(
-                Conversions.apiary().objectAcl().encode(ACL).setBucket(BUCKET_NAME1)))
-        .andReturn(Conversions.apiary().objectAcl().encode(returnedAcl));
+                Conversions.json().objectAcl().encode(ACL).setBucket(BUCKET_NAME1)))
+        .andReturn(Conversions.json().objectAcl().encode(returnedAcl));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Acl acl = storage.updateDefaultAcl(BUCKET_NAME1, ACL);
@@ -1850,8 +1752,8 @@ public class StorageImplTest {
     EasyMock.expect(storageRpcMock.listDefaultAcls(BUCKET_NAME1))
         .andReturn(
             ImmutableList.of(
-                Conversions.apiary().objectAcl().encode(ACL),
-                Conversions.apiary().objectAcl().encode(OTHER_ACL)));
+                Conversions.json().objectAcl().encode(ACL),
+                Conversions.json().objectAcl().encode(OTHER_ACL)));
     EasyMock.replay(storageRpcMock);
     initializeService();
     List<Acl> acls = storage.listDefaultAcls(BUCKET_NAME1);
@@ -1862,7 +1764,7 @@ public class StorageImplTest {
   public void testGetBlobAcl() {
     BlobId blobId = BlobId.of(BUCKET_NAME1, BLOB_NAME1, 42L);
     EasyMock.expect(storageRpcMock.getAcl(BUCKET_NAME1, BLOB_NAME1, 42L, "allAuthenticatedUsers"))
-        .andReturn(Conversions.apiary().objectAcl().encode(ACL));
+        .andReturn(Conversions.json().objectAcl().encode(ACL));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Acl acl = storage.getAcl(blobId, User.ofAllAuthenticatedUsers());
@@ -1896,13 +1798,13 @@ public class StorageImplTest {
     Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
     EasyMock.expect(
             storageRpcMock.createAcl(
-                Conversions.apiary()
+                Conversions.json()
                     .objectAcl()
                     .encode(ACL)
                     .setBucket(BUCKET_NAME1)
                     .setObject(BLOB_NAME1)
                     .setGeneration(42L)))
-        .andReturn(Conversions.apiary().objectAcl().encode(returnedAcl));
+        .andReturn(Conversions.json().objectAcl().encode(returnedAcl));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Acl acl = storage.createAcl(blobId, ACL);
@@ -1915,13 +1817,13 @@ public class StorageImplTest {
     Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
     EasyMock.expect(
             storageRpcMock.patchAcl(
-                Conversions.apiary()
+                Conversions.json()
                     .objectAcl()
                     .encode(ACL)
                     .setBucket(BUCKET_NAME1)
                     .setObject(BLOB_NAME1)
                     .setGeneration(42L)))
-        .andReturn(Conversions.apiary().objectAcl().encode(returnedAcl));
+        .andReturn(Conversions.json().objectAcl().encode(returnedAcl));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Acl acl = storage.updateAcl(blobId, ACL);
@@ -1934,8 +1836,8 @@ public class StorageImplTest {
     EasyMock.expect(storageRpcMock.listAcls(BUCKET_NAME1, BLOB_NAME1, 42L))
         .andReturn(
             ImmutableList.of(
-                Conversions.apiary().objectAcl().encode(ACL),
-                Conversions.apiary().objectAcl().encode(OTHER_ACL)));
+                Conversions.json().objectAcl().encode(ACL),
+                Conversions.json().objectAcl().encode(OTHER_ACL)));
     EasyMock.replay(storageRpcMock);
     initializeService();
     List<Acl> acls = storage.listAcls(blobId);
@@ -2056,9 +1958,9 @@ public class StorageImplTest {
   public void testLockRetentionPolicy() {
     EasyMock.expect(
             storageRpcMock.lockRetentionPolicy(
-                Conversions.apiary().bucketInfo().encode(BUCKET_INFO3),
+                Conversions.json().bucketInfo().encode(BUCKET_INFO3),
                 BUCKET_TARGET_OPTIONS_LOCK_RETENTION_POLICY))
-        .andReturn(Conversions.apiary().bucketInfo().encode(BUCKET_INFO3));
+        .andReturn(Conversions.json().bucketInfo().encode(BUCKET_INFO3));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Bucket bucket =
@@ -2070,7 +1972,7 @@ public class StorageImplTest {
   @Test
   public void testGetServiceAccount() {
     EasyMock.expect(storageRpcMock.getServiceAccount("projectId"))
-        .andReturn(Conversions.apiary().serviceAccount().encode(SERVICE_ACCOUNT));
+        .andReturn(Conversions.json().serviceAccount().encode(SERVICE_ACCOUNT));
     EasyMock.replay(storageRpcMock);
     initializeService();
     ServiceAccount serviceAccount = storage.getServiceAccount("projectId");
@@ -2080,10 +1982,9 @@ public class StorageImplTest {
   @Test
   public void testRetryableException() {
     BlobId blob = BlobId.of(BUCKET_NAME1, BLOB_NAME1);
-    EasyMock.expect(
-            storageRpcMock.get(Conversions.apiary().blobId().encode(blob), EMPTY_RPC_OPTIONS))
+    EasyMock.expect(storageRpcMock.get(Conversions.json().blobId().encode(blob), EMPTY_RPC_OPTIONS))
         .andThrow(new StorageException(500, "internalError"))
-        .andReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1));
+        .andReturn(Conversions.json().blobInfo().encode(BLOB_INFO1));
     EasyMock.replay(storageRpcMock);
     storage =
         options
@@ -2100,8 +2001,7 @@ public class StorageImplTest {
   public void testNonRetryableException() {
     BlobId blob = BlobId.of(BUCKET_NAME1, BLOB_NAME1);
     String exceptionMessage = "Not Implemented";
-    EasyMock.expect(
-            storageRpcMock.get(Conversions.apiary().blobId().encode(blob), EMPTY_RPC_OPTIONS))
+    EasyMock.expect(storageRpcMock.get(Conversions.json().blobId().encode(blob), EMPTY_RPC_OPTIONS))
         .andThrow(new StorageException(501, exceptionMessage));
     EasyMock.replay(storageRpcMock);
     storage =
@@ -2123,8 +2023,7 @@ public class StorageImplTest {
   public void testRuntimeException() {
     BlobId blob = BlobId.of(BUCKET_NAME1, BLOB_NAME1);
     String exceptionMessage = "Artificial runtime exception";
-    EasyMock.expect(
-            storageRpcMock.get(Conversions.apiary().blobId().encode(blob), EMPTY_RPC_OPTIONS))
+    EasyMock.expect(storageRpcMock.get(Conversions.json().blobId().encode(blob), EMPTY_RPC_OPTIONS))
         .andThrow(new RuntimeException(exceptionMessage));
     EasyMock.replay(storageRpcMock);
     storage =
@@ -2241,9 +2140,9 @@ public class StorageImplTest {
             .build();
     EasyMock.expect(
             storageRpcMock.create(
-                Conversions.apiary().bucketInfo().encode(bucketInfo),
+                Conversions.json().bucketInfo().encode(bucketInfo),
                 new HashMap<StorageRpc.Option, Object>()))
-        .andReturn(Conversions.apiary().bucketInfo().encode(bucketInfo));
+        .andReturn(Conversions.json().bucketInfo().encode(bucketInfo));
     EasyMock.replay(storageRpcMock);
     initializeService();
     Bucket bucket = storage.create(bucketInfo);
@@ -2258,97 +2157,4 @@ public class StorageImplTest {
     assertEquals(30, lifecycleRule.getCondition().getDaysSinceCustomTime().intValue());
     assertNotNull(lifecycleRule.getCondition().getCustomTimeBefore());
   }
-
-  @Test
-  public void testDownloadTo() throws Exception {
-    BlobId blob = BlobId.of(BUCKET_NAME1, BLOB_NAME1);
-    storage = options.toBuilder().build().getService();
-    final byte[] expected = {1, 2};
-    EasyMock.expect(
-            storageRpcMock.read(
-                anyObject(StorageObject.class),
-                anyObject(Map.class),
-                eq(0l),
-                anyObject(OutputStream.class)))
-        .andAnswer(
-            new IAnswer<Long>() {
-              @Override
-              public Long answer() throws Throwable {
-                ((OutputStream) getCurrentArguments()[3]).write(expected);
-                return 2l;
-              }
-            });
-    EasyMock.replay(storageRpcMock);
-    File file = File.createTempFile("blob", ".tmp");
-    storage.downloadTo(blob, file.toPath());
-    byte actual[] = Files.readAllBytes(file.toPath());
-    assertArrayEquals(expected, actual);
-  }
-
-  @Test
-  public void testDownloadToWithRetries() throws Exception {
-    BlobId blob = BlobId.of(BUCKET_NAME1, BLOB_NAME1);
-    storage =
-        options
-            .toBuilder()
-            .setRetrySettings(RetrySettings.newBuilder().setMaxAttempts(2).build())
-            .build()
-            .getService();
-    final byte[] expected = {1, 2};
-    expect(
-            storageRpcMock.read(
-                anyObject(StorageObject.class),
-                anyObject(Map.class),
-                eq(0l),
-                anyObject(OutputStream.class)))
-        .andAnswer(
-            new IAnswer<Long>() {
-              @Override
-              public Long answer() throws Throwable {
-                ((OutputStream) getCurrentArguments()[3]).write(expected[0]);
-                throw new StorageException(504, "error");
-              }
-            });
-    expect(
-            storageRpcMock.read(
-                anyObject(StorageObject.class),
-                anyObject(Map.class),
-                eq(1l),
-                anyObject(OutputStream.class)))
-        .andAnswer(
-            new IAnswer<Long>() {
-              @Override
-              public Long answer() throws Throwable {
-                ((OutputStream) getCurrentArguments()[3]).write(expected[1]);
-                return 1l;
-              }
-            });
-    replay(storageRpcMock);
-    File file = File.createTempFile("blob", ".tmp");
-    storage.downloadTo(blob, file.toPath());
-    byte actual[] = Files.readAllBytes(file.toPath());
-    assertArrayEquals(expected, actual);
-  }
-
-  @Test
-  public void testDownloadToWithException() throws Exception {
-    BlobId blob = BlobId.of(BUCKET_NAME1, BLOB_NAME1);
-    storage = options.toBuilder().build().getService();
-    Exception exception = new IllegalStateException("test");
-    expect(
-            storageRpcMock.read(
-                anyObject(StorageObject.class),
-                anyObject(Map.class),
-                eq(0l),
-                anyObject(OutputStream.class)))
-        .andThrow(exception);
-    replay(storageRpcMock);
-    File file = File.createTempFile("blob", ".tmp");
-    try {
-      storage.downloadTo(blob, file.toPath());
-      fail();
-    } catch (StorageException e) {
-      assertSame(exception, e.getCause());
-    }
-  }
 }
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/conformance/retry/ITRetryConformanceTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/ITRetryConformanceTest.java
index 89574734..6c93e268 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/ITRetryConformanceTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/ITRetryConformanceTest.java
@@ -21,6 +21,7 @@ import static com.google.cloud.storage.PackagePrivateMethodWorkarounds.bucketCop
 import static com.google.cloud.storage.conformance.retry.Ctx.ctx;
 import static com.google.cloud.storage.conformance.retry.State.empty;
 import static com.google.common.truth.Truth.assertThat;
+import static com.google.common.truth.Truth.assertWithMessage;
 import static java.util.Objects.requireNonNull;
 import static org.junit.Assert.assertNotNull;
 
@@ -80,7 +81,6 @@ import org.junit.runner.RunWith;
  * RpcMethodMappings}.
  */
 @RunWith(StorageITRunner.class)
-// @CrossRun(transports = Transport.HTTP, backends = Backend.TEST_BENCH)
 @SingleBackend(Backend.TEST_BENCH)
 @Parameterized(RetryConformanceParameterProvider.class)
 @ParallelFriendly
@@ -117,9 +117,10 @@ public class ITRetryConformanceTest {
   public void tearDown() throws Throwable {
     LOGGER.fine("Running teardown...");
     if (ctx != null) {
+      Ctx tearDownCtx = ctx.leftMap(s -> nonTestStorage);
       getReplaceStorageInObjectsFromCtx()
           .andThen(mapping.getTearDown())
-          .apply(ctx, testRetryConformance);
+          .apply(tearDownCtx, testRetryConformance);
     }
     retryTestFixture.finished(null);
     LOGGER.fine("Running teardown complete");
@@ -177,7 +178,9 @@ public class ITRetryConformanceTest {
       } catch (IOException e) {
         throw new RuntimeException(e);
       }
-      assertThat(retryTestCases).isNotEmpty();
+      assertWithMessage("Filter too strict. Resolved 0 retry test cases")
+          .that(retryTestCases)
+          .isNotEmpty();
       return retryTestCases.stream()
           .map(
               rtc ->
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMapping.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMapping.java
index 1c3947ad..4b2c8ccf 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMapping.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMapping.java
@@ -21,6 +21,7 @@ import static java.util.Objects.requireNonNull;
 import static org.junit.Assert.fail;
 
 import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSetup;
 import com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceTeardown;
 import com.google.cloud.storage.conformance.retry.Functions.CtxFunction;
@@ -104,8 +105,15 @@ final class RpcMethodMapping {
           if (instructions.contains("return-401") && code == 401) {
             matchExpectedCode = true;
           }
-          if (instructions.contains("return-reset-connection") && code == 0) {
-            matchExpectedCode = true;
+          if (instructions.contains("return-reset-connection")) {
+            if (c.getTransport() == Transport.HTTP && code == 0) {
+              matchExpectedCode = true;
+            }
+            // in grpc a broken connection is converted to an UNAVAILABLE
+            // our mapping from UNAVAILABLE to status code is 503
+            if (c.getTransport() == Transport.GRPC && code == 503) {
+              matchExpectedCode = true;
+            }
           }
 
           if (matchExpectedCode) {
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMappings.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMappings.java
index dbd0e925..a76173a1 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMappings.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMappings.java
@@ -16,6 +16,7 @@
 
 package com.google.cloud.storage.conformance.retry;
 
+import static com.google.cloud.storage.TestUtils.xxd;
 import static com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSetup.defaultSetup;
 import static com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSetup.notificationSetup;
 import static com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSetup.pubsubTopicSetup;
@@ -52,6 +53,7 @@ import com.google.cloud.storage.Storage.CopyRequest;
 import com.google.cloud.storage.Storage.SignUrlOption;
 import com.google.cloud.storage.Storage.UriScheme;
 import com.google.cloud.storage.StorageRoles;
+import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.conformance.retry.CtxFunctions.Local;
 import com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSetup;
 import com.google.cloud.storage.conformance.retry.CtxFunctions.Rpc;
@@ -1277,9 +1279,12 @@ final class RpcMethodMappings {
                               try {
                                 ReadChannel reader =
                                     ctx.getStorage().reader(ctx.getState().getBlob().getBlobId());
-                                WritableByteChannel write =
-                                    Channels.newChannel(ByteStreams.nullOutputStream());
+                                ByteArrayOutputStream baos = new ByteArrayOutputStream();
+                                WritableByteChannel write = Channels.newChannel(baos);
                                 ByteStreams.copy(reader, write);
+
+                                assertThat(xxd(baos.toByteArray()))
+                                    .isEqualTo(xxd(c.getHelloWorldUtf8Bytes()));
                               } catch (IOException e) {
                                 if (e.getCause() instanceof BaseServiceException) {
                                   throw e.getCause();
@@ -1299,9 +1304,12 @@ final class RpcMethodMappings {
                                         .reader(
                                             ctx.getState().getBlob().getBlobId().getBucket(),
                                             ctx.getState().getBlob().getBlobId().getName());
-                                WritableByteChannel write =
-                                    Channels.newChannel(ByteStreams.nullOutputStream());
+                                ByteArrayOutputStream baos = new ByteArrayOutputStream();
+                                WritableByteChannel write = Channels.newChannel(baos);
                                 ByteStreams.copy(reader, write);
+
+                                assertThat(xxd(baos.toByteArray()))
+                                    .isEqualTo(xxd(c.getHelloWorldUtf8Bytes()));
                               } catch (IOException e) {
                                 if (e.getCause() instanceof BaseServiceException) {
                                   throw e.getCause();
@@ -1600,7 +1608,9 @@ final class RpcMethodMappings {
                 .build());
         a.add(
             RpcMethodMapping.newBuilder(54, objects.insert)
-                .withApplicable(not(TestRetryConformance::isPreconditionsProvided))
+                .withApplicable(
+                    not(TestRetryConformance::isPreconditionsProvided)
+                        .and(TestRetryConformance.transportIs(Transport.HTTP)))
                 .withSetup(defaultSetup.andThen(Local.blobInfoWithoutGeneration))
                 .withTest(
                     (ctx, c) ->
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/TestRetryConformance.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/TestRetryConformance.java
index 65c07348..7644052f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/TestRetryConformance.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/TestRetryConformance.java
@@ -42,6 +42,7 @@ import java.time.Instant;
 import java.time.ZoneId;
 import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
@@ -239,6 +240,10 @@ final class TestRetryConformance {
     return getTestName();
   }
 
+  public static Predicate<TestRetryConformance> transportIs(Transport t) {
+    return trc -> trc.transport == t;
+  }
+
   private static Supplier<Path> resolvePathForResource(String objectName, Method method) {
     return () -> {
       try {
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/ITBlobReadChannelV2RetryTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelV2RetryTest.java
index 4186b73a..07ff4b0f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelV2RetryTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelV2RetryTest.java
@@ -160,8 +160,6 @@ public final class ITBlobReadChannelV2RetryTest {
             .setContentEncoding("gzip")
             .build();
     Blob gen1 = storage.create(info, gzipped.getBytes(), BlobTargetOption.doesNotExist());
-    String uri = gen1.getBlobId().toGsUtilUri();
-    System.out.println("uri = " + uri);
 
     JsonObject instructions = new JsonObject();
     JsonArray value = new JsonArray();
@@ -200,4 +198,57 @@ public final class ITBlobReadChannelV2RetryTest {
                   .isEqualTo(ImmutableList.of(String.format("bytes=%d-", 256 * 1024))));
     }
   }
+
+  @Test
+  public void resumeFromCorrectOffsetWhenPartialReadSuccess() throws Exception {
+    StorageOptions baseOptions = storage.getOptions();
+    Random rand = new Random(918273645);
+
+    ChecksummedTestContent uncompressed;
+    {
+      // must use random strategy, base64 characters compress too well. 512KiB uncompressed becomes
+      // ~1600 bytes which is smaller than our 'return-broken-stream-after-256K' rule
+      byte[] bytes = DataGenerator.rand(rand).genBytes(_512KiB);
+      // byte[] bytes = DataGenerator.base64Characters().genBytes(_512KiB);
+      uncompressed = ChecksummedTestContent.of(bytes);
+    }
+    BlobId id = BlobId.of(bucket.getName(), generator.randomObjectName());
+    BlobInfo info = BlobInfo.newBuilder(id).build();
+    Blob gen1 = storage.create(info, uncompressed.getBytes(), BlobTargetOption.doesNotExist());
+
+    JsonObject instructions = new JsonObject();
+    JsonArray value = new JsonArray();
+    value.add("return-broken-stream-after-256K");
+    instructions.add("storage.objects.get", value);
+    RetryTestResource retryTestResource = new RetryTestResource(instructions);
+    RetryTestResource retryTest = testBench.createRetryTest(retryTestResource);
+
+    ImmutableMap<String, String> headers = ImmutableMap.of("x-retry-test-id", retryTest.id);
+
+    RequestAuditing requestAuditing = new RequestAuditing();
+    StorageOptions testStorageOptions =
+        baseOptions
+            .toBuilder()
+            .setTransportOptions(requestAuditing)
+            .setHeaderProvider(FixedHeaderProvider.create(headers))
+            .build();
+
+    String expected = xxd(uncompressed.getBytes());
+
+    try (Storage testStorage = testStorageOptions.getService();
+        ReadChannel r = testStorage.reader(gen1.getBlobId());
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        WritableByteChannel w = Channels.newChannel(baos)) {
+      long copy = ByteStreams.copy(r, w);
+      String actual = xxd(baos.toByteArray());
+      ImmutableList<HttpRequest> requests = requestAuditing.getRequests();
+      assertAll(
+          () -> assertThat(copy).isEqualTo(uncompressed.getBytes().length),
+          () -> assertThat(actual).isEqualTo(expected),
+          () -> assertThat(requests.get(0).getHeaders().get("range")).isNull(),
+          () ->
+              assertThat(requests.get(1).getHeaders().get("range"))
+                  .isEqualTo(ImmutableList.of(String.format("bytes=%d-", 256 * 1024))));
+    }
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java
index 1113d023..bf8132b3 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java
@@ -17,6 +17,7 @@
 package com.google.cloud.storage.it;
 
 import static com.google.common.truth.Truth.assertThat;
+import static com.google.common.truth.Truth.assertWithMessage;
 import static org.junit.Assert.assertThrows;
 
 import com.google.cloud.storage.BlobInfo;
@@ -28,12 +29,12 @@ import com.google.cloud.storage.GrpcStorageOptions;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
 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.annotations.SingleBackend;
-import com.google.cloud.storage.it.runner.annotations.StorageFixture;
 import com.google.cloud.storage.it.runner.registry.Generator;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -41,16 +42,21 @@ import java.nio.channels.WritableByteChannel;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
 import org.junit.runner.RunWith;
 
 @RunWith(StorageITRunner.class)
-@SingleBackend(Backend.PROD)
+@CrossRun(
+    transports = {Transport.GRPC},
+    backends = {Backend.PROD})
 public final class ITBlobWriteSessionTest {
 
-  @Inject
-  @StorageFixture(Transport.GRPC)
-  public Storage storage;
+  @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+  @Inject public Storage storage;
+  @Inject public Transport transport;
 
   @Inject public BucketInfo bucket;
 
@@ -63,11 +69,16 @@ public final class ITBlobWriteSessionTest {
 
   @Test
   public void bufferToTempDirThenUpload() throws Exception {
-    GrpcStorageOptions options =
-        ((GrpcStorageOptions) storage.getOptions())
-            .toBuilder()
-            .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bufferToTempDirThenUpload())
-            .build();
+    StorageOptions options = null;
+    if (transport == Transport.GRPC) {
+      options =
+          ((GrpcStorageOptions) storage.getOptions())
+              .toBuilder()
+              .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bufferToTempDirThenUpload())
+              .build();
+    }
+    assertWithMessage("unable to resolve options").that(options).isNotNull();
+    //noinspection DataFlowIssue
     try (Storage s = options.getService()) {
       doTest(s);
     }
@@ -75,12 +86,17 @@ public final class ITBlobWriteSessionTest {
 
   @Test
   public void overrideDefaultBufferSize() throws Exception {
-    GrpcStorageOptions options =
-        ((GrpcStorageOptions) storage.getOptions())
-            .toBuilder()
-            .setBlobWriteSessionConfig(
-                BlobWriteSessionConfigs.getDefault().withChunkSize(256 * 1024))
-            .build();
+    StorageOptions options = null;
+    if (transport == Transport.GRPC) {
+      options =
+          ((GrpcStorageOptions) storage.getOptions())
+              .toBuilder()
+              .setBlobWriteSessionConfig(
+                  BlobWriteSessionConfigs.getDefault().withChunkSize(256 * 1024))
+              .build();
+    }
+    assertWithMessage("unable to resolve options").that(options).isNotNull();
+    //noinspection DataFlowIssue
     try (Storage s = options.getService()) {
       doTest(s);
     }
@@ -95,7 +111,6 @@ public final class ITBlobWriteSessionTest {
     WritableByteChannel open = session.open();
     open.close();
     BlobInfo gen1 = session.getResult().get(1, TimeUnit.SECONDS);
-    System.out.println("gen1 = " + gen1);
 
     assertThat(gen1.getSize()).isEqualTo(0);
   }
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);
-  }
-}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITParallelCompositeUploadBlobWriteSessionConfigTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITParallelCompositeUploadBlobWriteSessionConfigTest.java
index 6d3157c0..012a700f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITParallelCompositeUploadBlobWriteSessionConfigTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITParallelCompositeUploadBlobWriteSessionConfigTest.java
@@ -18,6 +18,7 @@ package com.google.cloud.storage.it;
 
 import static com.google.cloud.storage.TestUtils.xxd;
 import static com.google.common.truth.Truth.assertThat;
+import static com.google.common.truth.Truth.assertWithMessage;
 import static org.junit.Assert.assertThrows;
 
 import com.google.api.gax.rpc.ApiExceptions;
@@ -29,6 +30,7 @@ import com.google.cloud.storage.BlobWriteSessionConfigs;
 import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.DataGenerator;
 import com.google.cloud.storage.GrpcStorageOptions;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.BufferAllocationStrategy;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.ExecutorSupplier;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartCleanupStrategy;
@@ -38,14 +40,14 @@ import com.google.cloud.storage.Storage.BlobSourceOption;
 import com.google.cloud.storage.Storage.BlobTargetOption;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
 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.BucketFixture;
 import com.google.cloud.storage.it.runner.annotations.BucketType;
+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.annotations.SingleBackend;
-import com.google.cloud.storage.it.runner.annotations.StorageFixture;
 import com.google.cloud.storage.it.runner.registry.Generator;
 import com.google.cloud.storage.it.runner.registry.KmsFixture;
 import com.google.common.collect.ImmutableList;
@@ -67,7 +69,9 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 
 @RunWith(StorageITRunner.class)
-@SingleBackend(Backend.PROD)
+@CrossRun(
+    transports = {Transport.GRPC},
+    backends = {Backend.PROD})
 public final class ITParallelCompositeUploadBlobWriteSessionConfigTest {
 
   private static final int _1MiB = 1024 * 1024;
@@ -79,9 +83,9 @@ public final class ITParallelCompositeUploadBlobWriteSessionConfigTest {
   @BucketFixture(BucketType.REQUESTER_PAYS)
   public BucketInfo rpBucket;
 
-  @Inject
-  @StorageFixture(Transport.GRPC)
-  public Storage injectedStorage;
+  @Inject public Storage injectedStorage;
+
+  @Inject public Transport transport;
 
   @Inject public Generator generator;
   @Inject public KmsFixture kmsFixture;
@@ -99,18 +103,25 @@ public final class ITParallelCompositeUploadBlobWriteSessionConfigTest {
 
   @Before
   public void setUp() throws Exception {
-    GrpcStorageOptions storageOptions =
-        ((GrpcStorageOptions) injectedStorage.getOptions())
-            .toBuilder()
-            .setBlobWriteSessionConfig(
-                BlobWriteSessionConfigs.parallelCompositeUpload()
-                    .withExecutorSupplier(ExecutorSupplier.useExecutor(exec))
-                    // define a max part size that is fairly small to aid in test speed
-                    .withBufferAllocationStrategy(BufferAllocationStrategy.simple(_1MiB))
-                    .withPartNamingStrategy(PartNamingStrategy.prefix("prefix-a"))
-                    // let our fixtures take care of cleaning things up
-                    .withPartCleanupStrategy(PartCleanupStrategy.never()))
-            .build();
+    ParallelCompositeUploadBlobWriteSessionConfig pcu =
+        BlobWriteSessionConfigs.parallelCompositeUpload()
+            .withExecutorSupplier(ExecutorSupplier.useExecutor(exec))
+            // define a max part size that is fairly small to aid in test speed
+            .withBufferAllocationStrategy(BufferAllocationStrategy.simple(_1MiB))
+            .withPartNamingStrategy(PartNamingStrategy.prefix("prefix-a"))
+            // let our fixtures take care of cleaning things up if an upload fails
+            .withPartCleanupStrategy(PartCleanupStrategy.onlyOnSuccess());
+
+    StorageOptions storageOptions = null;
+    if (transport == Transport.GRPC) {
+      storageOptions =
+          ((GrpcStorageOptions) injectedStorage.getOptions())
+              .toBuilder()
+              .setBlobWriteSessionConfig(pcu)
+              .build();
+    }
+    assertWithMessage("unable to resolve options").that(storageOptions).isNotNull();
+    //noinspection DataFlowIssue
     storage = storageOptions.getService();
     rand = new Random();
   }
@@ -144,7 +155,6 @@ public final class ITParallelCompositeUploadBlobWriteSessionConfigTest {
       // it is okay if the exception is raised during write itself or close, if it happens during
       // close we should get an AsynchronousCloseException
     } catch (AsynchronousCloseException ace) {
-      assertThat(ace).hasCauseThat().hasMessageThat().contains("NOT_FOUND");
       assertThat(((StorageException) ace.getCause()).getCode()).isEqualTo(404);
     } catch (StorageException se) {
       assertThat(se.getCode()).isEqualTo(404);
diff --git a/google-cloud-storage/src/test/resources/com/google/cloud/storage/it/runner/registry/Dockerfile b/google-cloud-storage/src/test/resources/com/google/cloud/storage/it/runner/registry/Dockerfile
index 4c9f43c8..1638c439 100644
--- a/google-cloud-storage/src/test/resources/com/google/cloud/storage/it/runner/registry/Dockerfile
+++ b/google-cloud-storage/src/test/resources/com/google/cloud/storage/it/runner/registry/Dockerfile
@@ -1,1 +1,1 @@
-FROM gcr.io/cloud-devrel-public-resources/storage-testbench:v0.39.0
+FROM gcr.io/cloud-devrel-public-resources/storage-testbench:v0.40.0