Repository: googleapis/java-storage
Pull request #1593 (2022-08-29 21:29:05)

feat: implement GrpcStorageImpl#copy
Refactoring by pushing current CopyWriter implementation down to new
HttpCopyWriter and makeing CopyWriter abstract.




Patch:
diff --git a/google-cloud-storage/clirr-ignored-differences.xml b/google-cloud-storage/clirr-ignored-differences.xml
index f989ec09..e45beaea 100644
--- a/google-cloud-storage/clirr-ignored-differences.xml
+++ b/google-cloud-storage/clirr-ignored-differences.xml
@@ -52,4 +52,15 @@
     <method>void close()</method>
   </difference>
 
+  <!--
+  Allow com.google.cloud.storage.CopyWriter to now be abstract.
+  Previously, the class was only extendable within the package due to
+  constructor scope, making it abstract will be compatible for our
+  customers.
+  -->
+  <difference>
+    <differenceType>3005</differenceType>
+    <className>com/google/cloud/storage/CopyWriter</className>
+  </difference>
+
 </differences>
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/CopyWriter.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/CopyWriter.java
index 9240efa1..feeb9ef7 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/CopyWriter.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/CopyWriter.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Google LLC
+ * 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.
@@ -18,14 +18,6 @@ package com.google.cloud.storage;
 
 import com.google.cloud.Restorable;
 import com.google.cloud.RestorableState;
-import com.google.cloud.storage.spi.v1.StorageRpc;
-import com.google.cloud.storage.spi.v1.StorageRpc.RewriteRequest;
-import com.google.cloud.storage.spi.v1.StorageRpc.RewriteResponse;
-import com.google.common.base.MoreObjects;
-import java.io.Serializable;
-import java.util.Map;
-import java.util.Objects;
-import java.util.function.Function;
 
 /**
  * Google Storage blob copy writer. A {@code CopyWriter} object allows to copy both blob's data and
@@ -41,17 +33,10 @@ import java.util.function.Function;
  *
  * @see <a href="https://cloud.google.com/storage/docs/json_api/v1/objects/rewrite">Rewrite</a>
  */
