Repository: googleapis/java-storage
Pull request #1594 (2022-08-29 22:14:36)

test: ITStorage converted to ITObjectTest
Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a [bug/issue](https://github.com/googleapis/java-storage/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> ☕️

If you write sample code, please follow the [samples format](
https://github.com/GoogleCloudPlatform/java-docs-samples/blob/main/SAMPLE_FORMAT.md).



Patch:
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 8d862ec5..a2771b13 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
@@ -126,12 +126,13 @@ final class BackwardCompatibilityUtils {
       case UNAVAILABLE:
         return 503;
         // 504 Gateway Timeout
+      case DEADLINE_EXCEEDED:
+        return 504;
         // TODO
 
       case ABORTED: // ?
       case CANCELLED: // ?
       case UNKNOWN: // ?
-      case DEADLINE_EXCEEDED: // ?
       case DATA_LOSS: // ?
       default:
         return 0;
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
index 240113b8..442335b6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
@@ -57,7 +57,10 @@ final class GapicUnbufferedWritableByteChannel<
 
     this.writeCtx = new WriteCtx<>(requestFactory);
     this.flusher =
-        flusherFactory.newFlusher(writeCtx.getConfirmedBytes()::addAndGet, resultFuture::set);
+        flusherFactory.newFlusher(
+            requestFactory.bucketName(),
+            writeCtx.getConfirmedBytes()::addAndGet,
+            resultFuture::set);
   }
 
   @Override
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
index df0c4124..afba3bf9 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
@@ -825,7 +825,10 @@ final class GrpcConversions {
     if (from.hasCustomerEncryption()) {
       toBuilder.setCustomerEncryption(customerEncryptionCodec.decode(from.getCustomerEncryption()));
     }
-    ifNonNull(from.getStorageClass(), StorageClass::valueOf, toBuilder::setStorageClass);
+    String storageClass = from.getStorageClass();
+    if (!storageClass.isEmpty()) {
+      toBuilder.setStorageClass(StorageClass.valueOf(storageClass));
+    }
     if (from.hasUpdateStorageClassTime()) {
       toBuilder.setTimeStorageClassUpdatedOffsetDateTime(
           timestampCodec.decode(from.getUpdateStorageClassTime()));
@@ -836,7 +839,7 @@ final class GrpcConversions {
     if (from.hasEventBasedHold()) {
       toBuilder.setEventBasedHold(from.getEventBasedHold());
     }
-    ifNonNull(from.getTemporaryHold(), toBuilder::setTemporaryHold);
+    toBuilder.setTemporaryHold(from.getTemporaryHold());
     if (from.hasRetentionExpireTime()) {
       toBuilder.setRetentionExpirationTimeOffsetDateTime(
           timestampCodec.decode(from.getRetentionExpireTime()));
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 dcdf2952..8aa232a5 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
@@ -26,6 +26,7 @@ import static com.google.common.base.MoreObjects.firstNonNull;
 import static java.util.Objects.requireNonNull;
 
 import com.google.api.core.ApiFuture;
+import com.google.api.core.BetaApi;
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.grpc.GrpcStatusCode;
 import com.google.api.gax.paging.AbstractPage;
@@ -36,6 +37,7 @@ import com.google.api.gax.rpc.ApiExceptionFactory;
 import com.google.api.gax.rpc.ApiExceptions;
 import com.google.api.gax.rpc.StatusCode;
 import com.google.api.gax.rpc.UnaryCallable;
+import com.google.api.gax.rpc.UnimplementedException;
 import com.google.cloud.BaseService;
 import com.google.cloud.Policy;
 import com.google.cloud.WriteChannel;
@@ -126,6 +128,7 @@ import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
+@BetaApi
 final class GrpcStorageImpl extends BaseService<StorageOptions> implements Storage {
 
   private static final byte[] ZERO_BYTES = new byte[0];
@@ -143,10 +146,11 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
    * walking overhead for every Code for every invocation of read, construct the set of exceptions
    * only once and keep in this value.
    */
-  private static final Set<ApiException> CODE_API_EXCEPTIONS =
+  private static final Set<StorageException> CODE_API_EXCEPTIONS =
       Arrays.stream(StatusCode.Code.values())
           .map(GrpcStorageImpl::statusCodeFor)
           .map(c -> ApiExceptionFactory.createException(null, c, false))
+          .map(StorageException::asStorageException)
           .collect(Collectors.toSet());
 
   private final StorageClient storageClient;
@@ -154,7 +158,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   private final GrpcRetryAlgorithmManager retryAlgorithmManager;
   private final SyntaxDecoders syntaxDecoders;
 
-  @Deprecated private final transient ProjectId defaultProjectId;
+  @Deprecated private final ProjectId defaultProjectId;
 
   GrpcStorageImpl(GrpcStorageOptions options, StorageClient storageClient) {
     super(options);
@@ -169,6 +173,9 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   public void close() throws Exception {
     try (StorageClient s = storageClient) {
       s.shutdownNow();
+      org.threeten.bp.Duration terminationAwaitDuration =
+          getOptions().getTerminationAwaitDuration();
+      s.awaitTermination(terminationAwaitDuration.toMillis(), TimeUnit.MILLISECONDS);
     }
   }
 
@@ -356,8 +363,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     GetBucketRequest.Builder builder =
         GetBucketRequest.newBuilder().setName(bucketNameCodec.encode(bucket));
     GetBucketRequest req = opts.getBucketsRequest().apply(builder).build();
-    // TODO(frankyn): Do we care about projection because Apiary uses FULL for projection? Missing
-    // projection=full
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
@@ -381,7 +386,9 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     GetObjectRequest.Builder builder =
-        GetObjectRequest.newBuilder().setBucket(blob.getBucket()).setObject(blob.getName());
+        GetObjectRequest.newBuilder()
+            .setBucket(bucketNameCodec.encode(blob.getBucket()))
+            .setObject(blob.getName());
     GetObjectRequest req = opts.getObjectsRequest().apply(builder).build();
     return Retrying.run(
         getOptions(),
@@ -409,9 +416,13 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .apply(ListBucketsRequest.newBuilder())
             .build();
     ListBucketsPagedResponse call = listBucketsCallable.call(request, grpcCallContext);
-    ListBucketsPage page = call.getPage();
-    return new TransformingPageDecorator<>(
-        page, syntaxDecoders.bucket, getOptions(), retryAlgorithmManager.getFor(request));
+    try {
+      ListBucketsPage page = call.getPage();
+      return new TransformingPageDecorator<>(
+          page, syntaxDecoders.bucket, getOptions(), retryAlgorithmManager.getFor(request));
+    } catch (Exception e) {
+      throw StorageException.coalesce(e);
+    }
   }
 
   @Override
@@ -424,10 +435,14 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     ListObjectsRequest.Builder builder =
         ListObjectsRequest.newBuilder().setParent(bucketNameCodec.encode(bucket));
     ListObjectsRequest req = opts.listObjectsRequest().apply(builder).build();
-    ListObjectsPagedResponse call = listObjectsCallable.call(req, grpcCallContext);
-    ListObjectsPage page = call.getPage();
-    return new TransformingPageDecorator<>(
-        page, syntaxDecoders.blob, getOptions(), retryAlgorithmManager.getFor(req));
+    try {
+      ListObjectsPagedResponse call = listObjectsCallable.call(req, grpcCallContext);
+      ListObjectsPage page = call.getPage();
+      return new TransformingPageDecorator<>(
+          page, syntaxDecoders.blob, getOptions(), retryAlgorithmManager.getFor(req));
+    } catch (Exception e) {
+      throw StorageException.coalesce(e);
+    }
   }
 
   @Override
@@ -489,8 +504,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
           () -> storageClient.deleteBucketCallable().call(req, grpcCallContext),
           Decoder.identity());
       return true;
-    } catch (ApiException e) {
-      // TODO: We should throw a StorageException instead of ApiException when making the
+    } catch (StorageException e) {
       return false;
     }
   }
@@ -516,8 +530,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
           () -> storageClient.deleteObjectCallable().call(req, grpcCallContext),
           Decoder.identity());
       return true;
-    } catch (ApiException e) {
-      // TODO: We should throw a StorageException instead of ApiException when making the
+    } catch (StorageException e) {
       return false;
     }
   }
@@ -554,7 +567,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public CopyWriter copy(CopyRequest copyRequest) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("copy", CopyRequest.class));
   }
 
   @Override
@@ -570,7 +583,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     try (UnbufferedReadableByteChannel r = session.open();
         WritableByteChannel w = Channels.newChannel(baos)) {
       ByteStreams.copy(r, w);
-    } catch (IOException e) {
+    } catch (ApiException | IOException e) {
       throw StorageException.coalesce(e);
     }
     return baos.toByteArray();
@@ -607,7 +620,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     try (UnbufferedReadableByteChannel r = session.open();
         WritableByteChannel w = Files.newByteChannel(path, WRITE_OPS)) {
       ByteStreams.copy(r, w);
-    } catch (IOException e) {
+    } catch (ApiException | IOException e) {
       throw StorageException.coalesce(e);
     }
   }
@@ -620,7 +633,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     try (UnbufferedReadableByteChannel r = session.open();
         WritableByteChannel w = Channels.newChannel(outputStream)) {
       ByteStreams.copy(r, w);
-    } catch (IOException e) {
+    } catch (ApiException | IOException e) {
       throw StorageException.coalesce(e);
     }
   }
@@ -721,132 +734,137 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public List<Blob> get(BlobId... blobIds) {
-    return todo();
+    return throwHttpJsonOnly(fmtMethodName("get", BlobId[].class));
   }
 
   @Override
   public List<Blob> get(Iterable<BlobId> blobIds) {
-    return todo();
+    return throwHttpJsonOnly(fmtMethodName("get", Iterable.class));
   }
 
   @Override
   public List<Blob> update(BlobInfo... blobInfos) {
-    return todo();
+    return throwHttpJsonOnly(fmtMethodName("update", BlobInfo[].class));
   }
 
   @Override
   public List<Blob> update(Iterable<BlobInfo> blobInfos) {
-    return todo();
+    return throwHttpJsonOnly(fmtMethodName("update", Iterable.class));
   }
 
   @Override
   public List<Boolean> delete(BlobId... blobIds) {
-    return todo();
+    return throwHttpJsonOnly(fmtMethodName("delete", BlobId[].class));
   }
 
   @Override
   public List<Boolean> delete(Iterable<BlobId> blobIds) {
-    return todo();
+    return throwHttpJsonOnly(fmtMethodName("delete", Iterable.class));
   }
 
   @Override
   public Acl getAcl(String bucket, Entity entity, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("getAcl", String.class, Entity.class, BucketSourceOption[].class));
   }
 
   @Override
   public Acl getAcl(String bucket, Entity entity) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("getAcl", String.class, Entity.class));
   }
 
   @Override
   public boolean deleteAcl(String bucket, Entity entity, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("deleteAcl", String.class, Entity.class, BucketSourceOption[].class));
   }
 
   @Override
   public boolean deleteAcl(String bucket, Entity entity) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("deleteAcl", String.class, Entity.class));
   }
 
   @Override
   public Acl createAcl(String bucket, Acl acl, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("createAcl", String.class, Acl.class, BucketSourceOption[].class));
   }
 
   @Override
   public Acl createAcl(String bucket, Acl acl) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("createAcl", String.class, Acl.class));
   }
 
   @Override
   public Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("updateAcl", String.class, Acl.class, BucketSourceOption[].class));
   }
 
   @Override
   public Acl updateAcl(String bucket, Acl acl) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("updateAcl", String.class, Acl.class));
   }
 
   @Override
   public List<Acl> listAcls(String bucket, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("listAcls", String.class, BucketSourceOption[].class));
   }
 
   @Override
   public List<Acl> listAcls(String bucket) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("listAcls", String.class));
   }
 
   @Override
   public Acl getDefaultAcl(String bucket, Entity entity) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("getDefaultAcl", String.class, Entity.class));
   }
 
   @Override
   public boolean deleteDefaultAcl(String bucket, Entity entity) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("deleteDefaultAcl", String.class, Entity.class));
   }
 
   @Override
   public Acl createDefaultAcl(String bucket, Acl acl) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("createDefaultAcl", String.class, Acl.class));
   }
 
   @Override
   public Acl updateDefaultAcl(String bucket, Acl acl) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("updateDefaultAcl", String.class, Acl.class));
   }
 
   @Override
   public List<Acl> listDefaultAcls(String bucket) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("listDefaultAcls", String.class));
   }
 
   @Override
   public Acl getAcl(BlobId blob, Entity entity) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("getAcl", BlobId.class, Entity.class));
   }
 
   @Override
   public boolean deleteAcl(BlobId blob, Entity entity) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("deleteAcl", BlobId.class, Entity.class));
   }
 
   @Override
   public Acl createAcl(BlobId blob, Acl acl) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("createAcl", BlobId.class, Acl.class));
   }
 
   @Override
   public Acl updateAcl(BlobId blob, Acl acl) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("updateAcl", BlobId.class, Acl.class));
   }
 
   @Override
   public List<Acl> listAcls(BlobId blob) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("listAcls", BlobId.class));
   }
 
   @Override
@@ -888,10 +906,14 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .andThen(opts.listHmacKeysRequest())
             .apply(ListHmacKeysRequest.newBuilder())
             .build();
-    ListHmacKeysPagedResponse call = listHmacKeysCallable.call(request, grpcCallContext);
-    ListHmacKeysPage page = call.getPage();
-    return new TransformingPageDecorator<>(
-        page, codecs.hmacKeyMetadata(), getOptions(), retryAlgorithmManager.getFor(request));
+    try {
+      ListHmacKeysPagedResponse call = listHmacKeysCallable.call(request, grpcCallContext);
+      ListHmacKeysPage page = call.getPage();
+      return new TransformingPageDecorator<>(
+          page, codecs.hmacKeyMetadata(), getOptions(), retryAlgorithmManager.getFor(request));
+    } catch (Exception e) {
+      throw StorageException.coalesce(e);
+    }
   }
 
   @Override
@@ -955,18 +977,21 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Policy getIamPolicy(String bucket, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("getIamPolicy", String.class, BucketSourceOption[].class));
   }
 
   @Override
   public Policy setIamPolicy(String bucket, Policy policy, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("setIamPolicy", String.class, Policy.class, BucketSourceOption[].class));
   }
 
   @Override
   public List<Boolean> testIamPermissions(
       String bucket, List<String> permissions, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("testIamPermissions", String.class, List.class, BucketSourceOption.class));
   }
 
   @Override
@@ -982,22 +1007,23 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Notification createNotification(String bucket, NotificationInfo notificationInfo) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("createNotification", String.class, NotificationInfo.class));
   }
 
   @Override
   public Notification getNotification(String bucket, String notificationId) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("getNotification", String.class, String.class));
   }
 
   @Override
   public List<Notification> listNotifications(String bucket) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("listNotifications", String.class));
   }
 
   @Override
   public boolean deleteNotification(String bucket, String notificationId) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("deleteNotification", String.class, String.class));
   }
 
   @Override
@@ -1005,10 +1031,17 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     return (GrpcStorageOptions) super.getOptions();
   }
 
+  boolean isClosed() {
+    return storageClient.isShutdown();
+  }
+
   private Blob getBlob(ApiFuture<WriteObjectResponse> result) {
-    // TODO: investigate if we need to unnest any exception
-    WriteObjectResponse response = ApiExceptions.callAndTranslateApiException(result);
-    return syntaxDecoders.blob.decode(response.getResource());
+    try {
+      WriteObjectResponse response = ApiExceptions.callAndTranslateApiException(result);
+      return syntaxDecoders.blob.decode(response.getResource());
+    } catch (Exception e) {
+      throw StorageException.coalesce(e);
+    }
   }
 
   /** Bind some decoders for our "Syntax" classes to this instance of GrpcStorageImpl */
@@ -1145,6 +1178,15 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     throw new UnsupportedOperationException(message);
   }
 
+  private <T> T throwNotYetImplemented(String methodName) {
+    String message =
+        String.format(
+            "%s#%s is not yet implemented for GRPC transport. Please use StorageOptions.http() to construct a compatible instance in the interim.",
+            Storage.class.getName(), methodName);
+    throw new UnimplementedException(
+        message, null, statusCodeFor(StatusCode.Code.UNIMPLEMENTED), false);
+  }
+
   private static String fmtMethodName(String name, Class<?>... args) {
     return name
         + "("
@@ -1222,7 +1264,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   static Set<StatusCode.Code> resultRetryAlgorithmToCodes(ResultRetryAlgorithm<?> alg) {
     return CODE_API_EXCEPTIONS.stream()
         .filter(e -> alg.shouldRetry(e, null))
-        .map(e -> e.getStatusCode().getCode())
+        .map(e -> e.apiExceptionCause.getStatusCode().getCode())
         .collect(Collectors.toSet());
   }
 
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 875231e3..1a24f400 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
@@ -24,6 +24,7 @@ import com.google.api.core.InternalApi;
 import com.google.api.gax.core.CredentialsProvider;
 import com.google.api.gax.core.FixedCredentialsProvider;
 import com.google.api.gax.core.NoCredentialsProvider;
+import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.retrying.StreamResumptionStrategy;
 import com.google.api.gax.rpc.HeaderProvider;
@@ -48,6 +49,7 @@ import java.net.URI;
 import java.util.Set;
 import org.checkerframework.checker.nullness.qual.NonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
+import org.threeten.bp.Duration;
 
 @BetaApi
 @TransportCompatibility(Transport.GRPC)
@@ -60,6 +62,7 @@ public final class GrpcStorageOptions extends StorageOptions
   private static final String DEFAULT_HOST = "https://storage.googleapis.com";
 
   private final GrpcRetryAlgorithmManager retryAlgorithmManager;
+  private final Duration terminationAwaitDuration;
 
   @BetaApi
   public GrpcStorageOptions(Builder builder, GrpcStorageDefaults serviceDefaults) {
@@ -68,6 +71,9 @@ public final class GrpcStorageOptions extends StorageOptions
         new GrpcRetryAlgorithmManager(
             MoreObjects.firstNonNull(
                 builder.storageRetryStrategy, serviceDefaults.getStorageRetryStrategy()));
+    this.terminationAwaitDuration =
+        MoreObjects.firstNonNull(
+            builder.terminationAwaitDuration, serviceDefaults.getTerminationAwaitDuration());
   }
 
   @Override
@@ -80,6 +86,11 @@ public final class GrpcStorageOptions extends StorageOptions
     return retryAlgorithmManager;
   }
 
+  @InternalApi
+  Duration getTerminationAwaitDuration() {
+    return terminationAwaitDuration;
+  }
+
   @InternalApi
   StorageSettings getStorageSettings() throws IOException {
     URI uri = URI.create(getHost());
@@ -97,21 +108,33 @@ public final class GrpcStorageOptions extends StorageOptions
     StorageSettings.Builder builder =
         StorageSettings.newBuilder()
             .setEndpoint(endpoint)
-            .setCredentialsProvider(credentialsProvider);
+            .setCredentialsProvider(credentialsProvider)
+            .setClock(getClock());
 
     if (scheme.equals("http")) {
       builder.setTransportChannelProvider(
-          com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.newBuilder()
+          InstantiatingGrpcChannelProvider.newBuilder()
               .setEndpoint(endpoint)
               .setChannelConfigurator(ManagedChannelBuilder::usePlaintext)
               .build());
     }
-    RetrySettings retrySettings = getRetrySettings();
-    RetrySettings attemptOnce = retrySettings.toBuilder().setMaxAttempts(1).build();
+    RetrySettings readRetrySettings =
+        getRetrySettings()
+            .toBuilder()
+            // when performing a read via ReadObject, the ServerStream will have a default relative
+            // deadline set of `requestStartTime() + totalTimeout`, meaning if the specified
+            // RetrySettings have a totalTimeout of 10 seconds -- which should be plenty for
+            // metadata RPCs -- the entire ReadObject stream would need to complete within 10
+            // seconds.
+            // To allow read streams to have longer lifespans, crank up their timeouts, instead rely
+            // on idleTimeout below.
+            .setLogicalTimeout(Duration.ofDays(28))
+            .build();
+    Duration totalTimeout = getRetrySettings().getTotalTimeout();
     // all retries for unary methods are handled at a different level
     builder.applyToAllUnaryMethods(
         input -> {
-          input.setRetrySettings(attemptOnce);
+          input.setSimpleTimeoutNoRetries(totalTimeout);
           return null;
         });
     // for ReadObject we are configuring the server stream handling to do its own retries, so wire
@@ -119,13 +142,17 @@ public final class GrpcStorageOptions extends StorageOptions
     // considerations need to be made.
     builder
         .readObjectSettings()
-        .setRetrySettings(retrySettings)
+        .setRetrySettings(readRetrySettings)
         // even though we might want to default to the empty set for retryable codes, don't ever
         // actually do this. Doing so prevents any retry capability from being wired into the stream
         // pipeline, ever.
         // For our use, we will always set it one way or the other to ensure it's appropriate
         // DO NOT: .setRetryableCodes(Collections.emptySet())
-        .setResumptionStrategy(new ReadObjectResumptionStrategy());
+        .setResumptionStrategy(new ReadObjectResumptionStrategy())
+        // for reads, the stream can be held open for a long time in order to read all bytes,
+        // this is totally valid. instead we want to monitor if the stream is doing work and if not
+        // timeout.
+        .setIdleTimeout(totalTimeout);
     return builder.build();
   }
 
@@ -142,7 +169,7 @@ public final class GrpcStorageOptions extends StorageOptions
 
   @Override
   public boolean equals(Object obj) {
-    return obj instanceof StorageOptions && baseEquals((StorageOptions) obj);
+    return obj instanceof GrpcStorageOptions && baseEquals((GrpcStorageOptions) obj);
   }
 
   @BetaApi
@@ -160,10 +187,22 @@ public final class GrpcStorageOptions extends StorageOptions
     return GrpcStorageOptions.GrpcStorageDefaults.INSTANCE;
   }
 
+  // since our new GrpcStorageImpl can "close" we need to help ServiceOptions know whether it can
+  // use it's cached instance.
+  @Override
+  protected boolean shouldRefreshService(Storage cachedService) {
+    if (cachedService instanceof GrpcStorageImpl) {
+      GrpcStorageImpl service = (GrpcStorageImpl) cachedService;
+      return service.isClosed();
+    }
+    return super.shouldRefreshService(cachedService);
+  }
+
   @BetaApi
   public static class Builder extends StorageOptions.Builder {
 
     private StorageRetryStrategy storageRetryStrategy;
+    private Duration terminationAwaitDuration;
 
     Builder() {}
 
@@ -171,6 +210,20 @@ public final class GrpcStorageOptions extends StorageOptions
       super(options);
     }
 
+    /**
+     * Set the maximum duration in which to await termination of any outstanding requests when
+     * calling {@link Storage#close()}
+     *
+     * @param terminationAwaitDuration a non-null Duration to use
+     * @return the builder
+     */
+    @BetaApi
+    public Builder setTerminationAwaitDuration(Duration terminationAwaitDuration) {
+      this.terminationAwaitDuration =
+          requireNonNull(terminationAwaitDuration, "terminationAwaitDuration must be non null");
+      return this;
+    }
+
     @BetaApi
     @Override
     public GrpcStorageOptions.Builder setTransportOptions(TransportOptions transportOptions) {
@@ -310,6 +363,11 @@ public final class GrpcStorageOptions extends StorageOptions
     public StorageRetryStrategy getStorageRetryStrategy() {
       return StorageRetryStrategy.getDefaultStorageRetryStrategy();
     }
+
+    @BetaApi
+    public Duration getTerminationAwaitDuration() {
+      return Duration.ofMinutes(1);
+    }
   }
 
   /**
@@ -338,6 +396,7 @@ public final class GrpcStorageOptions extends StorageOptions
      *     GrpcStorageOptions.GrpcStorageDefaults#getDefaultServiceFactory()
      *     GrpcStorageOptions.defaults().getDefaultServiceFactory()}
      */
+    // this class needs to be public due to ServiceOptions forName'ing it in it's readObject method
     @InternalApi
     @Deprecated
     @SuppressWarnings("DeprecatedIsStillUsed")
@@ -371,6 +430,8 @@ public final class GrpcStorageOptions extends StorageOptions
    * @see GrpcStorageOptions.GrpcStorageDefaults#getDefaultRpcFactory()
    */
   @InternalApi