-public class CopyWriter implements Restorable<CopyWriter> {
+public abstract class CopyWriter implements Restorable<CopyWriter> {
 
-  private final HttpStorageOptions serviceOptions;
-  private final StorageRpc storageRpc;
-  private RewriteResponse rewriteResponse;
-
-  CopyWriter(HttpStorageOptions serviceOptions, RewriteResponse rewriteResponse) {
-    this.serviceOptions = serviceOptions;
-    this.rewriteResponse = rewriteResponse;
-    this.storageRpc = serviceOptions.getStorageRpcV1();
-  }
+  // keep this class only extendable within our package
+  CopyWriter() {}
 
   /**
    * Returns the updated information for the written blob. Calling this method when {@code isDone()}
@@ -67,28 +52,16 @@ public class CopyWriter implements Restorable<CopyWriter> {
    *
    * @throws StorageException upon failure
    */
-  public Blob getResult() {
-    while (!isDone()) {
-      copyChunk();
-    }
-    BlobInfo info = Conversions.apiary().blobInfo().decode(rewriteResponse.result);
-    return info.asBlob(serviceOptions.getService());
-  }
+  public abstract Blob getResult();
 
   /** Returns the size of the blob being copied. */
-  public long getBlobSize() {
-    return rewriteResponse.blobSize;
-  }
+  public abstract long getBlobSize();
 
   /** Returns {@code true} if blob copy has finished, {@code false} otherwise. */
-  public boolean isDone() {
-    return rewriteResponse.isDone;
-  }
+  public abstract boolean isDone();
 
   /** Returns the number of bytes copied. */
-  public long getTotalBytesCopied() {
-    return rewriteResponse.totalBytesRewritten;
-  }
+  public abstract long getTotalBytesCopied();
 
   /**
    * Copies the next chunk of the blob. An RPC is issued only if copy has not finished yet ({@link
@@ -96,220 +69,8 @@ public class CopyWriter implements Restorable<CopyWriter> {
    *
    * @throws StorageException upon failure
    */
-  public void copyChunk() {
-    if (!isDone()) {
-      RewriteRequest rewriteRequest = rewriteResponse.rewriteRequest;
-      this.rewriteResponse =
-          Retrying.run(
-              serviceOptions,
-              serviceOptions.getRetryAlgorithmManager().getForObjectsRewrite(rewriteRequest),
-              () -> storageRpc.continueRewrite(rewriteResponse),
-              Function.identity());
-    }
-  }
+  public abstract void copyChunk();
 
   @Override
-  public RestorableState<CopyWriter> capture() {
-    return StateImpl.newBuilder(
-            serviceOptions,
-            Conversions.apiary().blobId().decode(rewriteResponse.rewriteRequest.source),
-            rewriteResponse.rewriteRequest.sourceOptions,
-            rewriteResponse.rewriteRequest.overrideInfo,
-            Conversions.apiary().blobInfo().decode(rewriteResponse.rewriteRequest.target),
-            rewriteResponse.rewriteRequest.targetOptions)
-        .setResult(
-            rewriteResponse.result != null
-                ? Conversions.apiary().blobInfo().decode(rewriteResponse.result)
-                : null)
-        .setBlobSize(getBlobSize())
-        .setIsDone(isDone())
-        .setMegabytesCopiedPerChunk(rewriteResponse.rewriteRequest.megabytesRewrittenPerCall)
-        .setRewriteToken(rewriteResponse.rewriteToken)
-        .setTotalBytesRewritten(getTotalBytesCopied())
-        .build();
-  }
-
-  static class StateImpl implements RestorableState<CopyWriter>, Serializable {
-
-    private static final long serialVersionUID = 1693964441435822700L;
-
-    private final HttpStorageOptions serviceOptions;
-    private final BlobId source;
-    private final Map<StorageRpc.Option, ?> sourceOptions;
-    private final boolean overrideInfo;
-    private final BlobInfo target;
-    private final Map<StorageRpc.Option, ?> targetOptions;
-    private final BlobInfo result;
-    private final long blobSize;
-    private final boolean isDone;
-    private final String rewriteToken;
-    private final long totalBytesCopied;
-    private final Long megabytesCopiedPerChunk;
-
-    StateImpl(Builder builder) {
-      this.serviceOptions = builder.serviceOptions;
-      this.source = builder.source;
-      this.sourceOptions = builder.sourceOptions;
-      this.overrideInfo = builder.overrideInfo;
-      this.target = builder.target;
-      this.targetOptions = builder.targetOptions;
-      this.result = builder.result;
-      this.blobSize = builder.blobSize;
-      this.isDone = builder.isDone;
-      this.rewriteToken = builder.rewriteToken;
-      this.totalBytesCopied = builder.totalBytesCopied;
-      this.megabytesCopiedPerChunk = builder.megabytesCopiedPerChunk;
-    }
-
-    static class Builder {
-
-      private final HttpStorageOptions serviceOptions;
-      private final BlobId source;
-      private final Map<StorageRpc.Option, ?> sourceOptions;
-      private final boolean overrideInfo;
-      private final BlobInfo target;
-      private final Map<StorageRpc.Option, ?> targetOptions;
-      private BlobInfo result;
-      private long blobSize;
-      private boolean isDone;
-      private String rewriteToken;
-      private long totalBytesCopied;
-      private Long megabytesCopiedPerChunk;
-
-      private Builder(
-          HttpStorageOptions options,
-          BlobId source,
-          Map<StorageRpc.Option, ?> sourceOptions,
-          boolean overrideInfo,
-          BlobInfo target,
-          Map<StorageRpc.Option, ?> targetOptions) {
-        this.serviceOptions = options;
-        this.source = source;
-        this.sourceOptions = sourceOptions;
-        this.overrideInfo = overrideInfo;
-        this.target = target;
-        this.targetOptions = targetOptions;
-      }
-
-      Builder setResult(BlobInfo result) {
-        this.result = result;
-        return this;
-      }
-
-      Builder setBlobSize(long blobSize) {
-        this.blobSize = blobSize;
-        return this;
-      }
-
-      Builder setIsDone(boolean isDone) {
-        this.isDone = isDone;
-        return this;
-      }
-
-      Builder setRewriteToken(String rewriteToken) {
-        this.rewriteToken = rewriteToken;
-        return this;
-      }
-
-      Builder setTotalBytesRewritten(long totalBytesRewritten) {
-        this.totalBytesCopied = totalBytesRewritten;
-        return this;
-      }
-
-      Builder setMegabytesCopiedPerChunk(Long megabytesCopiedPerChunk) {
-        this.megabytesCopiedPerChunk = megabytesCopiedPerChunk;
-        return this;
-      }
-
-      RestorableState<CopyWriter> build() {
-        return new StateImpl(this);
-      }
-    }
-
-    static Builder newBuilder(
-        HttpStorageOptions options,
-        BlobId source,
-        Map<StorageRpc.Option, ?> sourceOptions,
-        boolean overrideInfo,
-        BlobInfo target,
-        Map<StorageRpc.Option, ?> targetOptions) {
-      return new Builder(options, source, sourceOptions, overrideInfo, target, targetOptions);
-    }
-
-    @Override
-    public CopyWriter restore() {
-      RewriteRequest rewriteRequest =
-          new RewriteRequest(
-              Conversions.apiary().blobId().encode(source),
-              sourceOptions,
-              overrideInfo,
-              Conversions.apiary().blobInfo().encode(target),
-              targetOptions,
-              megabytesCopiedPerChunk);
-      RewriteResponse rewriteResponse =
-          new RewriteResponse(
-              rewriteRequest,
-              result != null ? Conversions.apiary().blobInfo().encode(result) : null,
-              blobSize,
-              isDone,
-              rewriteToken,
-              totalBytesCopied);
-      return new CopyWriter(serviceOptions, rewriteResponse);
-    }
-
-    @Override
-    public int hashCode() {
-      return Objects.hash(
-          serviceOptions,
-          source,
-          sourceOptions,
-          overrideInfo,
-          target,
-          targetOptions,
-          result,
-          blobSize,
-          isDone,
-          megabytesCopiedPerChunk,
-          rewriteToken,
-          totalBytesCopied);
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-      if (obj == null) {
-        return false;
-      }
-      if (!(obj instanceof StateImpl)) {
-        return false;
-      }
-      final StateImpl other = (StateImpl) obj;
-      return Objects.equals(this.serviceOptions, other.serviceOptions)
-          && Objects.equals(this.source, other.source)
-          && Objects.equals(this.sourceOptions, other.sourceOptions)
-          && Objects.equals(this.overrideInfo, other.overrideInfo)
-          && Objects.equals(this.target, other.target)
-          && Objects.equals(this.targetOptions, other.targetOptions)
-          && Objects.equals(this.result, other.result)
-          && Objects.equals(this.rewriteToken, other.rewriteToken)
-          && Objects.equals(this.megabytesCopiedPerChunk, other.megabytesCopiedPerChunk)
-          && this.blobSize == other.blobSize
-          && this.isDone == other.isDone
-          && this.totalBytesCopied == other.totalBytesCopied;
-    }
-
-    @Override
-    public String toString() {
-      return MoreObjects.toStringHelper(this)
-          .add("source", source)
-          .add("overrideInfo", overrideInfo)
-          .add("target", target)
-          .add("result", result)
-          .add("blobSize", blobSize)
-          .add("isDone", isDone)
-          .add("rewriteToken", rewriteToken)
-          .add("totalBytesCopied", totalBytesCopied)
-          .add("megabytesCopiedPerChunk", megabytesCopiedPerChunk)
-          .toString();
-    }
-  }
+  public abstract RestorableState<CopyWriter> capture();
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java
new file mode 100644
index 00000000..96b83291
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java
@@ -0,0 +1,91 @@
+/*
+ * 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.Utils.todo;
+
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.gax.rpc.UnaryCallable;
+import com.google.cloud.RestorableState;
+import com.google.cloud.storage.Conversions.Decoder;
+import com.google.storage.v2.RewriteObjectRequest;
+import com.google.storage.v2.RewriteResponse;
+
+final class GapicCopyWriter extends CopyWriter {
+
+  // needed for #getResult
+  private final transient GrpcStorageImpl storage;
+  private final GrpcStorageOptions options;
+  private final UnaryCallable<RewriteObjectRequest, RewriteResponse> callable;
+  private final ResultRetryAlgorithm<?> alg;
+  private final RewriteResponse initialResponse;
+
+  private RewriteResponse mostRecentResponse;
+
+  GapicCopyWriter(
+      GrpcStorageImpl storage,
+      UnaryCallable<RewriteObjectRequest, RewriteResponse> callable,
+      ResultRetryAlgorithm<?> alg,
+      RewriteResponse initialResponse) {
+    this.storage = storage;
+    this.options = storage.getOptions();
+    this.callable = callable;
+    this.alg = alg;
+    this.initialResponse = initialResponse;
+    this.mostRecentResponse = initialResponse;
+  }
+
+  @Override
+  public Blob getResult() {
+    while (!isDone()) {
+      copyChunk();
+    }
+    BlobInfo info = Conversions.grpc().blobInfo().decode(mostRecentResponse.getResource());
+    return info.asBlob(storage);
+  }
+
+  @Override
+  public long getBlobSize() {
+    return initialResponse.getObjectSize();
+  }
+
+  @Override
+  public boolean isDone() {
+    return mostRecentResponse.getDone();
+  }
+
+  @Override
+  public long getTotalBytesCopied() {
+    return mostRecentResponse.getTotalBytesRewritten();
+  }
+
+  @Override
+  public void copyChunk() {
+    if (!isDone()) {
+      RewriteObjectRequest req =
+          RewriteObjectRequest.newBuilder()
+              .setRewriteToken(mostRecentResponse.getRewriteToken())
+              .build();
+      mostRecentResponse = Retrying.run(options, alg, () -> callable.call(req), Decoder.identity());
+    }
+  }
+
+  @Override
+  public RestorableState<CopyWriter> capture() {
+    return todo();
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcRetryAlgorithmManager.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcRetryAlgorithmManager.java
index 3b0f7f52..89ada1fe 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcRetryAlgorithmManager.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcRetryAlgorithmManager.java
@@ -57,6 +57,14 @@ final class GrpcRetryAlgorithmManager implements Serializable {
     this.retryStrategy = retryStrategy;
   }
 
+  /**
+   * Some operations are inherently idempotent after they're started (Resumable uploads, rewrites)
+   * provide access to the idempotent {@link ResultRetryAlgorithm} for those uses.
+   */
+  ResultRetryAlgorithm<?> idempotent() {
+    return retryStrategy.getIdempotentHandler();
+  }
+
   public ResultRetryAlgorithm<?> getFor(ComposeObjectRequest req) {
     return req.hasIfGenerationMatch()
         ? retryStrategy.getIdempotentHandler()
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 8aa232a5..97d7d4c5 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
@@ -56,6 +56,7 @@ import com.google.cloud.storage.UnifiedOpts.BucketTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.HmacKeyListOpt;
 import com.google.cloud.storage.UnifiedOpts.HmacKeySourceOpt;
 import com.google.cloud.storage.UnifiedOpts.HmacKeyTargetOpt;
+import com.google.cloud.storage.UnifiedOpts.Mapper;
 import com.google.cloud.storage.UnifiedOpts.ObjectListOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
@@ -84,6 +85,8 @@ import com.google.storage.v2.ListObjectsRequest;
 import com.google.storage.v2.Object;
 import com.google.storage.v2.ProjectName;
 import com.google.storage.v2.ReadObjectRequest;
+import com.google.storage.v2.RewriteObjectRequest;
+import com.google.storage.v2.RewriteResponse;
 import com.google.storage.v2.StorageClient;
 import com.google.storage.v2.StorageClient.ListBucketsPage;
 import com.google.storage.v2.StorageClient.ListBucketsPagedResponse;
@@ -567,7 +570,51 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public CopyWriter copy(CopyRequest copyRequest) {
-    return throwNotYetImplemented(fmtMethodName("copy", CopyRequest.class));
+    BlobId src = copyRequest.getSource();
+    BlobInfo dst = copyRequest.getTarget();
+    Opts<ObjectSourceOpt> srcOpts =
+        Opts.unwrap(copyRequest.getSourceOptions()).projectAsSource().resolveFrom(src);
+    Opts<ObjectTargetOpt> dstOpts = Opts.unwrap(copyRequest.getTargetOptions()).resolveFrom(dst);
+
+    Mapper<RewriteObjectRequest.Builder> mapper =
+        srcOpts.rewriteObjectsRequest().andThen(dstOpts.rewriteObjectsRequest());
+
+    Object srcProto = codecs.blobId().encode(src);
+    Object dstProto = codecs.blobInfo().encode(dst);
+
+    RewriteObjectRequest.Builder b =
+        RewriteObjectRequest.newBuilder()
+            .setDestinationName(dstProto.getName())
+            .setDestinationBucket(dstProto.getBucket())
+            // destination_kms_key comes from dstOpts
+            // according to the docs in the protos, it is illegal to populate the following fields,
+            // clear them out if they are set
+            // destination_predefined_acl comes from dstOpts
+            // if_*_match come from srcOpts and dstOpts
+            // copy_source_encryption_* come from srcOpts
+            // common_object_request_params come from dstOpts
+            .setDestination(dstProto.toBuilder().clearName().clearBucket().clearKmsKey().build())
+            .setSourceBucket(srcProto.getBucket())
+            .setSourceObject(srcProto.getName());
+
+    if (src.getGeneration() != null) {
+      b.setSourceGeneration(src.getGeneration());
+    }
+
+    if (copyRequest.getMegabytesCopiedPerChunk() != null) {
+      b.setMaxBytesRewrittenPerCall(copyRequest.getMegabytesCopiedPerChunk());
+    }
+
+    RewriteObjectRequest req = mapper.apply(b).build();
+    GrpcCallContext grpcCallContext =
+        srcOpts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
+    UnaryCallable<RewriteObjectRequest, RewriteResponse> callable =
+        storageClient.rewriteObjectCallable().withDefaultCallContext(grpcCallContext);
+    return Retrying.run(
+        getOptions(),
+        retryAlgorithmManager.getFor(req),
+        () -> callable.call(req),
+        (resp) -> new GapicCopyWriter(this, callable, retryAlgorithmManager.idempotent(), resp));
   }
 
   @Override
@@ -997,7 +1044,9 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   @Override
   public ServiceAccount getServiceAccount(String projectId) {
     GetServiceAccountRequest req =
-        GetServiceAccountRequest.newBuilder().setProject(projectId).build();
+        GetServiceAccountRequest.newBuilder()
+            .setProject(projectNameCodec.encode(projectId))
+            .build();
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
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 1a24f400..426f5756 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
@@ -381,6 +381,7 @@ public final class GrpcStorageOptions extends StorageOptions
    * @see GrpcStorageOptions.GrpcStorageDefaults#getDefaultServiceFactory()
    */
   @InternalApi
+  @BetaApi
   public static class GrpcStorageFactory implements StorageFactory {
 
     /**
@@ -400,6 +401,7 @@ public final class GrpcStorageOptions extends StorageOptions
     @InternalApi
     @Deprecated
     @SuppressWarnings("DeprecatedIsStillUsed")
+    @BetaApi
     public GrpcStorageFactory() {}
 
     @Override
@@ -450,6 +452,7 @@ public final class GrpcStorageOptions extends StorageOptions
     @InternalApi
     @Deprecated
     @SuppressWarnings("DeprecatedIsStillUsed")
+    @BetaApi
     public GrpcStorageRpcFactory() {}
 
     @Override
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java
new file mode 100644
index 00000000..42abd299
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java
@@ -0,0 +1,282 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import com.google.cloud.RestorableState;
+import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.cloud.storage.spi.v1.StorageRpc.RewriteRequest;
+import com.google.cloud.storage.spi.v1.StorageRpc.RewriteResponse;
+import com.google.common.base.MoreObjects;
+import java.io.Serializable;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Function;
+
+public class HttpCopyWriter extends CopyWriter {
+
+  private final HttpStorageOptions serviceOptions;
+  private final StorageRpc storageRpc;
+  private RewriteResponse rewriteResponse;
+
+  HttpCopyWriter(HttpStorageOptions serviceOptions, RewriteResponse rewriteResponse) {
+    this.serviceOptions = serviceOptions;
+    this.rewriteResponse = rewriteResponse;
+    this.storageRpc = serviceOptions.getStorageRpcV1();
+  }
+
+  @Override
+  public Blob getResult() {
+    while (!isDone()) {
+      copyChunk();
+    }
+    BlobInfo info = Conversions.apiary().blobInfo().decode(rewriteResponse.result);
+    return info.asBlob(serviceOptions.getService());
+  }
+
+  @Override
+  public long getBlobSize() {
+    return rewriteResponse.blobSize;
+  }
+
+  @Override
+  public boolean isDone() {
+    return rewriteResponse.isDone;
+  }
+
+  @Override
+  public long getTotalBytesCopied() {
+    return rewriteResponse.totalBytesRewritten;
+  }
+
+  @Override
+  public void copyChunk() {
+    if (!isDone()) {
+      RewriteRequest rewriteRequest = rewriteResponse.rewriteRequest;
+      this.rewriteResponse =
+          Retrying.run(
+              serviceOptions,
+              serviceOptions.getRetryAlgorithmManager().getForObjectsRewrite(rewriteRequest),
+              () -> storageRpc.continueRewrite(rewriteResponse),
+              Function.identity());
+    }
+  }
+
+  @Override
+  public RestorableState<CopyWriter> capture() {
+    return StateImpl.newBuilder(
+            serviceOptions,
+            Conversions.apiary().blobId().decode(rewriteResponse.rewriteRequest.source),
+            rewriteResponse.rewriteRequest.sourceOptions,
+            rewriteResponse.rewriteRequest.overrideInfo,
+            Conversions.apiary().blobInfo().decode(rewriteResponse.rewriteRequest.target),
+            rewriteResponse.rewriteRequest.targetOptions)
+        .setResult(
+            rewriteResponse.result != null
+                ? Conversions.apiary().blobInfo().decode(rewriteResponse.result)
+                : null)
+        .setBlobSize(getBlobSize())
+        .setIsDone(isDone())
+        .setMegabytesCopiedPerChunk(rewriteResponse.rewriteRequest.megabytesRewrittenPerCall)
+        .setRewriteToken(rewriteResponse.rewriteToken)
+        .setTotalBytesRewritten(getTotalBytesCopied())
+        .build();
+  }
+
+  static class StateImpl implements RestorableState<CopyWriter>, Serializable {
+
+    private static final long serialVersionUID = 1693964441435822700L;
+
+    private final HttpStorageOptions serviceOptions;
+    private final BlobId source;
+    private final Map<StorageRpc.Option, ?> sourceOptions;
+    private final boolean overrideInfo;
+    private final BlobInfo target;
+    private final Map<StorageRpc.Option, ?> targetOptions;
+    private final BlobInfo result;
+    private final long blobSize;
+    private final boolean isDone;
+    private final String rewriteToken;
+    private final long totalBytesCopied;
+    private final Long megabytesCopiedPerChunk;
+
+    StateImpl(Builder builder) {
+      this.serviceOptions = builder.serviceOptions;
+      this.source = builder.source;
+      this.sourceOptions = builder.sourceOptions;
+      this.overrideInfo = builder.overrideInfo;
+      this.target = builder.target;
+      this.targetOptions = builder.targetOptions;
+      this.result = builder.result;
+      this.blobSize = builder.blobSize;
+      this.isDone = builder.isDone;
+      this.rewriteToken = builder.rewriteToken;
+      this.totalBytesCopied = builder.totalBytesCopied;
+      this.megabytesCopiedPerChunk = builder.megabytesCopiedPerChunk;
+    }
+
+    static class Builder {
+
+      private final HttpStorageOptions serviceOptions;
+      private final BlobId source;
+      private final Map<StorageRpc.Option, ?> sourceOptions;
+      private final boolean overrideInfo;
+      private final BlobInfo target;
+      private final Map<StorageRpc.Option, ?> targetOptions;
+      private BlobInfo result;
+      private long blobSize;
+      private boolean isDone;
+      private String rewriteToken;
+      private long totalBytesCopied;
+      private Long megabytesCopiedPerChunk;
+
+      private Builder(
+          HttpStorageOptions options,
+          BlobId source,
+          Map<StorageRpc.Option, ?> sourceOptions,
+          boolean overrideInfo,
+          BlobInfo target,
+          Map<StorageRpc.Option, ?> targetOptions) {
+        this.serviceOptions = options;
+        this.source = source;
+        this.sourceOptions = sourceOptions;
+        this.overrideInfo = overrideInfo;
+        this.target = target;
+        this.targetOptions = targetOptions;
+      }
+
+      Builder setResult(BlobInfo result) {
+        this.result = result;
+        return this;
+      }
+
+      Builder setBlobSize(long blobSize) {
+        this.blobSize = blobSize;
+        return this;
+      }
+
+      Builder setIsDone(boolean isDone) {
+        this.isDone = isDone;
+        return this;
+      }
+
+      Builder setRewriteToken(String rewriteToken) {
+        this.rewriteToken = rewriteToken;
+        return this;
+      }
+
+      Builder setTotalBytesRewritten(long totalBytesRewritten) {
+        this.totalBytesCopied = totalBytesRewritten;
+        return this;
+      }
+
+      Builder setMegabytesCopiedPerChunk(Long megabytesCopiedPerChunk) {
+        this.megabytesCopiedPerChunk = megabytesCopiedPerChunk;
+        return this;
+      }
+
+      RestorableState<CopyWriter> build() {
+        return new StateImpl(this);
+      }
+    }
+
+    static Builder newBuilder(
+        HttpStorageOptions options,
+        BlobId source,
+        Map<StorageRpc.Option, ?> sourceOptions,
+        boolean overrideInfo,
+        BlobInfo target,
+        Map<StorageRpc.Option, ?> targetOptions) {
+      return new Builder(options, source, sourceOptions, overrideInfo, target, targetOptions);
+    }
+
+    @Override
+    public CopyWriter restore() {
+      RewriteRequest rewriteRequest =
+          new RewriteRequest(
+              Conversions.apiary().blobId().encode(source),
+              sourceOptions,
+              overrideInfo,
+              Conversions.apiary().blobInfo().encode(target),
+              targetOptions,
+              megabytesCopiedPerChunk);
+      RewriteResponse rewriteResponse =
+          new RewriteResponse(
+              rewriteRequest,
+              result != null ? Conversions.apiary().blobInfo().encode(result) : null,
+              blobSize,
+              isDone,
+              rewriteToken,
+              totalBytesCopied);
+      return new HttpCopyWriter(serviceOptions, rewriteResponse);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(
+          serviceOptions,
+          source,
+          sourceOptions,
+          overrideInfo,
+          target,
+          targetOptions,
+          result,
+          blobSize,
+          isDone,
+          megabytesCopiedPerChunk,
+          rewriteToken,
+          totalBytesCopied);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (obj == null) {
+        return false;
+      }
+      if (!(obj instanceof StateImpl)) {
+        return false;
+      }
+      final StateImpl other = (StateImpl) obj;
+      return Objects.equals(this.serviceOptions, other.serviceOptions)
+          && Objects.equals(this.source, other.source)
+          && Objects.equals(this.sourceOptions, other.sourceOptions)
+          && Objects.equals(this.overrideInfo, other.overrideInfo)
+          && Objects.equals(this.target, other.target)
+          && Objects.equals(this.targetOptions, other.targetOptions)
+          && Objects.equals(this.result, other.result)
+          && Objects.equals(this.rewriteToken, other.rewriteToken)
+          && Objects.equals(this.megabytesCopiedPerChunk, other.megabytesCopiedPerChunk)
+          && this.blobSize == other.blobSize
+          && this.isDone == other.isDone
+          && this.totalBytesCopied == other.totalBytesCopied;
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this)
+          .add("source", source)
+          .add("overrideInfo", overrideInfo)
+          .add("target", target)
+          .add("result", result)
+          .add("blobSize", blobSize)
+          .add("isDone", isDone)
+          .add("rewriteToken", rewriteToken)
+          .add("totalBytesCopied", totalBytesCopied)
+          .add("megabytesCopiedPerChunk", megabytesCopiedPerChunk)
+          .toString();
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
index 78181327..f8c73cbf 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
@@ -558,7 +558,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
     return run(
         algorithm,
         () -> storageRpc.openRewrite(rewriteRequest),
-        (r) -> new CopyWriter(getOptions(), r));
+        (r) -> new HttpCopyWriter(getOptions(), r));
   }
 
   @Override
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
index ddfae852..c735bd1a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
@@ -712,7 +712,7 @@ final class UnifiedOpts {
 
     @Override
     public Mapper<RewriteObjectRequest.Builder> rewriteObject() {
-      return Mapper.identity();
+      return b -> b.setIfGenerationMatch(val);
     }
 
     @Override
@@ -763,7 +763,7 @@ final class UnifiedOpts {
 
     @Override
     public Mapper<RewriteObjectRequest.Builder> rewriteObject() {
-      return Mapper.identity();
+      return b -> b.setIfGenerationNotMatch(val);
     }
 
     @Override
@@ -791,6 +791,11 @@ final class UnifiedOpts {
     public Mapper<ComposeObjectRequest.Builder> composeObject() {
       return b -> b.setKmsKey(val);
     }
+
+    @Override
+    public Mapper<RewriteObjectRequest.Builder> rewriteObject() {
+      return b -> b.setDestinationKmsKey(val);
+    }
   }
 
   @Deprecated
@@ -881,7 +886,7 @@ final class UnifiedOpts {
 
     @Override
     public Mapper<RewriteObjectRequest.Builder> rewriteObject() {
-      return Mapper.identity();
+      return b -> b.setIfMetagenerationMatch(val);
     }
 
     @Override
@@ -951,7 +956,7 @@ final class UnifiedOpts {
 
     @Override
     public Mapper<RewriteObjectRequest.Builder> rewriteObject() {
-      return Mapper.identity();
+      return b -> b.setIfMetagenerationNotMatch(val);
     }
 
     @Override
@@ -1058,6 +1063,11 @@ final class UnifiedOpts {
     public Mapper<CreateBucketRequest.Builder> createBucket() {
       return b -> b.setPredefinedAcl(val);
     }
+
+    @Override
+    public Mapper<RewriteObjectRequest.Builder> rewriteObject() {
+      return b -> b.setDestinationPredefinedAcl(val);
+    }
   }
 
   static final class PredefinedDefaultObjectAcl extends RpcOptVal<String>


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java
index bd91da40..cd51a1e9 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java
@@ -119,7 +119,7 @@ public class CopyWriterTest {
     EasyMock.expect(storageRpcMock.continueRewrite(RESPONSE_WITH_OBJECT))
         .andReturn(RESPONSE_WITH_OBJECT_DONE);
     EasyMock.replay(storageRpcMock);
-    copyWriter = new CopyWriter(options, RESPONSE_WITH_OBJECT);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITH_OBJECT);
     assertEquals(result, copyWriter.getResult());
     assertTrue(copyWriter.isDone());
     assertEquals(42L, copyWriter.getTotalBytesCopied());
@@ -131,7 +131,7 @@ public class CopyWriterTest {
     EasyMock.expect(storageRpcMock.continueRewrite(RESPONSE_WITHOUT_OBJECT))
         .andReturn(RESPONSE_WITHOUT_OBJECT_DONE);
     EasyMock.replay(storageRpcMock);
-    copyWriter = new CopyWriter(options, RESPONSE_WITHOUT_OBJECT);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITHOUT_OBJECT);
     assertEquals(result, copyWriter.getResult());
     assertTrue(copyWriter.isDone());
     assertEquals(42L, copyWriter.getTotalBytesCopied());
@@ -145,7 +145,7 @@ public class CopyWriterTest {
     EasyMock.expect(storageRpcMock.continueRewrite(RESPONSE_WITH_OBJECT))
         .andReturn(RESPONSE_WITH_OBJECT_DONE);
     EasyMock.replay(storageRpcMock);
-    copyWriter = new CopyWriter(options, RESPONSE_WITH_OBJECT);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITH_OBJECT);
     assertEquals(result, copyWriter.getResult());
     assertTrue(copyWriter.isDone());
     assertEquals(42L, copyWriter.getTotalBytesCopied());
@@ -159,7 +159,7 @@ public class CopyWriterTest {
     EasyMock.expect(storageRpcMock.continueRewrite(RESPONSE_WITHOUT_OBJECT))
         .andReturn(RESPONSE_WITHOUT_OBJECT_DONE);
     EasyMock.replay(storageRpcMock);
-    copyWriter = new CopyWriter(options, RESPONSE_WITHOUT_OBJECT);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITHOUT_OBJECT);
     assertEquals(result, copyWriter.getResult());
     assertTrue(copyWriter.isDone());
     assertEquals(42L, copyWriter.getTotalBytesCopied());
@@ -173,7 +173,7 @@ public class CopyWriterTest {
     EasyMock.expect(storageRpcMock.continueRewrite(RESPONSE_WITH_OBJECT))
         .andReturn(RESPONSE_WITH_OBJECT_DONE);
     EasyMock.replay(storageRpcMock);
-    copyWriter = new CopyWriter(options, RESPONSE_WITH_OBJECT);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITH_OBJECT);
     copyWriter.copyChunk();
     assertTrue(!copyWriter.isDone());
     assertEquals(21L, copyWriter.getTotalBytesCopied());
@@ -193,7 +193,7 @@ public class CopyWriterTest {
     EasyMock.expect(storageRpcMock.continueRewrite(RESPONSE_WITHOUT_OBJECT))
         .andReturn(RESPONSE_WITHOUT_OBJECT_DONE);
     EasyMock.replay(storageRpcMock);
-    copyWriter = new CopyWriter(options, RESPONSE_WITHOUT_OBJECT);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITHOUT_OBJECT);
     copyWriter.copyChunk();
     assertTrue(!copyWriter.isDone());
     assertEquals(21L, copyWriter.getTotalBytesCopied());
@@ -211,7 +211,7 @@ public class CopyWriterTest {
     EasyMock.expect(storageRpcMock.continueRewrite(RESPONSE_WITH_OBJECT))
         .andReturn(RESPONSE_WITH_OBJECT_DONE);
     EasyMock.replay(storageRpcMock);
-    copyWriter = new CopyWriter(options, RESPONSE_WITH_OBJECT);
+    copyWriter = new HttpCopyWriter(options, RESPONSE_WITH_OBJECT);
     copyWriter.copyChunk();
     assertEquals(result, copyWriter.getResult());
     assertTrue(copyWriter.isDone());
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java
index 6e384bb3..7925eb50 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java
@@ -302,6 +302,18 @@ public final class UnifiedOptsGrpcTest {
                 .build();
         assertThat(actual).isEqualTo(expected);
       }
+
+      @Test
+      public void rewriteObject() {
+        RewriteObjectRequest expected =
+            RewriteObjectRequest.newBuilder().setIfGenerationMatch(1L).build();
+        RewriteObjectRequest actual =
+            UnifiedOpts.generationMatch(1L)
+                .rewriteObject()
+                .apply(RewriteObjectRequest.newBuilder())
+                .build();
+        assertThat(actual).isEqualTo(expected);
+      }
     }
 
     public static final class GenerationNotMatchTest {
@@ -366,6 +378,18 @@ public final class UnifiedOptsGrpcTest {
                 .build();
         assertThat(actual).isEqualTo(expected);
       }
+
+      @Test
+      public void rewriteObject() {
+        RewriteObjectRequest expected =
+            RewriteObjectRequest.newBuilder().setIfGenerationNotMatch(1L).build();
+        RewriteObjectRequest actual =
+            UnifiedOpts.generationNotMatch(1L)
+                .rewriteObject()
+                .apply(RewriteObjectRequest.newBuilder())
+                .build();
+        assertThat(actual).isEqualTo(expected);
+      }
     }
 
     public static final class KmsKeyNameTest {
@@ -400,6 +424,20 @@ public final class UnifiedOptsGrpcTest {
 
         assertThat(actual).isEqualTo(expected);
       }
+
+      @Test
+      public void rewriteObject() {
+        RewriteObjectRequest expected =
+            RewriteObjectRequest.newBuilder().setDestinationKmsKey("key").build();
+
+        RewriteObjectRequest actual =
+            UnifiedOpts.kmsKeyName("key")
+                .rewriteObject()
+                .apply(RewriteObjectRequest.newBuilder())
+                .build();
+
+        assertThat(actual).isEqualTo(expected);
+      }
     }
 
     public static final class MetagenerationMatchTest {
@@ -477,6 +515,18 @@ public final class UnifiedOptsGrpcTest {
         assertThat(actual).isEqualTo(expected);
       }
 
+      @Test
+      public void rewriteObject() {
+        RewriteObjectRequest expected =
+            RewriteObjectRequest.newBuilder().setIfMetagenerationMatch(1L).build();
+        RewriteObjectRequest actual =
+            UnifiedOpts.metagenerationMatch(1L)
+                .rewriteObject()
+                .apply(RewriteObjectRequest.newBuilder())
+                .build();
+        assertThat(actual).isEqualTo(expected);
+      }
+
       @Test
       public void updateBucket() {
         UpdateBucketRequest expected =
@@ -577,6 +627,18 @@ public final class UnifiedOptsGrpcTest {
         assertThat(actual).isEqualTo(expected);
       }
 
+      @Test
+      public void rewriteObject() {
+        RewriteObjectRequest expected =
+            RewriteObjectRequest.newBuilder().setIfMetagenerationNotMatch(1L).build();
+        RewriteObjectRequest actual =
+            UnifiedOpts.metagenerationNotMatch(1L)
+                .rewriteObject()
+                .apply(RewriteObjectRequest.newBuilder())
+                .build();
+        assertThat(actual).isEqualTo(expected);
+      }
+
       @Test
       public void updateBucket() {
         UpdateBucketRequest expected =
@@ -723,6 +785,18 @@ public final class UnifiedOptsGrpcTest {
         assertThat(actual).isEqualTo(expected);
       }
 
+      @Test
+      public void rewriteObject() {
+        RewriteObjectRequest expected =
+            RewriteObjectRequest.newBuilder().setDestinationPredefinedAcl("private").build();
+        RewriteObjectRequest actual =
+            UnifiedOpts.predefinedAcl(PredefinedAcl.PRIVATE)
+                .rewriteObject()
+                .apply(RewriteObjectRequest.newBuilder())
+                .build();
+        assertThat(actual).isEqualTo(expected);
+      }
+
       @Test
       public void updateBucket() {
         UpdateBucketRequest expected =
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 53be1a63..c952e707 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
@@ -26,14 +26,17 @@ 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.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.BlobSourceOption;
 import com.google.cloud.storage.Storage.BlobTargetOption;
 import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.Storage.CopyRequest;
 import com.google.cloud.storage.Storage.CreateHmacKeyOption;
 import com.google.cloud.storage.Storage.ListHmacKeysOption;
 import com.google.cloud.storage.StorageFixture;
@@ -237,4 +240,29 @@ public final class ITGrpcTest {
     byte[] actual = storageFixture.getInstance().readAllBytes(info.getBlobId());
     assertThat(actual).isEqualTo(content);
   }
+
+  @Test
+  public void storageCopy() {
+    Storage s = storageFixture.getInstance();
+
+    byte[] expected = "Hello, World!".getBytes(StandardCharsets.UTF_8);
+
+    BlobInfo info = BlobInfo.newBuilder(bucketFixture.getBucketInfo(), "copy/src").build();
+    Blob cpySrc = s.create(info, expected, BlobTargetOption.doesNotExist());
+
+    BlobInfo dst = BlobInfo.newBuilder(bucketFixture.getBucketInfo(), "copy/dst").build();
+
+    CopyRequest copyRequest =
+        CopyRequest.newBuilder()
+            .setSource(cpySrc.getBlobId())
+            .setSourceOptions(BlobSourceOption.generationMatch(cpySrc.getGeneration()))
+            .setTarget(dst, BlobTargetOption.doesNotExist())
+            .build();
+
+    CopyWriter copyWriter = s.copy(copyRequest);
+    Blob result = copyWriter.getResult();
+
+    byte[] actualBytes = s.readAllBytes(result.getBlobId());
+    assertThat(actualBytes).isEqualTo(expected);
+  }
 }
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..7bc5cec0
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITServiceAccountTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.NoCredentials;
+import com.google.cloud.storage.ServiceAccount;
+import com.google.cloud.storage.StorageFixture;
+import com.google.cloud.storage.StorageOptions;
+import com.google.cloud.storage.conformance.retry.ParallelParameterized;
+import com.google.cloud.storage.conformance.retry.TestBench;
+import java.util.Arrays;
+import java.util.Collection;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized.Parameters;
+
+@RunWith(ParallelParameterized.class)
+public class ITServiceAccountTest {
+  @ClassRule
+  public static final TestBench TEST_BENCH =
+      TestBench.newBuilder().setContainerName("it-grpc").build();
+
+  @Rule public final StorageFixture storageFixture;
+
+  private static final String SERVICE_ACCOUNT_EMAIL_SUFFIX =
+      "@gs-project-accounts.iam.gserviceaccount.com";
+
+  public ITServiceAccountTest(String clientName, StorageFixture storageFixture) {
+    this.storageFixture = storageFixture;
+  }
+
+  @Parameters(name = "{0}")
+  public static Collection<Object[]> data() {
+    StorageFixture grpcStorageFixture =
+        StorageFixture.from(
+            () ->
+                StorageOptions.grpc()
+                    .setHost(TEST_BENCH.getGRPCBaseUri())
+                    .setCredentials(NoCredentials.getInstance())
+                    .setProjectId("test-project-id")
+                    .build());
+    StorageFixture jsonStorageFixture = StorageFixture.defaultHttp();
+    return Arrays.asList(
+        new Object[] {"JSON/storage.googleapis.com", jsonStorageFixture},
+        new Object[] {"GRPC/" + TEST_BENCH.getGRPCBaseUri(), grpcStorageFixture});
+  }
+
+  @Test
+  public void testGetServiceAccount() {
+    String projectId = storageFixture.getInstance().getOptions().getProjectId();
+    ServiceAccount serviceAccount = storageFixture.getInstance().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));
+    }
+  }
+}