+  @BetaApi
+  @Deprecated
   public static class GrpcStorageRpcFactory implements StorageRpcFactory {
 
     /**
@@ -385,6 +446,7 @@ public final class GrpcStorageOptions extends StorageOptions
      * @deprecated instead use {@link GrpcStorageOptions.GrpcStorageDefaults#getDefaultRpcFactory()
      *     GrpcStorageOptions.defaults().getDefaultRpcFactory()}
      */
+    // this class needs to be public due to ServiceOptions forName'ing it in it's readObject method
     @InternalApi
     @Deprecated
     @SuppressWarnings("DeprecatedIsStillUsed")
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java
index e3414706..fc6176ea 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java
@@ -33,6 +33,7 @@ import com.google.cloud.storage.spi.v1.HttpStorageRpc;
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableSet;
+import java.io.Serializable;
 import java.util.Set;
 
 // non-final because of mocking frameworks
@@ -239,7 +240,8 @@ public class HttpStorageOptions extends StorageOptions {
    * @see HttpStorageDefaults#getDefaultServiceFactory()
    */
   @InternalApi
-  public static class HttpStorageFactory implements StorageFactory {
+  public static class HttpStorageFactory implements StorageFactory, Serializable {
+    private static final long serialVersionUID = -509515936479640159L;
 
     /**
      * Internal implementation detail, only public to allow for {@link java.io.Serializable}.
@@ -253,6 +255,7 @@ public class HttpStorageOptions extends StorageOptions {
      * @deprecated instead use {@link HttpStorageDefaults#getDefaultServiceFactory()
      *     HttpStorageOptions.defaults().getDefaultServiceFactory()}
      */
+    // this class needs to be public due to ServiceOptions forName'ing it in it's readObject method
     @InternalApi
     @Deprecated
     @SuppressWarnings("DeprecatedIsStillUsed")
@@ -280,7 +283,8 @@ public class HttpStorageOptions extends StorageOptions {
    * @see HttpStorageDefaults#getDefaultRpcFactory()
    */
   @InternalApi
-  public static class HttpStorageRpcFactory implements StorageRpcFactory {
+  public static class HttpStorageRpcFactory implements StorageRpcFactory, Serializable {
+    private static final long serialVersionUID = -7487671939555953705L;
 
     /**
      * Internal implementation detail, only public to allow for {@link java.io.Serializable}.
@@ -294,6 +298,7 @@ public class HttpStorageOptions extends StorageOptions {
      * @deprecated instead use {@link HttpStorageDefaults#getDefaultRpcFactory()
      *     HttpStorageOptions.defaults().getDefaultRpcFactory()}
      */
+    // this class needs to be public due to ServiceOptions forName'ing it in it's readObject method
     @InternalApi
     @Deprecated
     @SuppressWarnings("DeprecatedIsStillUsed")
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
index b804ba4c..47996e7b 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
@@ -24,6 +24,7 @@ import com.google.storage.v2.StartResumableWriteResponse;
 import com.google.storage.v2.WriteObjectRequest;
 import java.util.Objects;
 import java.util.function.Function;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 final class ResumableWrite implements WriteObjectRequestBuilderFactory {
 
@@ -55,6 +56,14 @@ final class ResumableWrite implements WriteObjectRequestBuilderFactory {
     return writeRequest.toBuilder();
   }
 
+  @Override
+  public @Nullable String bucketName() {
+    if (req.hasWriteObjectSpec() && req.getWriteObjectSpec().hasResource()) {
+      return req.getWriteObjectSpec().getResource().getBucket();
+    }
+    return null;
+  }
+
   @Override
   public String toString() {
     return "ResumableWrite{" + "req=" + fmtProto(req) + ", res=" + fmtProto(res) + '}';
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
index 8c22ba6e..752cd199 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
@@ -3743,6 +3743,10 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
   @TransportCompatibility({Transport.HTTP})
   boolean deleteNotification(String bucket, String notificationId);
 
+  /**
+   * @throws InterruptedException thrown if interrupted while awaiting termination of underlying
+   *     resources
+   */
   @Override
   default void close() throws Exception {}
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
index 708de195..6f613d09 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
@@ -56,24 +56,30 @@ public final class StorageException extends BaseHttpServiceException {
 
   private static final long serialVersionUID = -4168430271327813063L;
 
+  final ApiException apiExceptionCause;
+
   public StorageException(int code, String message) {
     this(code, message, null);
   }
 
   public StorageException(int code, String message, Throwable cause) {
     super(code, message, null, true, RETRYABLE_ERRORS, cause);
+    this.apiExceptionCause = asApiExceptionOrNull(cause);
   }
 
   public StorageException(int code, String message, String reason, Throwable cause) {
     super(code, message, reason, true, RETRYABLE_ERRORS, cause);
+    this.apiExceptionCause = asApiExceptionOrNull(cause);
   }
 
   public StorageException(IOException exception) {
     super(exception, true, RETRYABLE_ERRORS);
+    this.apiExceptionCause = null;
   }
 
   public StorageException(GoogleJsonError error) {
     super(error, true, RETRYABLE_ERRORS);
+    this.apiExceptionCause = null;
   }
 
   /**
@@ -108,35 +114,36 @@ public final class StorageException extends BaseHttpServiceException {
       return (BaseServiceException) t.getCause();
     }
     if (t instanceof ApiException) {
-      ApiException apiEx = (ApiException) t;
-
-      // https://cloud.google.com/storage/docs/json_api/v1/status-codes
-      // https://cloud.google.com/apis/design/errors#http_mapping
-      // https://cloud.google.com/apis/design/errors#error_payloads
-      // TODO: flush this out more to wire through "error" and "details" when we're able to get real
-      //   errors from GCS
-      int httpStatusCode = 0;
-      StatusCode statusCode = apiEx.getStatusCode();
-      if (statusCode instanceof GrpcStatusCode) {
-        GrpcStatusCode gsc = (GrpcStatusCode) statusCode;
-        httpStatusCode =
-            BackwardCompatibilityUtils.grpcCodeToHttpStatusCode(gsc.getTransportCode());
-      }
-      // If there is a gRPC exception in our cause change pull it's error message up to be our
-      // message otherwise, create a generic error message with the status code.
-      String message =
-          Utils.firstNonNull(
-              () -> getStatusExceptionMessage(apiEx),
-              () -> String.format("Error: %s", statusCode.getCode().name()));
-
-      // It'd be better to use ExceptionData and BaseServiceException#<init>(ExceptionData) but,
-      // BaseHttpServiceException does not pass that through so we're stuck using this for now.
-      // TODO: When we can break the coupling to BaseHttpServiceException replace this
-      return new StorageException(httpStatusCode, message, apiEx.getReason(), apiEx);
+      return asStorageException((ApiException) t);
     }
     return getStorageException(t);
   }
 
+  static StorageException asStorageException(ApiException apiEx) {
+    // https://cloud.google.com/storage/docs/json_api/v1/status-codes
+    // https://cloud.google.com/apis/design/errors#http_mapping
+    // https://cloud.google.com/apis/design/errors#error_payloads
+    // TODO: flush this out more to wire through "error" and "details" when we're able to get real
+    //   errors from GCS
+    int httpStatusCode = 0;
+    StatusCode statusCode = apiEx.getStatusCode();
+    if (statusCode instanceof GrpcStatusCode) {
+      GrpcStatusCode gsc = (GrpcStatusCode) statusCode;
+      httpStatusCode = BackwardCompatibilityUtils.grpcCodeToHttpStatusCode(gsc.getTransportCode());
+    }
+    // If there is a gRPC exception in our cause change pull it's error message up to be our
+    // message otherwise, create a generic error message with the status code.
+    String message =
+        Utils.firstNonNull(
+            () -> getStatusExceptionMessage(apiEx),
+            () -> String.format("Error: %s", statusCode.getCode().name()));
+
+    // It'd be better to use ExceptionData and BaseServiceException#<init>(ExceptionData) but,
+    // BaseHttpServiceException does not pass that through so we're stuck using this for now.
+    // TODO: When we can break the coupling to BaseHttpServiceException replace this
+    return new StorageException(httpStatusCode, message, apiEx.getReason(), apiEx);
+  }
+
   /**
    * Translate IOException to a StorageException representing the cause of the error. This method
    * defaults to idempotent always being {@code true}. Additionally, this method translates
@@ -164,4 +171,13 @@ public final class StorageException extends BaseHttpServiceException {
     }
     return null;
   }
+
+  @Nullable
+  private static ApiException asApiExceptionOrNull(Throwable cause) {
+    if (cause instanceof ApiException) {
+      return (ApiException) cause;
+    } else {
+      return null;
+    }
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java
index 9e55d8cc..2488cfcb 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java
@@ -23,6 +23,7 @@ import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
 import com.google.storage.v2.WriteObjectRequest;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 final class WriteCtx<RequestFactoryT extends WriteObjectRequestBuilderFactory> {
 
@@ -74,10 +75,12 @@ final class WriteCtx<RequestFactoryT extends WriteObjectRequestBuilderFactory> {
         + '}';
   }
 
-  @FunctionalInterface
   interface WriteObjectRequestBuilderFactory {
     WriteObjectRequest.Builder newBuilder();
 
+    @Nullable
+    String bucketName();
+
     static SimpleWriteObjectRequestBuilderFactory simple(WriteObjectRequest req) {
       return new SimpleWriteObjectRequestBuilderFactory(req);
     }
@@ -96,6 +99,14 @@ final class WriteCtx<RequestFactoryT extends WriteObjectRequestBuilderFactory> {
       return req.toBuilder();
     }
 
+    @Override
+    public @Nullable String bucketName() {
+      if (req.hasWriteObjectSpec() && req.getWriteObjectSpec().hasResource()) {
+        return req.getWriteObjectSpec().getResource().getBucket();
+      }
+      return null;
+    }
+
     @Override
     public String toString() {
       return "SimpleWriteObjectRequestBuilderFactory{" + "req=" + fmtProto(req) + '}';
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
index 8dfe99f2..6da9c5e3 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
@@ -17,9 +17,11 @@
 package com.google.cloud.storage;
 
 import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.api.gax.rpc.ClientStreamingCallable;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.storage.v2.WriteObjectRequest;
 import com.google.storage.v2.WriteObjectResponse;
 import java.util.List;
@@ -51,22 +53,34 @@ final class WriteFlushStrategy {
    */
   static FlusherFactory fsyncEveryFlush(
       ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write) {
-    return (LongConsumer committedTotalBytesCallback,
+    return (String bucketName,
+        LongConsumer committedTotalBytesCallback,
         Consumer<WriteObjectResponse> onSuccessCallback) ->
-        new FsyncEveryFlusher(write, committedTotalBytesCallback, onSuccessCallback);
+        new FsyncEveryFlusher(write, bucketName, committedTotalBytesCallback, onSuccessCallback);
   }
 
   /**
    * Create a {@link Flusher} which will "fsync" only on {@link Flusher#close(WriteObjectRequest)}.
    * Calls to {@link Flusher#flush(List)} will be sent but not synced.
    *
-   * @see FlusherFactory#newFlusher(LongConsumer, Consumer)
+   * @see FlusherFactory#newFlusher(String, LongConsumer, Consumer)
    */
   static FlusherFactory fsyncOnClose(
       ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write) {
-    return (LongConsumer committedTotalBytesCallback,
+    return (String bucketName,
+        LongConsumer committedTotalBytesCallback,
         Consumer<WriteObjectResponse> onSuccessCallback) ->
-        new FsyncOnClose(write, committedTotalBytesCallback, onSuccessCallback);
+        new FsyncOnClose(write, bucketName, committedTotalBytesCallback, onSuccessCallback);
+  }
+
+  private static GrpcCallContext contextWithBucketName(String bucketName) {
+    GrpcCallContext ret = GrpcCallContext.createDefault();
+    if (bucketName != null && !bucketName.isEmpty()) {
+      return ret.withExtraHeaders(
+          ImmutableMap.of(
+              "x-goog-request-params", ImmutableList.of(String.format("bucket=%s", bucketName))));
+    }
+    return ret;
   }
 
   @FunctionalInterface
@@ -77,7 +91,9 @@ final class WriteFlushStrategy {
      * @param onSuccessCallback Callback to signal success, and provide the final response.
      */
     Flusher newFlusher(
-        LongConsumer committedTotalBytesCallback, Consumer<WriteObjectResponse> onSuccessCallback);
+        String bucketName,
+        LongConsumer committedTotalBytesCallback,
+        Consumer<WriteObjectResponse> onSuccessCallback);
   }
 
   interface Flusher {
@@ -89,14 +105,17 @@ final class WriteFlushStrategy {
   private static final class FsyncEveryFlusher implements Flusher {
 
     private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+    private final String bucketName;
     private final LongConsumer sizeCallback;
     private final Consumer<WriteObjectResponse> completeCallback;
 
     private FsyncEveryFlusher(
         ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+        String bucketName,
         LongConsumer sizeCallback,
         Consumer<WriteObjectResponse> completeCallback) {
       this.write = write;
+      this.bucketName = bucketName;
       this.sizeCallback = sizeCallback;
       this.completeCallback = completeCallback;
     }
@@ -104,7 +123,9 @@ final class WriteFlushStrategy {
     public void flush(@NonNull List<WriteObjectRequest> segments) {
 
       Observer observer = new Observer(sizeCallback, completeCallback);
-      ApiStreamObserver<WriteObjectRequest> write = this.write.clientStreamingCall(observer);
+      GrpcCallContext internalContext = contextWithBucketName(bucketName);
+      ApiStreamObserver<WriteObjectRequest> write =
+          this.write.withDefaultCallContext(internalContext).clientStreamingCall(observer);
 
       boolean first = true;
       for (WriteObjectRequest message : segments) {
@@ -129,6 +150,7 @@ final class WriteFlushStrategy {
   private static final class FsyncOnClose implements Flusher {
 
     private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+    private final String bucketName;
     private final Observer responseObserver;
 
     private volatile ApiStreamObserver<WriteObjectRequest> stream;
@@ -136,9 +158,11 @@ final class WriteFlushStrategy {
 
     private FsyncOnClose(
         ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+        String bucketName,
         LongConsumer sizeCallback,
         Consumer<WriteObjectResponse> completeCallback) {
       this.write = write;
+      this.bucketName = bucketName;
       this.responseObserver = new Observer(sizeCallback, completeCallback);
     }
 
@@ -172,7 +196,11 @@ final class WriteFlushStrategy {
       if (stream == null) {
         synchronized (this) {
           if (stream == null) {
-            stream = write.clientStreamingCall(responseObserver);
+            GrpcCallContext internalContext = contextWithBucketName(bucketName);
+            stream =
+                this.write
+                    .withDefaultCallContext(internalContext)
+                    .clientStreamingCall(responseObserver);
           }
         }
       }


Test patch:
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 ef3d7985..b84a59a0 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
@@ -112,7 +112,7 @@ public final class BackwardCompatibilityUtilsTest {
     expected.put(Code.ABORTED, 0);
     expected.put(Code.CANCELLED, 0);
     expected.put(Code.UNKNOWN, 0);
-    expected.put(Code.DEADLINE_EXCEEDED, 0);
+    expected.put(Code.DEADLINE_EXCEEDED, 504);
     expected.put(Code.DATA_LOSS, 0);
 
     EnumMap<Code, Integer> actual = new EnumMap<>(Code.class);
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketFixture.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketFixture.java
index ef1a79ad..c81881dc 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketFixture.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketFixture.java
@@ -18,13 +18,21 @@ package com.google.cloud.storage;
 
 import static java.util.Objects.requireNonNull;
 
+import com.google.api.gax.paging.Page;
 import com.google.cloud.storage.conformance.retry.CleanupStrategy;
-import com.google.cloud.storage.testing.RemoteStorageHelper;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 import org.junit.rules.TestRule;
 import org.junit.runner.Description;
 import org.junit.runners.model.Statement;
@@ -94,7 +102,30 @@ public final class BucketFixture implements TestRule {
   private void doCleanup(String bucketName, Storage s) {
     LOGGER.info("Starting bucket cleanup...");
     try {
-      RemoteStorageHelper.forceDelete(s, bucketName, 5, TimeUnit.MINUTES);
+      // TODO: probe bucket existence, a bad test could have deleted the bucket
+      Page<Blob> page1 = s.list(bucketName);
+
+      int availableProcessors = Runtime.getRuntime().availableProcessors();
+      ListeningExecutorService exec =
+          MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2 * availableProcessors));
+
+      List<ListenableFuture<DeleteResult>> futures =
+          StreamSupport.stream(page1.iterateAll().spliterator(), false)
+              .map(b -> exec.submit(() -> new DeleteResult(b.getName(), b.delete())))
+              .collect(Collectors.toList());
+
+      List<DeleteResult> deleteResults = Futures.allAsList(futures).get(5, TimeUnit.MINUTES);
+      List<DeleteResult> failedDeletes =
+          deleteResults.stream().filter(r -> !r.success).collect(Collectors.toList());
+      failedDeletes.forEach(
+          r -> LOGGER.warning(String.format("Failed to delete object %s/%s", bucketName, r.name)));
+
+      if (failedDeletes.isEmpty()) {
+        s.delete(bucketName);
+      } else {
+        LOGGER.warning("Unable to delete bucket due to previous failed object deletes");
+      }
+      exec.shutdown();
       LOGGER.info("Bucket cleanup complete");
     } catch (Exception e) {
       LOGGER.log(Level.SEVERE, e, () -> "Error during bucket cleanup.");
@@ -127,4 +158,14 @@ public final class BucketFixture implements TestRule {
           requireNonNull(handle, "handle must be non null"), bucketNameFmtString, cleanupStrategy);
     }
   }
+
+  private static final class DeleteResult {
+    private final String name;
+    private final boolean success;
+
+    DeleteResult(String name, boolean success) {
+      this.name = name;
+      this.success = success;
+    }
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicReadTimeoutTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicReadTimeoutTest.java
new file mode 100644
index 00000000..d18d2c54
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicReadTimeoutTest.java
@@ -0,0 +1,219 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static com.google.cloud.storage.TestUtils.apiException;
+import static com.google.cloud.storage.TestUtils.getChecksummedData;
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+
+import com.google.api.gax.retrying.RetrySettings;
+import com.google.api.gax.rpc.ServerStream;
+import com.google.api.gax.rpc.WatchdogTimeoutException;
+import com.google.common.base.Stopwatch;
+import com.google.protobuf.ByteString;
+import com.google.storage.v2.ContentRange;
+import com.google.storage.v2.Object;
+import com.google.storage.v2.ReadObjectRequest;
+import com.google.storage.v2.ReadObjectResponse;
+import com.google.storage.v2.StorageClient;
+import com.google.storage.v2.StorageGrpc;
+import com.google.storage.v2.StorageSettings;
+import io.grpc.Status.Code;
+import io.grpc.stub.StreamObserver;
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.concurrent.TimeUnit;
+import org.junit.Test;
+import org.threeten.bp.Duration;
+
+/**
+ * ReadObject leverages gRPC ServerStream to read a stream of ReadObjectResponse messages spanning
+ * the size of the object. If a large object is downloaded and it would take longer than the
+ * totalTimeout from {@link StorageOptions#getRetrySettings()}{@link RetrySettings#getTotalTimeout()
+ * #getTotalTimeout()}, gax will interrupt the stream with a DEADLINE_EXCEEDED error.
+ *
+ * <p>Instead of relying on total stream timeout, we rely on idleTimeout for the stream via {@link
+ * com.google.api.gax.rpc.ServerStreamingCallSettings.Builder#setIdleTimeout(Duration)}.
+ *
+ * <p>These tests force specific timeout scenarios to happen against an in-process grpc server to
+ * ensure our configuration of the StorageClient properly translates to the behavior we want.
+ *
+ * <p><i>NOTE:</i>Unfortunately, these tests are slow as they are waiting on wall clock time in
+ * several circumstances.
+ */
+public final class GapicReadTimeoutTest {
+
+  private final String objectName = "name";
+  private final Object expectedResult =
+      Object.newBuilder()
+          .setName(objectName)
+          .setGeneration(3L)
+          .setContentType("application/octet-stream")
+          .build();
+
+  @Test
+  public void readObjectStreamTimeoutIsLongerThanDefaultTotalTimeout()
+      throws IOException, InterruptedException {
+    byte[] bytes = DataGenerator.base64Characters().genBytes(40);
+    ByteString data1 = ByteString.copyFrom(bytes, 0, 20);
+    ByteString data2 = ByteString.copyFrom(bytes, 20, 20);
+
+    ReadObjectRequest req1 =
+        ReadObjectRequest.newBuilder().setObject(objectName).setReadOffset(0).build();
+    ReadObjectResponse resp1 =
+        ReadObjectResponse.newBuilder()
+            .setMetadata(expectedResult)
+            .setContentRange(ContentRange.newBuilder().setStart(0).build())
+            .setChecksummedData(getChecksummedData(data1))
+            .build();
+    ReadObjectResponse resp2 =
+        ReadObjectResponse.newBuilder()
+            .setContentRange(ContentRange.newBuilder().setStart(20).build())
+            .setChecksummedData(getChecksummedData(data2))
+            .build();
+
+    int totalTimeoutMillis = 10_000;
+    int sleepDurationMillis = totalTimeoutMillis + 5_000;
+    StorageGrpc.StorageImplBase fakeStorage =
+        new StorageGrpc.StorageImplBase() {
+          @Override
+          public void readObject(
+              ReadObjectRequest request, StreamObserver<ReadObjectResponse> out) {
+            if (request.equals(req1)) {
+              out.onNext(resp1);
+              out.onNext(resp2);
+              try {
+                Thread.sleep(sleepDurationMillis);
+              } catch (InterruptedException e) {
+                out.onError(e);
+              }
+              out.onCompleted();
+            } else {
+              out.onError(apiException(Code.PERMISSION_DENIED));
+            }
+          }
+        };
+
+    try (FakeServer server = FakeServer.of(fakeStorage)) {
+      StorageSettings settings =
+          server
+              .getGrpcStorageOptions()
+              .toBuilder()
+              .setRetrySettings(
+                  RetrySettings.newBuilder()
+                      .setMaxAttempts(3)
+                      .setTotalTimeout(Duration.ofMillis(totalTimeoutMillis))
+                      .build())
+              .build()
+              .getStorageSettings();
+      try (StorageClient sc = StorageClient.create(settings)) {
+        ServerStream<ReadObjectResponse> ss = sc.readObjectCallable().call(req1);
+        Iterator<ReadObjectResponse> iter = ss.iterator();
+
+        Stopwatch started = Stopwatch.createStarted();
+        ReadObjectResponse actualResponse1 = iter.next();
+        ReadObjectResponse actualResponse2 = iter.next();
+        boolean hasNext = iter.hasNext();
+        Stopwatch stop = started.stop();
+        assertThat(stop.elapsed(TimeUnit.MILLISECONDS)).isAtLeast(sleepDurationMillis);
+        assertThat(actualResponse1).isEqualTo(resp1);
+        assertThat(actualResponse2).isEqualTo(resp2);
+        assertThat(hasNext).isFalse();
+      }
+    }
+  }
+
+  @Test
+  public void ifTheStreamIsIdleItWillBeCanceled() throws IOException, InterruptedException {
+    byte[] bytes = DataGenerator.base64Characters().genBytes(40);
+    ReadObjectRequest req1 =
+        ReadObjectRequest.newBuilder().setObject(objectName).setReadOffset(0).build();
+    ReadObjectResponse resp1 =
+        ReadObjectResponse.newBuilder()
+            .setMetadata(expectedResult)
+            .setContentRange(ContentRange.newBuilder().setStart(0).build())
+            .setChecksummedData(getChecksummedData(ByteString.copyFrom(bytes, 0, 10)))
+            .build();
+
+    ReadObjectResponse resp2 =
+        ReadObjectResponse.newBuilder()
+            .setContentRange(ContentRange.newBuilder().setStart(10).build())
+            .setChecksummedData(getChecksummedData(ByteString.copyFrom(bytes, 10, 10)))
+            .build();
+    ReadObjectResponse resp3 =
+        ReadObjectResponse.newBuilder()
+            .setContentRange(ContentRange.newBuilder().setStart(20).build())
+            .setChecksummedData(getChecksummedData(ByteString.copyFrom(bytes, 20, 10)))
+            .build();
+    ReadObjectResponse resp4 =
+        ReadObjectResponse.newBuilder()
+            .setContentRange(ContentRange.newBuilder().setStart(30).build())
+            .setChecksummedData(getChecksummedData(ByteString.copyFrom(bytes, 30, 10)))
+            .build();
+
+    int totalTimeoutMillis = 10_000;
+    int sleepDurationMillis = totalTimeoutMillis + 5_000;
+    StorageGrpc.StorageImplBase fakeStorage =
+        new StorageGrpc.StorageImplBase() {
+          @Override
+          public void readObject(
+              ReadObjectRequest request, StreamObserver<ReadObjectResponse> out) {
+            if (request.equals(req1)) {
+              out.onNext(resp1);
+              out.onNext(resp2);
+              out.onNext(resp3);
+              out.onNext(resp4);
+              out.onCompleted();
+            } else {
+              out.onError(apiException(Code.PERMISSION_DENIED));
+            }
+          }
+        };
+
+    try (FakeServer server = FakeServer.of(fakeStorage)) {
+      StorageSettings settings =
+          server
+              .getGrpcStorageOptions()
+              .toBuilder()
+              .setRetrySettings(
+                  RetrySettings.newBuilder()
+                      .setMaxAttempts(3)
+                      .setTotalTimeout(Duration.ofMillis(totalTimeoutMillis))
+                      .build())
+              .build()
+              .getStorageSettings();
+      try (StorageClient sc = StorageClient.create(settings)) {
+        ServerStream<ReadObjectResponse> ss = sc.readObjectCallable().call(req1);
+        Iterator<ReadObjectResponse> iter = ss.iterator();
+
+        Stopwatch started = Stopwatch.createStarted();
+        assertThrows(
+            WatchdogTimeoutException.class,
+            () -> {
+              iter.next();
+              Thread.sleep(20_000);
+              iter.next();
+              iter.next();
+              iter.next();
+            });
+        Stopwatch stop = started.stop();
+        assertThat(stop.elapsed(TimeUnit.MILLISECONDS)).isAtLeast(sleepDurationMillis);
+      }
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
index bc32ccea..70ad6a34 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
@@ -77,7 +77,7 @@ public final class GapicUnbufferedWritableByteChannelTest {
         WriteObjectResponse.newBuilder().setResource(obj.toBuilder().setSize(40)).build();
 
     WriteObjectRequest base = WriteObjectRequest.newBuilder().setWriteObjectSpec(spec).build();
-    WriteObjectRequestBuilderFactory reqFactory = base::toBuilder;
+    WriteObjectRequestBuilderFactory reqFactory = WriteObjectRequestBuilderFactory.simple(base);
 
     StorageImplBase service =
         new DirectWriteService(
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITStorageLifecycleTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITStorageLifecycleTest.java
new file mode 100644
index 00000000..c3e9c32b
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITStorageLifecycleTest.java
@@ -0,0 +1,117 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.conformance.retry.TestBench;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+/**
+ * The interaction of {@link com.google.cloud.ServiceOptions} instance caching has differing
+ * behavior depending on whether {@link StorageOptions#grpc()} or {@link StorageOptions#http()} are
+ * used.
+ *
+ * <p>Define some tests to ensue we are correctly integrating with the caching lifecycle
+ */
+// Not in com.google.cloud.storage.it because we're testing package local things
+public final class ITStorageLifecycleTest {
+  @ClassRule(order = 1)
+  public static final TestBench TEST_BENCH =
+      TestBench.newBuilder().setContainerName("lifecycle-test").build();
+
+  @Test
+  public void grpc() throws Exception {
+    GrpcStorageOptions options =
+        StorageOptions.grpc()
+            .setHost(TEST_BENCH.getGRPCBaseUri())
+            .setCredentials(NoCredentials.getInstance())
+            .setProjectId("test-project-id")
+            .build();
+
+    Storage service1 = options.getService();
+    Storage service2 = options.getService();
+
+    // ensure both instances are the same
+    assertThat(service2).isSameInstanceAs(service1);
+
+    // make sure and RPCs can be done
+    List<Bucket> buckets1 =
+        StreamSupport.stream(service1.list().iterateAll().spliterator(), false)
+            .collect(Collectors.toList());
+    assertThat(buckets1).isEmpty();
+    List<Bucket> buckets2 =
+        StreamSupport.stream(service2.list().iterateAll().spliterator(), false)
+            .collect(Collectors.toList());
+    assertThat(buckets2).isEmpty();
+
+    // close the instance
+    service1.close();
+
+    // expect a new instance to be returned
+    Storage service3 = options.getService();
+
+    assertThat(service3).isNotSameInstanceAs(service1);
+    // make sure and RPCs can be done
+    List<Bucket> buckets3 =
+        StreamSupport.stream(service3.list().iterateAll().spliterator(), false)
+            .collect(Collectors.toList());
+    assertThat(buckets3).isEmpty();
+  }
+
+  @Test
+  public void http() throws Exception {
+    HttpStorageOptions options =
+        StorageOptions.http()
+            .setHost(TEST_BENCH.getBaseUri())
+            .setCredentials(NoCredentials.getInstance())
+            .setProjectId("test-project-id")
+            .build();
+
+    Storage service1 = options.getService();
+    Storage service2 = options.getService();
+
+    // ensure both instances are the same
+    assertThat(service2).isSameInstanceAs(service1);
+    // make sure and RPCs can be done
+    List<Bucket> buckets1 =
+        StreamSupport.stream(service1.list().iterateAll().spliterator(), false)
+            .collect(Collectors.toList());
+    assertThat(buckets1).isEmpty();
+    List<Bucket> buckets2 =
+        StreamSupport.stream(service2.list().iterateAll().spliterator(), false)
+            .collect(Collectors.toList());
+    assertThat(buckets2).isEmpty();
+
+    service1.close(); // this should be a no-op for http
+
+    // expect the original instance to still be returned
+    Storage service3 = options.getService();
+
+    assertThat(service3).isSameInstanceAs(service1);
+    // make sure and RPCs can be done
+    List<Bucket> buckets3 =
+        StreamSupport.stream(service3.list().iterateAll().spliterator(), false)
+            .collect(Collectors.toList());
+    assertThat(buckets3).isEmpty();
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ResultRetryAlgorithmCompatibilityTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ResultRetryAlgorithmCompatibilityTest.java
index 717500d7..9c907175 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ResultRetryAlgorithmCompatibilityTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ResultRetryAlgorithmCompatibilityTest.java
@@ -18,8 +18,7 @@ package com.google.cloud.storage;
 
 import static com.google.common.truth.Truth.assertThat;
 
-import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
-import com.google.api.gax.rpc.DataLossException;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.rpc.StatusCode.Code;
 import com.google.common.collect.ImmutableSet;
 import java.util.Set;
@@ -27,17 +26,14 @@ import org.junit.Test;
 
 public final class ResultRetryAlgorithmCompatibilityTest {
   @Test
-  public void validateConversion() {
-    BasicResultRetryAlgorithm<Object> alg =
-        new BasicResultRetryAlgorithm<java.lang.Object>() {
-          @Override
-          public boolean shouldRetry(
-              Throwable previousThrowable, java.lang.Object previousResponse) {
-            return previousThrowable instanceof DataLossException;
-          }
-        };
-    Set<Code> codes = GrpcStorageImpl.resultRetryAlgorithmToCodes(alg);
+  public void validateDefaultStorageRetryStrategy_idempotent() {
+    ResultRetryAlgorithm<?> idempotentHandler =
+        StorageRetryStrategy.getDefaultStorageRetryStrategy().getIdempotentHandler();
 
-    assertThat(codes).isEqualTo(ImmutableSet.of(Code.DATA_LOSS));
+    Set<Code> codes = GrpcStorageImpl.resultRetryAlgorithmToCodes(idempotentHandler);
+    ImmutableSet<Code> expected =
+        ImmutableSet.of(
+            Code.INTERNAL, Code.UNAVAILABLE, Code.RESOURCE_EXHAUSTED, Code.DEADLINE_EXCEEDED);
+    assertThat(codes).isEqualTo(expected);
   }
 }
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 6420bd8a..5c527874 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
@@ -26,17 +26,20 @@ import com.google.cloud.PageImpl;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.Restorable;
 import com.google.cloud.storage.Acl.Project.ProjectRole;
+import com.google.cloud.storage.Storage.PredefinedAcl;
+import com.google.cloud.storage.UnifiedOpts.Opt;
 import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import java.io.Serializable;
 import java.util.Collections;
 import java.util.Map;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 public class SerializationTest extends BaseSerializationTest {
 
-  private static final Storage STORAGE =
-      StorageOptions.newBuilder().setProjectId("p").build().getService();
   private static final Acl.Domain ACL_DOMAIN = new Acl.Domain("domain");
   private static final Acl.Group ACL_GROUP = new Acl.Group("group");
   private static final Acl.Project ACL_PROJECT_ = new Acl.Project(ProjectRole.VIEWERS, "pid");
@@ -45,13 +48,9 @@ public class SerializationTest extends BaseSerializationTest {
   private static final Acl ACL = Acl.of(ACL_DOMAIN, Acl.Role.OWNER);
   private static final BlobInfo BLOB_INFO = BlobInfo.newBuilder("b", "n").build();
   private static final BucketInfo BUCKET_INFO = BucketInfo.of("b");
-  private static final Blob BLOB = new Blob(STORAGE, new BlobInfo.BuilderImpl(BLOB_INFO));
-  private static final Bucket BUCKET = new Bucket(STORAGE, new BucketInfo.BuilderImpl(BUCKET_INFO));
   private static final Cors.Origin ORIGIN = Cors.Origin.any();
   private static final Cors CORS =
       Cors.newBuilder().setMaxAgeSeconds(1).setOrigins(Collections.singleton(ORIGIN)).build();
-  private static final PageImpl<Blob> PAGE_RESULT =
-      new PageImpl<>(null, "c", Collections.singletonList(BLOB));
   private static final StorageException STORAGE_EXCEPTION = new StorageException(42, "message");
   private static final Storage.BlobListOption BLOB_LIST_OPTIONS =
       Storage.BlobListOption.pageSize(100);
@@ -67,14 +66,96 @@ public class SerializationTest extends BaseSerializationTest {
       Storage.BucketTargetOption.metagenerationNotMatch();
   private static final Map<StorageRpc.Option, ?> EMPTY_RPC_OPTIONS = ImmutableMap.of();
 
+  private static Storage STORAGE;
+  private static Blob BLOB;
+  private static Bucket BUCKET;
+  private static PageImpl<Blob> PAGE_RESULT;
+
+  @BeforeClass
+  public static void beforeClass() {
+    StorageOptions storageOptions =
+        StorageOptions.newBuilder()
+            .setProjectId("p")
+            .setCredentials(NoCredentials.getInstance())
+            .build();
+    STORAGE = storageOptions.getService();
+    BLOB = BLOB_INFO.asBlob(STORAGE);
+    BUCKET = BUCKET_INFO.asBucket(STORAGE);
+    PAGE_RESULT = new PageImpl<>(null, "c", Collections.singletonList(BLOB));
+  }
+
+  @AfterClass
+  public static void afterClass() throws Exception {
+    if (STORAGE != null) {
+      STORAGE.close();
+    }
+  }
+
   @Override
   protected Serializable[] serializableObjects() {
-    StorageOptions options =
+    StorageOptions optionsDefault1 =
         StorageOptions.newBuilder()
             .setProjectId("p1")
             .setCredentials(NoCredentials.getInstance())
             .build();
-    StorageOptions otherOptions = options.toBuilder().setProjectId("p2").build();
+    StorageOptions optionsDefault2 = optionsDefault1.toBuilder().setProjectId("p2").build();
+    StorageOptions optionsHttp1 =
+        StorageOptions.http()
+            .setProjectId("http1")
+            .setCredentials(NoCredentials.getInstance())
+            .build();
+    StorageOptions optionsHttp2 = optionsHttp1.toBuilder().setProjectId("http2").build();
+    StorageOptions optionsGrpc1 =
+        StorageOptions.http()
+            .setProjectId("grpc1")
+            .setCredentials(NoCredentials.getInstance())
+            .build();
+    StorageOptions optionsGrpc2 = optionsGrpc1.toBuilder().setProjectId("grpc2").build();
+
+    // echo -n "key" | base64
+    String keyBase64 = "a2V5";
+
+    ImmutableList<Opt> serializableOpts =
+        ImmutableList.<Opt>builder()
+            .add(UnifiedOpts.crc32cMatch("crc32c"))
+            .add(UnifiedOpts.currentDirectory())
+            .add(UnifiedOpts.decryptionKey(keyBase64))
+            .add(UnifiedOpts.delimiter("/"))
+            .add(UnifiedOpts.detectContentType())
+            .add(UnifiedOpts.disableGzipContent())
+            .add(UnifiedOpts.doesNotExist())
+            .add(UnifiedOpts.encryptionKey(keyBase64))
+            .add(UnifiedOpts.endOffset("end"))
+            .add(UnifiedOpts.fields(""))
+            .add(UnifiedOpts.generationMatch(0))
+            .add(UnifiedOpts.generationNotMatch(0))
+            .add(UnifiedOpts.kmsKeyName("key"))
+            .add(UnifiedOpts.md5Match("md5"))
+            .add(UnifiedOpts.metagenerationMatch(1))
+            .add(UnifiedOpts.metagenerationNotMatch(1))
+            .add(UnifiedOpts.pageSize(3))
+            .add(UnifiedOpts.pageToken("token"))
+            .add(UnifiedOpts.predefinedAcl(PredefinedAcl.PRIVATE))
+            .add(UnifiedOpts.predefinedDefaultObjectAcl(PredefinedAcl.PRIVATE))
+            .add(UnifiedOpts.prefix("prefix"))
+            .add(UnifiedOpts.projectId("proj"))
+            .add(UnifiedOpts.projection("full"))
+            .add(UnifiedOpts.requestedPolicyVersion(2))
+            .add(UnifiedOpts.returnRawInputStream(false))
+            .add(UnifiedOpts.serviceAccount(ServiceAccount.of("x@y.z")))
+            .add(UnifiedOpts.setContentType("text/plain"))
+            .add(UnifiedOpts.showDeletedKeys(false))
+            .add(UnifiedOpts.startOffset("start"))
+            .add(UnifiedOpts.userProject("user-proj"))
+            .add(UnifiedOpts.versionsFilter(false))
+            .add(UnifiedOpts.generationMatchExtractor())
+            .add(UnifiedOpts.generationNotMatchExtractor())
+            .add(UnifiedOpts.metagenerationMatchExtractor())
+            .add(UnifiedOpts.metagenerationNotMatchExtractor())
+            .add(UnifiedOpts.crc32cMatchExtractor())
+            .add(UnifiedOpts.md5MatchExtractor())
+            .build();
+
     return new Serializable[] {
       ACL_DOMAIN,
       ACL_GROUP,
@@ -96,8 +177,13 @@ public class SerializationTest extends BaseSerializationTest {
       BUCKET_SOURCE_OPTIONS,
       BUCKET_TARGET_OPTIONS,
       STORAGE_EXCEPTION,
-      options,
-      otherOptions
+      optionsDefault1,
+      optionsDefault2,
+      optionsHttp1,
+      optionsHttp2,
+      optionsGrpc1,
+      optionsGrpc2,
+      serializableOpts
     };
   }
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java
new file mode 100644
index 00000000..9ad61533
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java
@@ -0,0 +1,108 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.api.gax.rpc.ApiCallContext;
+import com.google.api.gax.rpc.ApiStreamObserver;
+import com.google.api.gax.rpc.ClientStreamingCallable;
+import com.google.cloud.storage.WriteFlushStrategy.Flusher;
+import com.google.cloud.storage.WriteFlushStrategy.FlusherFactory;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.storage.v2.WriteObjectRequest;
+import com.google.storage.v2.WriteObjectResponse;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import org.junit.Test;
+
+public final class WriteFlushStrategyTest {
+  private static final Map<String, List<String>> expectedHeaderNonNullNonEmpty =
+      ImmutableMap.of("x-goog-request-params", ImmutableList.of("bucket=bucket-name"));
+  private static final Map<String, List<String>> expectedHeaderNonNullEmpty = ImmutableMap.of();
+  private static final Map<String, List<String>> expectedHeaderNull = ImmutableMap.of();
+
+  @Test
+  public void bucketNameAddedToXGoogRequestParams_nonNull_nonEmpty_fsyncEveryFlush() {
+    doTest(WriteFlushStrategy::fsyncEveryFlush, "bucket-name", expectedHeaderNonNullNonEmpty);
+  }
+
+  @Test
+  public void bucketNameAddedToXGoogRequestParams_nonNull_nonEmpty_fsyncOnClose() {
+    doTest(WriteFlushStrategy::fsyncOnClose, "bucket-name", expectedHeaderNonNullNonEmpty);
+  }
+
+  @Test
+  public void bucketNameNotAddedToXGoogRequestParams_nonNull_empty_fsyncEveryFlush() {
+    doTest(WriteFlushStrategy::fsyncEveryFlush, "", expectedHeaderNonNullEmpty);
+  }
+
+  @Test
+  public void bucketNameNotAddedToXGoogRequestParams_nonNull_empty_fsyncOnClose() {
+    doTest(WriteFlushStrategy::fsyncOnClose, "", expectedHeaderNonNullEmpty);
+  }
+
+  @Test
+  public void bucketNameNotAddedToXGoogRequestParams_null_fsyncEveryFlush() {
+    doTest(WriteFlushStrategy::fsyncEveryFlush, null, expectedHeaderNull);
+  }
+
+  @Test
+  public void bucketNameNotAddedToXGoogRequestParams_null_fsyncOnClose() {
+    doTest(WriteFlushStrategy::fsyncOnClose, null, expectedHeaderNull);
+  }
+
+  private static void doTest(
+      Function<ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse>, FlusherFactory> ff,
+      String bucketName,
+      Map<String, List<String>> expectedHeader) {
+    AtomicLong c = new AtomicLong(0);
+    AtomicReference<WriteObjectResponse> ref = new AtomicReference<>();
+    AtomicReference<Map<String, List<String>>> actualHeader = new AtomicReference<>();
+    ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write =
+        new ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse>() {
+          @Override
+          public ApiStreamObserver<WriteObjectRequest> clientStreamingCall(
+              ApiStreamObserver<WriteObjectResponse> responseObserver, ApiCallContext context) {
+            Map<String, List<String>> extraHeaders = context.getExtraHeaders();
+            actualHeader.compareAndSet(null, extraHeaders);
+            return new ApiStreamObserver<WriteObjectRequest>() {
+              @Override
+              public void onNext(WriteObjectRequest value) {}
+
+              @Override
+              public void onError(Throwable t) {}
+
+              @Override
+              public void onCompleted() {
+                responseObserver.onCompleted();
+              }
+            };
+          }
+        };
+    FlusherFactory factory = ff.apply(write);
+    Flusher flusher = factory.newFlusher(bucketName, c::addAndGet, ref::set);
+    flusher.flush(Collections.emptyList());
+    flusher.close(null);
+    assertThat(actualHeader.get()).isEqualTo(expectedHeader);
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITAccessTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITAccessTest.java
index f7f6c234..7a95070a 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITAccessTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITAccessTest.java
@@ -16,6 +16,7 @@
 
 package com.google.cloud.storage.it;
 
+import static com.google.common.truth.Truth.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -30,6 +31,7 @@ import com.google.cloud.storage.Acl;
 import com.google.cloud.storage.Acl.Role;
 import com.google.cloud.storage.Acl.User;
 import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.Bucket;
 import com.google.cloud.storage.BucketFixture;
@@ -770,6 +772,109 @@ public class ITAccessTest {
     }
   }
 
+  @Test
+  @SuppressWarnings({"unchecked", "deprecation"})
+  public void testEnableAndDisableBucketPolicyOnlyOnExistingBucket() throws Exception {
+    String bpoBucket = RemoteStorageHelper.generateBucketName();
+    try {
+      // BPO is disabled by default.
+      Bucket bucket =
+          storage.create(
+              Bucket.newBuilder(bpoBucket)
+                  .setAcl(ImmutableList.of(Acl.of(User.ofAllAuthenticatedUsers(), Role.READER)))
+                  .setDefaultAcl(
+                      ImmutableList.of(Acl.of(User.ofAllAuthenticatedUsers(), Role.READER)))
+                  .build());
+
+      BucketInfo.IamConfiguration bpoEnabledIamConfiguration =
+          BucketInfo.IamConfiguration.newBuilder().setIsBucketPolicyOnlyEnabled(true).build();
+      bucket
+          .toBuilder()
+          .setAcl(null)
+          .setDefaultAcl(null)
+          .setIamConfiguration(bpoEnabledIamConfiguration)
+          .build()
+          .update();
+
+      Bucket remoteBucket =
+          storage.get(bpoBucket, Storage.BucketGetOption.fields(BucketField.IAMCONFIGURATION));
+
+      assertTrue(remoteBucket.getIamConfiguration().isBucketPolicyOnlyEnabled());
+      assertNotNull(remoteBucket.getIamConfiguration().getBucketPolicyOnlyLockedTime());
+
+      remoteBucket
+          .toBuilder()
+          .setIamConfiguration(
+              bpoEnabledIamConfiguration.toBuilder().setIsBucketPolicyOnlyEnabled(false).build())
+          .build()
+          .update();
+
+      remoteBucket =
+          storage.get(
+              bpoBucket,
+              Storage.BucketGetOption.fields(
+                  BucketField.IAMCONFIGURATION, BucketField.ACL, BucketField.DEFAULT_OBJECT_ACL));
+
+      assertFalse(remoteBucket.getIamConfiguration().isBucketPolicyOnlyEnabled());
+      assertEquals(User.ofAllAuthenticatedUsers(), remoteBucket.getDefaultAcl().get(0).getEntity());
+      assertEquals(Role.READER, remoteBucket.getDefaultAcl().get(0).getRole());
+      assertEquals(User.ofAllAuthenticatedUsers(), remoteBucket.getAcl().get(0).getEntity());
+      assertEquals(Role.READER, remoteBucket.getAcl().get(0).getRole());
+    } finally {
+      RemoteStorageHelper.forceDelete(storage, bpoBucket, 1, TimeUnit.MINUTES);
+    }
+  }
+
+  @Test
+  public void testBlobAcl() {
+    BlobId blobId = BlobId.of(bucketName, "test-blob-acl");
+    BlobInfo blob = BlobInfo.newBuilder(blobId).build();
+    storage.create(blob);
+    assertNull(storage.getAcl(blobId, User.ofAllAuthenticatedUsers()));
+    Acl acl = Acl.of(User.ofAllAuthenticatedUsers(), Role.READER);
+    assertNotNull(storage.createAcl(blobId, acl));
+    Acl updatedAcl = storage.updateAcl(blobId, acl.toBuilder().setRole(Role.OWNER).build());
+    assertEquals(Role.OWNER, updatedAcl.getRole());
+    Set<Acl> acls = new HashSet<>(storage.listAcls(blobId));
+    assertTrue(acls.contains(updatedAcl));
+    assertTrue(storage.deleteAcl(blobId, User.ofAllAuthenticatedUsers()));
+    assertNull(storage.getAcl(blobId, User.ofAllAuthenticatedUsers()));
+    // test non-existing blob
+    BlobId otherBlobId = BlobId.of(bucketName, "test-blob-acl", -1L);
+    try {
+      assertNull(storage.getAcl(otherBlobId, User.ofAllAuthenticatedUsers()));
+      fail("Expected an 'Invalid argument' exception");
+    } catch (StorageException e) {
+      assertThat(e.getMessage()).contains("Invalid argument");
+    }
+
+    try {
+      assertFalse(storage.deleteAcl(otherBlobId, User.ofAllAuthenticatedUsers()));
+      fail("Expected an 'Invalid argument' exception");
+    } catch (StorageException e) {
+      assertThat(e.getMessage()).contains("Invalid argument");
+    }
+
+    try {
+      storage.createAcl(otherBlobId, acl);
+      fail("Expected StorageException");
+    } catch (StorageException ex) {
+      // expected
+    }
+    try {
+      storage.updateAcl(otherBlobId, acl);
+      fail("Expected StorageException");
+    } catch (StorageException ex) {
+      // expected
+    }
+    try {
+      storage.listAcls(otherBlobId);
+      fail("Expected StorageException");
+    } catch (StorageException ex) {
+      // expected
+    }
+  }
+
   private Bucket generatePublicAccessPreventionBucket(String bucketName, boolean enforced) {
     return storage.create(
         Bucket.newBuilder(bucketName)
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java
index 36d6af94..47dbe5e2 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java
@@ -17,16 +17,25 @@
 package com.google.cloud.storage.it;
 
 import static com.google.common.truth.Truth.assertThat;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
 
 import com.google.cloud.ReadChannel;
 import com.google.cloud.WriteChannel;
+import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.BucketFixture;
 import com.google.cloud.storage.DataGeneration;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.StorageFixture;
+import com.google.common.io.BaseEncoding;
 import com.google.common.io.ByteStreams;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -34,7 +43,9 @@ import java.nio.channels.FileChannel;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.nio.file.StandardOpenOption;
+import java.util.Arrays;
 import java.util.Random;
+import java.util.zip.GZIPInputStream;
 import org.junit.Before;
 import org.junit.ClassRule;
 import org.junit.Rule;
@@ -46,6 +57,10 @@ public final class ITBlobReadChannelTest {
 
   private static final int _16MiB = 16 * 1024 * 1024;
   private static final int _256KiB = 256 * 1024;
+  private static final String BLOB_STRING_CONTENT = "Hello Google Cloud Storage!";
+  private static final byte[] COMPRESSED_CONTENT =
+      BaseEncoding.base64()
+          .decode("H4sIAAAAAAAAAPNIzcnJV3DPz0/PSVVwzskvTVEILskvSkxPVQQA/LySchsAAAA=");
 
   @Rule public final TestName testName = new TestName();
 
@@ -141,6 +156,99 @@ public final class ITBlobReadChannelTest {
     }
   }
 
+  @Test
+  public void testReadChannelFail() {
+    String blobName = "test-read-channel-blob-fail";
+    BlobInfo blob = BlobInfo.newBuilder(bucketName, blobName).build();
+    Blob remoteBlob = storage.create(blob);
+    assertNotNull(remoteBlob);
+    try (ReadChannel reader =
+        storage.reader(blob.getBlobId(), Storage.BlobSourceOption.metagenerationMatch(-1L))) {
+      reader.read(ByteBuffer.allocate(42));
+      fail("StorageException was expected");
+    } catch (IOException ex) {
+      // expected
+    }
+    try (ReadChannel reader =
+        storage.reader(blob.getBlobId(), Storage.BlobSourceOption.generationMatch(-1L))) {
+      reader.read(ByteBuffer.allocate(42));
+      fail("StorageException was expected");
+    } catch (IOException ex) {
+      // expected
+    }
+    BlobId blobIdWrongGeneration = BlobId.of(bucketName, blobName, -1L);
+    try (ReadChannel reader =
+        storage.reader(blobIdWrongGeneration, Storage.BlobSourceOption.generationMatch())) {
+      reader.read(ByteBuffer.allocate(42));
+      fail("StorageException was expected");
+    } catch (IOException ex) {
+      // expected
+    }
+  }
+
+  @Test
+  public void testReadChannelFailUpdatedGeneration() throws IOException {
+    String blobName = "test-read-blob-fail-updated-generation";
+    BlobInfo blob = BlobInfo.newBuilder(bucketName, blobName).build();
+    Random random = new Random();
+    int chunkSize = 1024;
+    int blobSize = 2 * chunkSize;
+    byte[] content = new byte[blobSize];
+    random.nextBytes(content);
+    Blob remoteBlob = storage.create(blob, content);
+    assertNotNull(remoteBlob);
+    assertEquals(blobSize, (long) remoteBlob.getSize());
+    try (ReadChannel reader = storage.reader(blob.getBlobId())) {
+      reader.setChunkSize(chunkSize);
+      ByteBuffer readBytes = ByteBuffer.allocate(chunkSize);
+      int numReadBytes = reader.read(readBytes);
+      assertEquals(chunkSize, numReadBytes);
+      assertArrayEquals(Arrays.copyOf(content, chunkSize), readBytes.array());
+      try (WriteChannel writer = storage.writer(blob)) {
+        byte[] newContent = new byte[blobSize];
+        random.nextBytes(newContent);
+        int numWrittenBytes = writer.write(ByteBuffer.wrap(newContent));
+        assertEquals(blobSize, numWrittenBytes);
+      }
+      readBytes = ByteBuffer.allocate(chunkSize);
+      reader.read(readBytes);
+      fail("StorageException was expected");
+    } catch (IOException ex) {
+      StringBuilder messageBuilder = new StringBuilder();
+      messageBuilder.append("Blob ").append(blob.getBlobId()).append(" was updated while reading");
+      assertEquals(messageBuilder.toString(), ex.getMessage());
+    }
+  }
+
+  @Test
+  public void testReadCompressedBlob() throws IOException {
+    String blobName = "test-read-compressed-blob";
+    BlobInfo blobInfo =
+        BlobInfo.newBuilder(BlobId.of(bucketName, blobName))
+            .setContentType("text/plain")
+            .setContentEncoding("gzip")
+            .build();
+    Blob blob = storage.create(blobInfo, COMPRESSED_CONTENT);
+    try (ByteArrayOutputStream output = new ByteArrayOutputStream()) {
+      try (ReadChannel reader = storage.reader(BlobId.of(bucketName, blobName))) {
+        reader.setChunkSize(8);
+        ByteBuffer buffer = ByteBuffer.allocate(8);
+        while (reader.read(buffer) != -1) {
+          buffer.flip();
+          output.write(buffer.array(), 0, buffer.limit());
+          buffer.clear();
+        }
+      }
+      assertArrayEquals(
+          BLOB_STRING_CONTENT.getBytes(UTF_8), storage.readAllBytes(bucketName, blobName));
+      assertArrayEquals(COMPRESSED_CONTENT, output.toByteArray());
+      try (GZIPInputStream zipInput =
+          new GZIPInputStream(new ByteArrayInputStream(output.toByteArray()))) {
+        assertArrayEquals(BLOB_STRING_CONTENT.getBytes(UTF_8), ByteStreams.toByteArray(zipInput));
+      }
+    }
+  }
+
   private void doLimitTest(int srcContentSize, int rangeBegin, int rangeEnd, int chunkSize)
       throws IOException {
     BlobInfo src = BlobInfo.newBuilder(bucketName, blobName).build();
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteChannelTest.java
index 936fe800..d8f1c58c 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteChannelTest.java
@@ -17,6 +17,8 @@
 package com.google.cloud.storage.it;
 
 import static com.google.common.truth.Truth.assertThat;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertTrue;
@@ -67,6 +69,7 @@ import org.threeten.bp.format.DateTimeFormatter;
 public final class ITBlobWriteChannelTest {
   private static final Logger LOGGER = Logger.getLogger(ITBlobWriteChannelTest.class.getName());
   private static final String NOW_STRING;
+  private static final String BLOB_STRING_CONTENT = "Hello Google Cloud Storage!";
 
   static {
     Instant now = Clock.systemUTC().instant();
@@ -125,6 +128,34 @@ public final class ITBlobWriteChannelTest {
     blobWriteChannel_handlesRecoveryOnLastChunkWhenGenerationIsPresent(_4MiB, contentSize);
   }
 
+  @Test
+  public void testWriteChannelExistingBlob() throws IOException {
+    HttpStorageOptions baseStorageOptions =
+        StorageOptions.http()
+            .setCredentials(NoCredentials.getInstance())
+            .setHost(testBench.getBaseUri())
+            .setProjectId("test-project-id")
+            .build();
+    Storage storage = baseStorageOptions.getService();
+    Instant now = Clock.systemUTC().instant();
+    DateTimeFormatter formatter =
+        DateTimeFormatter.ISO_LOCAL_DATE_TIME.withZone(ZoneId.from(ZoneOffset.UTC));
+    String nowString = formatter.format(now);
+    BucketInfo bucketInfo = BucketInfo.of(dataGeneration.getBucketName());
+    String blobPath = String.format("%s/%s/blob", testName.getMethodName(), nowString);
+    BlobId blobId = BlobId.of(dataGeneration.getBucketName(), blobPath);
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();
+    storage.create(bucketInfo);
+    storage.create(blobInfo);
+    byte[] stringBytes;
+    try (WriteChannel writer = storage.writer(blobInfo)) {
+      stringBytes = BLOB_STRING_CONTENT.getBytes(UTF_8);
+      writer.write(ByteBuffer.wrap(stringBytes));
+    }
+    assertArrayEquals(stringBytes, storage.readAllBytes(blobInfo.getBlobId()));
+    assertTrue(storage.delete(bucketInfo.getName(), blobInfo.getName()));
+  }
+
   private void doJsonUnexpectedEOFTest(int contentSize, int cappedByteCount) throws IOException {
     String blobPath = String.format("%s/%s/blob", testName.getMethodName(), NOW_STRING);
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITDownloadBlobWithoutAuth.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITDownloadBlobWithoutAuth.java
new file mode 100644
index 00000000..bf3d76e6
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITDownloadBlobWithoutAuth.java
@@ -0,0 +1,108 @@
+/*
+ * 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 static org.junit.Assume.assumeFalse;
+
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketFixture;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageFixture;
+import com.google.cloud.storage.StorageOptions;
+import java.util.Iterator;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+public class ITDownloadBlobWithoutAuth {
+  @ClassRule(order = 1)
+  public static final StorageFixture storageFixture = StorageFixture.defaultHttp();
+
+  @ClassRule(order = 2)
+  public static final BucketFixture bucketFixture =
+      BucketFixture.newBuilder().setHandle(storageFixture::getInstance).build();
+
+  private static final boolean IS_VPC_TEST =
+      System.getenv("GOOGLE_CLOUD_TESTS_IN_VPCSC") != null
+          && System.getenv("GOOGLE_CLOUD_TESTS_IN_VPCSC").equalsIgnoreCase("true");
+
+  private static Storage storage;
+  private static String bucketName;
+
+  @BeforeClass
+  public static void setup() {
+    storage = storageFixture.getInstance();
+    bucketName = bucketFixture.getBucketInfo().getName();
+  }
+
+  @Test
+  public void testDownloadPublicBlobWithoutAuthentication() {
+    assumeFalse(IS_VPC_TEST);
+    // create an unauthorized user
+    Storage unauthorizedStorage = StorageOptions.getUnauthenticatedInstance().getService();
+
+    // try to download blobs from a public bucket
+    String landsatBucket = "gcp-public-data-landsat";
+    String landsatPrefix = "LC08/01/001/002/LC08_L1GT_001002_20160817_20170322_01_T2/";
+    String landsatBlob = landsatPrefix + "LC08_L1GT_001002_20160817_20170322_01_T2_ANG.txt";
+    byte[] bytes = unauthorizedStorage.readAllBytes(landsatBucket, landsatBlob);
+
+    assertThat(bytes.length).isEqualTo(117255);
+    int numBlobs = 0;
+    Iterator<Blob> blobIterator =
+        unauthorizedStorage
+            .list(landsatBucket, Storage.BlobListOption.prefix(landsatPrefix))
+            .iterateAll()
+            .iterator();
+    while (blobIterator.hasNext()) {
+      numBlobs++;
+      blobIterator.next();
+    }
+    assertThat(numBlobs).isEqualTo(14);
+
+    // try to download blobs from a bucket that requires authentication
+    // authenticated client will succeed
+    // unauthenticated client will receive an exception
+    String sourceBlobName = "source-blob-name";
+    BlobInfo sourceBlob = BlobInfo.newBuilder(bucketName, sourceBlobName).build();
+    assertThat(storage.create(sourceBlob)).isNotNull();
+    assertThat(storage.readAllBytes(bucketName, sourceBlobName)).isNotNull();
+    try {
+      unauthorizedStorage.readAllBytes(bucketName, sourceBlobName);
+      fail("Expected StorageException");
+    } catch (StorageException ex) {
+      // expected
+    }
+    assertThat(storage.get(sourceBlob.getBlobId()).delete()).isTrue();
+
+    // try to upload blobs to a bucket that requires authentication
+    // authenticated client will succeed
+    // unauthenticated client will receive an exception
+    assertThat(storage.create(sourceBlob)).isNotNull();
+    try {
+      unauthorizedStorage.create(sourceBlob);
+      fail("Expected StorageException");
+    } catch (StorageException ex) {
+      // expected
+    }
+    assertThat(storage.get(sourceBlob.getBlobId()).delete()).isTrue();
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
index cca6645d..53be1a63 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
@@ -20,7 +20,9 @@ import static com.google.common.truth.Truth.assertThat;
 
 import com.google.api.gax.paging.Page;
 import com.google.cloud.NoCredentials;
+import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.Bucket;
 import com.google.cloud.storage.BucketFixture;
 import com.google.cloud.storage.BucketInfo;
@@ -28,6 +30,10 @@ import com.google.cloud.storage.HmacKey;
 import com.google.cloud.storage.HmacKey.HmacKeyMetadata;
 import com.google.cloud.storage.HmacKey.HmacKeyState;
 import com.google.cloud.storage.ServiceAccount;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.Storage.CreateHmacKeyOption;
 import com.google.cloud.storage.Storage.ListHmacKeysOption;
 import com.google.cloud.storage.StorageFixture;
@@ -36,10 +42,17 @@ import com.google.cloud.storage.conformance.retry.CleanupStrategy;
 import com.google.cloud.storage.conformance.retry.TestBench;
 import com.google.cloud.storage.testing.RemoteStorageHelper;
 import com.google.common.collect.ImmutableList;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
 import java.util.stream.IntStream;
 import java.util.stream.StreamSupport;
 import org.junit.ClassRule;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TestName;
 
 public final class ITGrpcTest {
   @ClassRule(order = 1)
@@ -64,6 +77,8 @@ public final class ITGrpcTest {
           .setHandle(storageFixture::getInstance)
           .build();
 
+  @Rule public final TestName testName = new TestName();
+
   @Test
   public void testCreateBucket() {
     final String bucketName = RemoteStorageHelper.generateBucketName();
@@ -74,13 +89,30 @@ public final class ITGrpcTest {
   @Test
   public void listBlobs() {
     BucketInfo bucketInfo = bucketFixture.getBucketInfo();
-    Page<Blob> list = storageFixture.getInstance().list(bucketInfo.getName());
-    ImmutableList<String> bucketNames =
+    byte[] content = "Hello, World!".getBytes(StandardCharsets.UTF_8);
+    String prefix = testName.getMethodName();
+    List<Blob> blobs =
+        IntStream.rangeClosed(1, 10)
+            .mapToObj(i -> String.format("%s/%02d", prefix, i))
+            .map(n -> BlobInfo.newBuilder(bucketInfo, n).build())
+            .map(
+                info ->
+                    storageFixture
+                        .getInstance()
+                        .create(info, content, BlobTargetOption.doesNotExist()))
+            .collect(ImmutableList.toImmutableList());
+
+    List<String> expected =
+        blobs.stream().map(Blob::getName).collect(ImmutableList.toImmutableList());
+
+    Page<Blob> list =
+        storageFixture.getInstance().list(bucketInfo.getName(), BlobListOption.prefix(prefix));
+    ImmutableList<String> actual =
         StreamSupport.stream(list.iterateAll().spliterator(), false)
             .map(Blob::getName)
             .collect(ImmutableList.toImmutableList());
 
-    assertThat(bucketNames).isEmpty();
+    assertThat(actual).isEqualTo(expected);
   }
 
   @Test
@@ -156,4 +188,53 @@ public final class ITGrpcTest {
     storageFixture.getInstance().updateHmacKeyState(hmacKey.getMetadata(), HmacKeyState.INACTIVE);
     storageFixture.getInstance().deleteHmacKey(hmacKey.getMetadata());
   }
+
+  @Test
+  public void object_writeGetRead() {
+    Storage s = storageFixture.getInstance();
+    BlobInfo info = BlobInfo.newBuilder(bucketFixture.getBucketInfo(), "writeGetRead").build();
+    byte[] content = "hello, world".getBytes(StandardCharsets.UTF_8);
+    s.create(info, content, BlobTargetOption.doesNotExist());
+
+    Blob blob = s.get(info.getBlobId());
+
+    byte[] actualContent = blob.getContent();
+    assertThat(actualContent).isEqualTo(content);
+  }
+
+  @Test
+  public void objectWrite_storage_create() {
+    BucketInfo bucketInfo = bucketFixture.getBucketInfo();
+    BlobInfo info = BlobInfo.newBuilder(bucketInfo, testName.getMethodName()).build();
+    byte[] content = "Hello, World!".getBytes(StandardCharsets.UTF_8);
+    Blob blob = storageFixture.getInstance().create(info, content, BlobTargetOption.doesNotExist());
+    byte[] actual = blob.getContent();
+    assertThat(actual).isEqualTo(content);
+  }
+
+  @Test
+  public void objectWrite_storage_create_stream() {
+    BucketInfo bucketInfo = bucketFixture.getBucketInfo();
+    BlobInfo info = BlobInfo.newBuilder(bucketInfo, testName.getMethodName()).build();
+    byte[] content = "Hello, World!".getBytes(StandardCharsets.UTF_8);
+    Blob blob =
+        storageFixture
+            .getInstance()
+            .create(info, new ByteArrayInputStream(content), BlobWriteOption.doesNotExist());
+    byte[] actual = blob.getContent();
+    assertThat(actual).isEqualTo(content);
+  }
+
+  @Test
+  public void objectWrite_storage_writer() throws IOException {
+    BucketInfo bucketInfo = bucketFixture.getBucketInfo();
+    BlobInfo info = BlobInfo.newBuilder(bucketInfo, testName.getMethodName()).build();
+    byte[] content = "Hello, World!".getBytes(StandardCharsets.UTF_8);
+    try (WriteChannel c =
+        storageFixture.getInstance().writer(info, BlobWriteOption.doesNotExist())) {
+      c.write(ByteBuffer.wrap(content));
+    }
+    byte[] actual = storageFixture.getInstance().readAllBytes(info.getBlobId());
+    assertThat(actual).isEqualTo(content);
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITHmacTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITHmacTest.java
new file mode 100644
index 00000000..1fd5d99c
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITHmacTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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.assertWithMessage;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import com.google.api.gax.paging.Page;
+import com.google.cloud.storage.HmacKey;
+import com.google.cloud.storage.HmacKey.HmacKeyMetadata;
+import com.google.cloud.storage.HmacKey.HmacKeyState;
+import com.google.cloud.storage.ServiceAccount;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageFixture;
+import java.util.stream.StreamSupport;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.threeten.bp.Duration;
+import org.threeten.bp.Instant;
+
+public class ITHmacTest {
+  @ClassRule public static final StorageFixture storageFixture = StorageFixture.defaultHttp();
+
+  private static Storage storage;
+
+  @BeforeClass
+  public static void setup() {
+    storage = storageFixture.getInstance();
+  }
+
+  // when modifying this test or {@link #cleanUpHmacKeys} be sure to remember multiple simultaneous
+  // runs of the integration suite can run with the same service account. Be sure to not clobber
+  // any possible run state for the other run.
+  @Test
+  public void testHmacKey() {
+    String serviceAccountEmail = System.getenv("IT_SERVICE_ACCOUNT_EMAIL");
+    assertNotNull("Unable to determine service account email", serviceAccountEmail);
+    ServiceAccount serviceAccount = ServiceAccount.of(serviceAccountEmail);
+    cleanUpHmacKeys(serviceAccount);
+
+    HmacKey hmacKey = storage.createHmacKey(serviceAccount);
+    String secretKey = hmacKey.getSecretKey();
+    assertNotNull(secretKey);
+    HmacKey.HmacKeyMetadata metadata = hmacKey.getMetadata();
+    String accessId = metadata.getAccessId();
+
+    assertNotNull(accessId);
+    assertNotNull(metadata.getEtag());
+    assertNotNull(metadata.getId());
+    assertEquals(storage.getOptions().getProjectId(), metadata.getProjectId());
+    assertEquals(serviceAccount.getEmail(), metadata.getServiceAccount().getEmail());
+    assertEquals(HmacKey.HmacKeyState.ACTIVE, metadata.getState());
+    assertNotNull(metadata.getCreateTime());
+    assertNotNull(metadata.getUpdateTime());
+
+    Page<HmacKeyMetadata> metadatas =
+        storage.listHmacKeys(Storage.ListHmacKeysOption.serviceAccount(serviceAccount));
+    boolean createdInList =
+        StreamSupport.stream(metadatas.iterateAll().spliterator(), false)
+            .map(HmacKey.HmacKeyMetadata::getAccessId)
+            .anyMatch(accessId::equals);
+
+    assertWithMessage("Created an HMAC key but it didn't show up in list()")
+        .that(createdInList)
+        .isTrue();
+
+    HmacKey.HmacKeyMetadata getResult = storage.getHmacKey(accessId);
+    assertEquals(metadata, getResult);
+
+    storage.updateHmacKeyState(metadata, HmacKey.HmacKeyState.INACTIVE);
+
+    storage.deleteHmacKey(metadata);
+
+    metadatas = storage.listHmacKeys(Storage.ListHmacKeysOption.serviceAccount(serviceAccount));
+    boolean deletedInList =
+        StreamSupport.stream(metadatas.iterateAll().spliterator(), false)
+            .map(HmacKey.HmacKeyMetadata::getAccessId)
+            .anyMatch(accessId::equals);
+
+    assertWithMessage("Deleted an HMAC key but it showed up in list()")
+        .that(deletedInList)
+        .isFalse();
+  }
+
+  private void cleanUpHmacKeys(ServiceAccount serviceAccount) {
+    Instant now = Instant.now();
+    Instant yesterday = now.minus(Duration.ofDays(1L));
+
+    Page<HmacKey.HmacKeyMetadata> metadatas =
+        storage.listHmacKeys(Storage.ListHmacKeysOption.serviceAccount(serviceAccount));
+    for (HmacKey.HmacKeyMetadata hmacKeyMetadata : metadatas.iterateAll()) {
+      Instant updated = Instant.ofEpochMilli(hmacKeyMetadata.getUpdateTime());
+      if (updated.isBefore(yesterday)) {
+
+        if (hmacKeyMetadata.getState() == HmacKeyState.ACTIVE) {
+          hmacKeyMetadata = storage.updateHmacKeyState(hmacKeyMetadata, HmacKeyState.INACTIVE);
+        }
+
+        if (hmacKeyMetadata.getState() == HmacKeyState.INACTIVE) {
+          try {
+            storage.deleteHmacKey(hmacKeyMetadata);
+          } catch (StorageException e) {
+            // attempted to delete concurrently, if the other succeeded swallow the error
+            if (!(e.getReason().equals("invalid") && e.getMessage().contains("deleted"))) {
+              throw e;
+            }
+          }
+        }
+      }
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITStorageTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
similarity index 74%
rename from google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITStorageTest.java
rename to google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
index f974938c..3249e03a 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITStorageTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
@@ -17,7 +17,6 @@
 package com.google.cloud.storage.it;
 
 import static com.google.common.truth.Truth.assertThat;
-import static com.google.common.truth.Truth.assertWithMessage;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
@@ -27,30 +26,19 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
-import static org.junit.Assume.assumeFalse;
 
-import com.google.api.client.http.HttpTransport;
-import com.google.api.client.http.apache.ApacheHttpTransport;
 import com.google.api.gax.paging.Page;
-import com.google.auth.http.HttpTransportFactory;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.RestorableState;
-import com.google.cloud.TransportOptions;
 import com.google.cloud.WriteChannel;
-import com.google.cloud.http.HttpTransportOptions;
-import com.google.cloud.storage.Acl;
-import com.google.cloud.storage.Acl.Role;
 import com.google.cloud.storage.Acl.User;
 import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketFixture;
 import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.CopyWriter;
-import com.google.cloud.storage.DataGeneration;
-import com.google.cloud.storage.HmacKey;
-import com.google.cloud.storage.HmacKey.HmacKeyState;
-import com.google.cloud.storage.ServiceAccount;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobField;
 import com.google.cloud.storage.Storage.BlobWriteOption;
@@ -58,7 +46,6 @@ import com.google.cloud.storage.Storage.BucketField;
 import com.google.cloud.storage.StorageClass;
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageFixture;
-import com.google.cloud.storage.StorageOptions;
 import com.google.cloud.storage.testing.RemoteStorageHelper;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -66,7 +53,6 @@ import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterators;
 import com.google.common.collect.Lists;
 import com.google.common.io.BaseEncoding;
-import com.google.common.io.ByteStreams;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -75,7 +61,6 @@ import java.nio.ByteBuffer;
 import java.security.Key;
 import java.util.Arrays;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -83,28 +68,13 @@ import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-import java.util.stream.StreamSupport;
-import java.util.zip.GZIPInputStream;
 import javax.crypto.spec.SecretKeySpec;
-import org.apache.http.impl.client.HttpClients;
-import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
-import org.junit.Rule;
 import org.junit.Test;
-import org.junit.rules.TestName;
-import org.threeten.bp.Duration;
-import org.threeten.bp.Instant;
 
-public class ITStorageTest {
-
-  private static Storage storage;
-  private static final Logger log = Logger.getLogger(ITStorageTest.class.getName());
-  private static final String BUCKET = RemoteStorageHelper.generateBucketName();
-  private static final String BUCKET_REQUESTER_PAYS = RemoteStorageHelper.generateBucketName();
+public class ITObjectTest {
   private static final String CONTENT_TYPE = "text/plain";
   private static final byte[] BLOB_BYTE_CONTENT = {0xD, 0xE, 0xA, 0xD};
   private static final String BLOB_STRING_CONTENT = "Hello Google Cloud Storage!";
@@ -112,38 +82,31 @@ public class ITStorageTest {
   private static final String OTHER_BASE64_KEY = "IcOIQGlliNr5pr3vJb63l+XMqc7NjXqjfw/deBoNxPA=";
   private static final Key KEY =
       new SecretKeySpec(BaseEncoding.base64().decode(BASE64_KEY), "AES256");
-  private static final byte[] COMPRESSED_CONTENT =
-      BaseEncoding.base64()
-          .decode("H4sIAAAAAAAAAPNIzcnJV3DPz0/PSVVwzskvTVEILskvSkxPVQQA/LySchsAAAA=");
-  private static final Map<String, String> REMOVE_BUCKET_LABELS;
-
-  static {
-    REMOVE_BUCKET_LABELS = new HashMap<>();
-    REMOVE_BUCKET_LABELS.put("label1", null);
-  }
 
   private static final Long RETENTION_PERIOD = 5L;
   private static final Long RETENTION_PERIOD_IN_MILLISECONDS = RETENTION_PERIOD * 1000;
-  private static final String SERVICE_ACCOUNT_EMAIL_SUFFIX =
-      "@gs-project-accounts.iam.gserviceaccount.com";
-  private static final boolean IS_VPC_TEST =
-      System.getenv("GOOGLE_CLOUD_TESTS_IN_VPCSC") != null
-          && System.getenv("GOOGLE_CLOUD_TESTS_IN_VPCSC").equalsIgnoreCase("true");
 
-  @ClassRule
+  @ClassRule(order = 1)
   public static final StorageFixture storageFixture =
       StorageFixture.of(() -> RemoteStorageHelper.create().getOptions().getService());
 
-  @Rule public final TestName testName = new TestName();
-  @Rule public final DataGeneration dataGeneration = new DataGeneration(new Random(1234567890));
+  @ClassRule(order = 2)
+  public static final BucketFixture bucketFixture =
+      BucketFixture.newBuilder().setHandle(storageFixture::getInstance).build();
+
+  @ClassRule(order = 3)
+  public static final BucketFixture requesterPaysFixture =
+      BucketFixture.newBuilder().setHandle(storageFixture::getInstance).build();
+
+  private static String BUCKET;
+  private static String BUCKET_REQUESTER_PAYS;
+  private static Storage storage;
 
   @BeforeClass
   public static void beforeClass() throws IOException {
     storage = storageFixture.getInstance();
-
-    storage.create(BucketInfo.newBuilder(BUCKET).setLocation("us").build());
-
-    storage.create(BucketInfo.newBuilder(BUCKET_REQUESTER_PAYS).build());
+    BUCKET = bucketFixture.getBucketInfo().getName();
+    BUCKET_REQUESTER_PAYS = requesterPaysFixture.getBucketInfo().getName();
   }
 
   private static void unsetRequesterPays() {
@@ -163,32 +126,8 @@ public class ITStorageTest {
   }
 
   @AfterClass
-  public static void afterClass() throws ExecutionException, InterruptedException {
-
-    if (storage != null) {
-      // In beforeClass, we make buckets auto-delete blobs older than a day old.
-      // Here, delete all buckets older than 2 days. They should already be empty and easy.
-      long cleanTime = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(2);
-      long cleanTimeout = System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(1);
-      RemoteStorageHelper.cleanBuckets(storage, cleanTime, cleanTimeout);
-
-      boolean wasDeleted = RemoteStorageHelper.forceDelete(storage, BUCKET, 1, TimeUnit.MINUTES);
-      if (!wasDeleted && log.isLoggable(Level.WARNING)) {
-        log.log(Level.WARNING, "Deletion of bucket {0} timed out, bucket is not empty", BUCKET);
-      }
-      unsetRequesterPays();
-      RemoteStorageHelper.forceDelete(storage, BUCKET_REQUESTER_PAYS, 5, TimeUnit.SECONDS);
-    }
-  }
-
-  private static class CustomHttpTransportFactory implements HttpTransportFactory {
-    @Override
-    @SuppressWarnings({"unchecked", "deprecation"})
-    public HttpTransport create() {
-      PoolingHttpClientConnectionManager manager = new PoolingHttpClientConnectionManager();
-      manager.setMaxTotal(1);
-      return new ApacheHttpTransport(HttpClients.createMinimal(manager));
-    }
+  public static void afterClass() {
+    unsetRequesterPays();
   }
 
   @Test
@@ -1069,28 +1008,6 @@ public class ITStorageTest {
     }
   }
 
-  @Test
-  public void testReadAndWriteChannels() throws IOException {
-    String blobName = "test-read-and-write-channels-blob";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
-    byte[] stringBytes;
-    try (WriteChannel writer = storage.writer(blob)) {
-      stringBytes = BLOB_STRING_CONTENT.getBytes(UTF_8);
-      writer.write(ByteBuffer.wrap(BLOB_BYTE_CONTENT));
-      writer.write(ByteBuffer.wrap(stringBytes));
-    }
-    ByteBuffer readBytes;
-    ByteBuffer readStringBytes;
-    try (ReadChannel reader = storage.reader(blob.getBlobId())) {
-      readBytes = ByteBuffer.allocate(BLOB_BYTE_CONTENT.length);
-      readStringBytes = ByteBuffer.allocate(stringBytes.length);
-      reader.read(readBytes);
-      reader.read(readStringBytes);
-    }
-    assertArrayEquals(BLOB_BYTE_CONTENT, readBytes.array());
-    assertEquals(BLOB_STRING_CONTENT, new String(readStringBytes.array(), UTF_8));
-  }
-
   @Test
   public void testReadAndWriteChannelWithEncryptionKey() throws IOException {
     String blobName = "test-read-write-channel-with-customer-key-blob";
@@ -1172,122 +1089,6 @@ public class ITStorageTest {
     assertTrue(storage.delete(BUCKET, blobName));
   }
 
-  @Test
-  public void testReadChannelFail() throws IOException {
-    String blobName = "test-read-channel-blob-fail";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
-    Blob remoteBlob = storage.create(blob);
-    assertNotNull(remoteBlob);
-    try (ReadChannel reader =
-        storage.reader(blob.getBlobId(), Storage.BlobSourceOption.metagenerationMatch(-1L))) {
-      reader.read(ByteBuffer.allocate(42));
-      fail("StorageException was expected");
-    } catch (IOException ex) {
-      // expected
-    }
-    try (ReadChannel reader =
-        storage.reader(blob.getBlobId(), Storage.BlobSourceOption.generationMatch(-1L))) {
-      reader.read(ByteBuffer.allocate(42));
-      fail("StorageException was expected");
-    } catch (IOException ex) {
-      // expected
-    }
-    BlobId blobIdWrongGeneration = BlobId.of(BUCKET, blobName, -1L);
-    try (ReadChannel reader =
-        storage.reader(blobIdWrongGeneration, Storage.BlobSourceOption.generationMatch())) {
-      reader.read(ByteBuffer.allocate(42));
-      fail("StorageException was expected");
-    } catch (IOException ex) {
-      // expected
-    }
-  }
-
-  @Test
-  public void testReadChannelFailUpdatedGeneration() throws IOException {
-    String blobName = "test-read-blob-fail-updated-generation";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
-    Random random = new Random();
-    int chunkSize = 1024;
-    int blobSize = 2 * chunkSize;
-    byte[] content = new byte[blobSize];
-    random.nextBytes(content);
-    Blob remoteBlob = storage.create(blob, content);
-    assertNotNull(remoteBlob);
-    assertEquals(blobSize, (long) remoteBlob.getSize());
-    try (ReadChannel reader = storage.reader(blob.getBlobId())) {
-      reader.setChunkSize(chunkSize);
-      ByteBuffer readBytes = ByteBuffer.allocate(chunkSize);
-      int numReadBytes = reader.read(readBytes);
-      assertEquals(chunkSize, numReadBytes);
-      assertArrayEquals(Arrays.copyOf(content, chunkSize), readBytes.array());
-      try (WriteChannel writer = storage.writer(blob)) {
-        byte[] newContent = new byte[blobSize];
-        random.nextBytes(newContent);
-        int numWrittenBytes = writer.write(ByteBuffer.wrap(newContent));
-        assertEquals(blobSize, numWrittenBytes);
-      }
-      readBytes = ByteBuffer.allocate(chunkSize);
-      reader.read(readBytes);
-      fail("StorageException was expected");
-    } catch (IOException ex) {
-      StringBuilder messageBuilder = new StringBuilder();
-      messageBuilder.append("Blob ").append(blob.getBlobId()).append(" was updated while reading");
-      assertEquals(messageBuilder.toString(), ex.getMessage());
-    }
-    assertTrue(storage.delete(BUCKET, blobName));
-  }
-
-  @Test
-  public void testWriteChannelFail() throws IOException {
-    String blobName = "test-write-channel-blob-fail";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName, -1L).build();
-    try {
-      try (WriteChannel writer = storage.writer(blob, Storage.BlobWriteOption.generationMatch())) {
-        writer.write(ByteBuffer.allocate(42));
-      }
-      fail("StorageException was expected");
-    } catch (StorageException ex) {
-      // expected
-    }
-  }
-
-  @Test
-  public void testWriteChannelExistingBlob() throws IOException {
-    String blobName = "test-write-channel-existing-blob";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
-    storage.create(blob);
-    byte[] stringBytes;
-    try (WriteChannel writer = storage.writer(blob)) {
-      stringBytes = BLOB_STRING_CONTENT.getBytes(UTF_8);
-      writer.write(ByteBuffer.wrap(stringBytes));
-    }
-    assertArrayEquals(stringBytes, storage.readAllBytes(blob.getBlobId()));
-    assertTrue(storage.delete(BUCKET, blobName));
-  }
-
-  @Test(timeout = 5000)
-  public void testWriteChannelWithConnectionPool() throws IOException {
-    TransportOptions transportOptions =
-        HttpTransportOptions.newBuilder()
-            .setHttpTransportFactory(new CustomHttpTransportFactory())
-            .build();
-    Storage storageWithPool =
-        StorageOptions.http().setTransportOptions(transportOptions).build().getService();
-    String blobName = "test-custom-pool-management";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
-    byte[] stringBytes;
-    try (WriteChannel writer = storageWithPool.writer(blob)) {
-      stringBytes = BLOB_STRING_CONTENT.getBytes(UTF_8);
-      writer.write(ByteBuffer.wrap(BLOB_BYTE_CONTENT));
-      writer.write(ByteBuffer.wrap(stringBytes));
-    }
-    try (WriteChannel writer = storageWithPool.writer(blob)) {
-      stringBytes = BLOB_STRING_CONTENT.getBytes(UTF_8);
-      writer.write(ByteBuffer.wrap(BLOB_BYTE_CONTENT));
-      writer.write(ByteBuffer.wrap(stringBytes));
-    }
-  }
-
   @Test
   public void testGetBlobs() {
     String sourceBlobName1 = "test-get-blobs-1";
@@ -1303,59 +1104,6 @@ public class ITStorageTest {
     assertEquals(sourceBlob2.getName(), remoteBlobs.get(1).getName());
   }
 
-  @Test
-  public void testDownloadPublicBlobWithoutAuthentication() {
-    assumeFalse(IS_VPC_TEST);
-    // create an unauthorized user
-    Storage unauthorizedStorage = StorageOptions.getUnauthenticatedInstance().getService();
-
-    // try to download blobs from a public bucket
-    String landsatBucket = "gcp-public-data-landsat";
-    String landsatPrefix = "LC08/01/001/002/LC08_L1GT_001002_20160817_20170322_01_T2/";
-    String landsatBlob = landsatPrefix + "LC08_L1GT_001002_20160817_20170322_01_T2_ANG.txt";
-    byte[] bytes = unauthorizedStorage.readAllBytes(landsatBucket, landsatBlob);
-
-    assertThat(bytes.length).isEqualTo(117255);
-    int numBlobs = 0;
-    Iterator<Blob> blobIterator =
-        unauthorizedStorage
-            .list(landsatBucket, Storage.BlobListOption.prefix(landsatPrefix))
-            .iterateAll()
-            .iterator();
-    while (blobIterator.hasNext()) {
-      numBlobs++;
-      blobIterator.next();
-    }
-    assertThat(numBlobs).isEqualTo(14);
-
-    // try to download blobs from a bucket that requires authentication
-    // authenticated client will succeed
-    // unauthenticated client will receive an exception
-    String sourceBlobName = "source-blob-name";
-    BlobInfo sourceBlob = BlobInfo.newBuilder(BUCKET, sourceBlobName).build();
-    assertThat(storage.create(sourceBlob)).isNotNull();
-    assertThat(storage.readAllBytes(BUCKET, sourceBlobName)).isNotNull();
-    try {
-      unauthorizedStorage.readAllBytes(BUCKET, sourceBlobName);
-      fail("Expected StorageException");
-    } catch (StorageException ex) {
-      // expected
-    }
-    assertThat(storage.get(sourceBlob.getBlobId()).delete()).isTrue();
-
-    // try to upload blobs to a bucket that requires authentication
-    // authenticated client will succeed
-    // unauthenticated client will receive an exception
-    assertThat(storage.create(sourceBlob)).isNotNull();
-    try {
-      unauthorizedStorage.create(sourceBlob);
-      fail("Expected StorageException");
-    } catch (StorageException ex) {
-      // expected
-    }
-    assertThat(storage.get(sourceBlob.getBlobId()).delete()).isTrue();
-  }
-
   @Test
   public void testGetBlobsFail() {
     String sourceBlobName1 = "test-get-blobs-fail-1";
@@ -1434,167 +1182,6 @@ public class ITStorageTest {
     assertNull(updatedBlobs.get(1));
   }
 
-  @Test
-  public void testBlobAcl() {
-    BlobId blobId = BlobId.of(BUCKET, "test-blob-acl");
-    BlobInfo blob = BlobInfo.newBuilder(blobId).build();
-    storage.create(blob);
-    assertNull(storage.getAcl(blobId, User.ofAllAuthenticatedUsers()));
-    Acl acl = Acl.of(User.ofAllAuthenticatedUsers(), Role.READER);
-    assertNotNull(storage.createAcl(blobId, acl));
-    Acl updatedAcl = storage.updateAcl(blobId, acl.toBuilder().setRole(Role.OWNER).build());
-    assertEquals(Role.OWNER, updatedAcl.getRole());
-    Set<Acl> acls = new HashSet<>(storage.listAcls(blobId));
-    assertTrue(acls.contains(updatedAcl));
-    assertTrue(storage.deleteAcl(blobId, User.ofAllAuthenticatedUsers()));
-    assertNull(storage.getAcl(blobId, User.ofAllAuthenticatedUsers()));
-    // test non-existing blob
-    BlobId otherBlobId = BlobId.of(BUCKET, "test-blob-acl", -1L);
-    try {
-      assertNull(storage.getAcl(otherBlobId, User.ofAllAuthenticatedUsers()));
-      fail("Expected an 'Invalid argument' exception");
-    } catch (StorageException e) {
-      assertThat(e.getMessage()).contains("Invalid argument");
-    }
-
-    try {
-      assertFalse(storage.deleteAcl(otherBlobId, User.ofAllAuthenticatedUsers()));
-      fail("Expected an 'Invalid argument' exception");
-    } catch (StorageException e) {
-      assertThat(e.getMessage()).contains("Invalid argument");
-    }
-
-    try {
-      storage.createAcl(otherBlobId, acl);
-      fail("Expected StorageException");
-    } catch (StorageException ex) {
-      // expected
-    }
-    try {
-      storage.updateAcl(otherBlobId, acl);
-      fail("Expected StorageException");
-    } catch (StorageException ex) {
-      // expected
-    }
-    try {
-      storage.listAcls(otherBlobId);
-      fail("Expected StorageException");
-    } catch (StorageException ex) {
-      // expected
-    }
-  }
-
-  // when modifying this test or {@link #cleanUpHmacKeys} be sure to remember multiple simultaneous
-  // runs of the integration suite can run with the same service account. Be sure to not clobber
-  // any possible run state for the other run.
-  @Test
-  public void testHmacKey() {
-    String serviceAccountEmail = System.getenv("IT_SERVICE_ACCOUNT_EMAIL");
-    assertNotNull("Unable to determine service account email", serviceAccountEmail);
-    ServiceAccount serviceAccount = ServiceAccount.of(serviceAccountEmail);
-    cleanUpHmacKeys(serviceAccount);
-
-    HmacKey hmacKey = storage.createHmacKey(serviceAccount);
-    String secretKey = hmacKey.getSecretKey();
-    assertNotNull(secretKey);
-    HmacKey.HmacKeyMetadata metadata = hmacKey.getMetadata();
-    String accessId = metadata.getAccessId();
-
-    assertNotNull(accessId);
-    assertNotNull(metadata.getEtag());
-    assertNotNull(metadata.getId());
-    assertEquals(storage.getOptions().getProjectId(), metadata.getProjectId());
-    assertEquals(serviceAccount.getEmail(), metadata.getServiceAccount().getEmail());
-    assertEquals(HmacKey.HmacKeyState.ACTIVE, metadata.getState());
-    assertNotNull(metadata.getCreateTime());
-    assertNotNull(metadata.getUpdateTime());
-
-    Page<HmacKey.HmacKeyMetadata> metadatas =
-        storage.listHmacKeys(Storage.ListHmacKeysOption.serviceAccount(serviceAccount));
-    boolean createdInList =
-        StreamSupport.stream(metadatas.iterateAll().spliterator(), false)
-            .map(HmacKey.HmacKeyMetadata::getAccessId)
-            .anyMatch(accessId::equals);
-
-    assertWithMessage("Created an HMAC key but it didn't show up in list()")
-        .that(createdInList)
-        .isTrue();
-
-    HmacKey.HmacKeyMetadata getResult = storage.getHmacKey(accessId);
-    assertEquals(metadata, getResult);
-
-    storage.updateHmacKeyState(metadata, HmacKey.HmacKeyState.INACTIVE);
-
-    storage.deleteHmacKey(metadata);
-
-    metadatas = storage.listHmacKeys(Storage.ListHmacKeysOption.serviceAccount(serviceAccount));
-    boolean deletedInList =
-        StreamSupport.stream(metadatas.iterateAll().spliterator(), false)
-            .map(HmacKey.HmacKeyMetadata::getAccessId)
-            .anyMatch(accessId::equals);
-
-    assertWithMessage("Deleted an HMAC key but it showed up in list()")
-        .that(deletedInList)
-        .isFalse();
-  }
-
-  private void cleanUpHmacKeys(ServiceAccount serviceAccount) {
-    Instant now = Instant.now();
-    Instant yesterday = now.minus(Duration.ofDays(1L));
-
-    Page<HmacKey.HmacKeyMetadata> metadatas =
-        storage.listHmacKeys(Storage.ListHmacKeysOption.serviceAccount(serviceAccount));
-    for (HmacKey.HmacKeyMetadata hmacKeyMetadata : metadatas.iterateAll()) {
-      Instant updated = Instant.ofEpochMilli(hmacKeyMetadata.getUpdateTime());
-      if (updated.isBefore(yesterday)) {
-
-        if (hmacKeyMetadata.getState() == HmacKeyState.ACTIVE) {
-          hmacKeyMetadata = storage.updateHmacKeyState(hmacKeyMetadata, HmacKeyState.INACTIVE);
-        }
-
-        if (hmacKeyMetadata.getState() == HmacKeyState.INACTIVE) {
-          try {
-            storage.deleteHmacKey(hmacKeyMetadata);
-          } catch (StorageException e) {
-            // attempted to delete concurrently, if the other succeeded swallow the error
-            if (!(e.getReason().equals("invalid") && e.getMessage().contains("deleted"))) {
-              throw e;
-            }
-          }
-        }
-      }
-    }
-  }
-
-  @Test
-  public void testReadCompressedBlob() throws IOException {
-    String blobName = "test-read-compressed-blob";
-    BlobInfo blobInfo =
-        BlobInfo.newBuilder(BlobId.of(BUCKET, blobName))
-            .setContentType("text/plain")
-            .setContentEncoding("gzip")
-            .build();
-    Blob blob = storage.create(blobInfo, COMPRESSED_CONTENT);
-    try (ByteArrayOutputStream output = new ByteArrayOutputStream()) {
-      try (ReadChannel reader = storage.reader(BlobId.of(BUCKET, blobName))) {
-        reader.setChunkSize(8);
-        ByteBuffer buffer = ByteBuffer.allocate(8);
-        while (reader.read(buffer) != -1) {
-          buffer.flip();
-          output.write(buffer.array(), 0, buffer.limit());
-          buffer.clear();
-        }
-      }
-      assertArrayEquals(
-          BLOB_STRING_CONTENT.getBytes(UTF_8), storage.readAllBytes(BUCKET, blobName));
-      assertArrayEquals(COMPRESSED_CONTENT, output.toByteArray());
-      try (GZIPInputStream zipInput =
-          new GZIPInputStream(new ByteArrayInputStream(output.toByteArray()))) {
-        assertArrayEquals(BLOB_STRING_CONTENT.getBytes(UTF_8), ByteStreams.toByteArray(zipInput));
-      }
-    }
-  }
-
   @Test
   public void testAttemptObjectDeleteWithRetentionPolicy()
       throws ExecutionException, InterruptedException {
@@ -1663,67 +1250,6 @@ public class ITStorageTest {
     }
   }
 
-  @Test
-  public void testGetServiceAccount() {
-    String projectId = storage.getOptions().getProjectId();
-    ServiceAccount serviceAccount = storage.getServiceAccount(projectId);
-    assertNotNull(serviceAccount);
-    assertTrue(serviceAccount.getEmail().endsWith(SERVICE_ACCOUNT_EMAIL_SUFFIX));
-  }
-
-  @Test
-  @SuppressWarnings({"unchecked", "deprecation"})
-  public void testEnableAndDisableBucketPolicyOnlyOnExistingBucket() throws Exception {
-    String bpoBucket = RemoteStorageHelper.generateBucketName();
-    try {
-      // BPO is disabled by default.
-      Bucket bucket =
-          storage.create(
-              Bucket.newBuilder(bpoBucket)
-                  .setAcl(ImmutableList.of(Acl.of(User.ofAllAuthenticatedUsers(), Role.READER)))
-                  .setDefaultAcl(
-                      ImmutableList.of(Acl.of(User.ofAllAuthenticatedUsers(), Role.READER)))
-                  .build());
-
-      BucketInfo.IamConfiguration bpoEnabledIamConfiguration =
-          BucketInfo.IamConfiguration.newBuilder().setIsBucketPolicyOnlyEnabled(true).build();
-      bucket
-          .toBuilder()
-          .setAcl(null)
-          .setDefaultAcl(null)
-          .setIamConfiguration(bpoEnabledIamConfiguration)
-          .build()
-          .update();
-
-      Bucket remoteBucket =
-          storage.get(bpoBucket, Storage.BucketGetOption.fields(BucketField.IAMCONFIGURATION));
-
-      assertTrue(remoteBucket.getIamConfiguration().isBucketPolicyOnlyEnabled());
-      assertNotNull(remoteBucket.getIamConfiguration().getBucketPolicyOnlyLockedTime());
-
-      remoteBucket
-          .toBuilder()
-          .setIamConfiguration(
-              bpoEnabledIamConfiguration.toBuilder().setIsBucketPolicyOnlyEnabled(false).build())
-          .build()
-          .update();
-
-      remoteBucket =
-          storage.get(
-              bpoBucket,
-              Storage.BucketGetOption.fields(
-                  BucketField.IAMCONFIGURATION, BucketField.ACL, BucketField.DEFAULT_OBJECT_ACL));
-
-      assertFalse(remoteBucket.getIamConfiguration().isBucketPolicyOnlyEnabled());
-      assertEquals(User.ofAllAuthenticatedUsers(), remoteBucket.getDefaultAcl().get(0).getEntity());
-      assertEquals(Role.READER, remoteBucket.getDefaultAcl().get(0).getRole());
-      assertEquals(User.ofAllAuthenticatedUsers(), remoteBucket.getAcl().get(0).getEntity());
-      assertEquals(Role.READER, remoteBucket.getAcl().get(0).getRole());
-    } finally {
-      RemoteStorageHelper.forceDelete(storage, bpoBucket, 1, TimeUnit.MINUTES);
-    }
-  }
-
   @Test
   public void testBlobReload() throws Exception {
     String blobName = "test-blob-reload";
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITServiceAccountTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITServiceAccountTest.java
new file mode 100644
index 00000000..fe1d517f
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITServiceAccountTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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 org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import com.google.cloud.storage.ServiceAccount;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageFixture;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+public class ITServiceAccountTest {
+
+  @ClassRule public static final StorageFixture storageFixture = StorageFixture.defaultHttp();
+
+  private static final String SERVICE_ACCOUNT_EMAIL_SUFFIX =
+      "@gs-project-accounts.iam.gserviceaccount.com";
+
+  private static Storage storage;
+
+  @BeforeClass
+  public static void setup() {
+    storage = storageFixture.getInstance();
+  }
+
+  @Test
+  public void testGetServiceAccount() {
+    String projectId = storage.getOptions().getProjectId();
+    ServiceAccount serviceAccount = storage.getServiceAccount(projectId);
+    assertNotNull(serviceAccount);
+    assertTrue(serviceAccount.getEmail().endsWith(SERVICE_ACCOUNT_EMAIL_SUFFIX));
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITWriteChannelConnectionPoolTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITWriteChannelConnectionPoolTest.java
new file mode 100644
index 00000000..092a0205
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITWriteChannelConnectionPoolTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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 java.nio.charset.StandardCharsets.UTF_8;
+
+import com.google.api.client.http.HttpTransport;
+import com.google.api.client.http.apache.ApacheHttpTransport;
+import com.google.auth.http.HttpTransportFactory;
+import com.google.cloud.TransportOptions;
+import com.google.cloud.WriteChannel;
+import com.google.cloud.http.HttpTransportOptions;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketFixture;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageFixture;
+import com.google.cloud.storage.StorageOptions;
+import com.google.cloud.storage.testing.RemoteStorageHelper;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+public class ITWriteChannelConnectionPoolTest {
+  @ClassRule(order = 1)
+  public static final StorageFixture storageFixture =
+      StorageFixture.of(() -> RemoteStorageHelper.create().getOptions().getService());
+
+  @ClassRule(order = 2)
+  public static final BucketFixture bucketFixture =
+      BucketFixture.newBuilder().setHandle(storageFixture::getInstance).build();
+
+  private static final byte[] BLOB_BYTE_CONTENT = {0xD, 0xE, 0xA, 0xD};
+  private static final String BLOB_STRING_CONTENT = "Hello Google Cloud Storage!";
+
+  private static String bucketName;
+  private static Storage storage;
+
+  @BeforeClass
+  public static void beforeClass() throws IOException {
+    storage = storageFixture.getInstance();
+    bucketName = bucketFixture.getBucketInfo().getName();
+  }
+
+  private static class CustomHttpTransportFactory implements HttpTransportFactory {
+    @Override
+    @SuppressWarnings({"unchecked", "deprecation"})
+    public HttpTransport create() {
+      PoolingHttpClientConnectionManager manager = new PoolingHttpClientConnectionManager();
+      manager.setMaxTotal(1);
+      return new ApacheHttpTransport(HttpClients.createMinimal(manager));
+    }
+  }
+
+  @Test(timeout = 5000)
+  public void testWriteChannelWithConnectionPool() throws IOException {
+    TransportOptions transportOptions =
+        HttpTransportOptions.newBuilder()
+            .setHttpTransportFactory(new CustomHttpTransportFactory())
+            .build();
+    Storage storageWithPool =
+        StorageOptions.http().setTransportOptions(transportOptions).build().getService();
+    String blobName = "test-custom-pool-management";
+    BlobInfo blob = BlobInfo.newBuilder(bucketName, blobName).build();
+    byte[] stringBytes;
+    try (WriteChannel writer = storageWithPool.writer(blob)) {
+      stringBytes = BLOB_STRING_CONTENT.getBytes(UTF_8);
+      writer.write(ByteBuffer.wrap(BLOB_BYTE_CONTENT));
+      writer.write(ByteBuffer.wrap(stringBytes));
+    }
+    try (WriteChannel writer = storageWithPool.writer(blob)) {
+      stringBytes = BLOB_STRING_CONTENT.getBytes(UTF_8);
+      writer.write(ByteBuffer.wrap(BLOB_BYTE_CONTENT));
+      writer.write(ByteBuffer.wrap(stringBytes));
+    }
+  }
+}