Repository: googleapis/java-storage
Pull request #1563 (2022-08-09 21:04:10)

chore: preliminary implementation of gRPC status code to HTTP Status code translation
Update StorageException#coalesce to account for subclasses of ApiException returned by the gapic client.



Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BackwardCompatibilityUtils.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BackwardCompatibilityUtils.java
index cce2610d..8d862ec5 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BackwardCompatibilityUtils.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BackwardCompatibilityUtils.java
@@ -18,6 +18,8 @@ package com.google.cloud.storage;
 
 import static java.util.Objects.requireNonNull;
 
+import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.api.gax.rpc.StatusCode;
 import com.google.api.services.storage.model.Bucket.Lifecycle.Rule;
 import com.google.cloud.storage.BucketInfo.AgeDeleteRule;
 import com.google.cloud.storage.BucketInfo.CreatedBeforeDeleteRule;
@@ -29,6 +31,8 @@ import com.google.cloud.storage.BucketInfo.LifecycleRule.LifecycleCondition;
 import com.google.cloud.storage.BucketInfo.NumNewerVersionsDeleteRule;
 import com.google.cloud.storage.BucketInfo.RawDeleteRule;
 import com.google.cloud.storage.Conversions.Codec;
+import com.google.common.annotations.VisibleForTesting;
+import io.grpc.Status.Code;
 import java.time.Duration;
 import java.time.Instant;
 import java.time.OffsetDateTime;
@@ -74,6 +78,66 @@ final class BackwardCompatibilityUtils {
 
   private BackwardCompatibilityUtils() {}
 
+  /**
+   * When translating from gRPC Status Codes to the HTTP codes all of our middle ware expects, we
+   * must take care to translate in accordance with the expected retry semantics already outlined
+   * and validated for the JSON implementation. This is why we do not simply use {@link
+   * GrpcStatusCode#of(Code)}{@link GrpcStatusCode#getCode() .getCode}{@link
+   * StatusCode.Code#getHttpStatusCode() .getHttpStatusCode()} as it sometimes returns conflicting
+   * HTTP codes for our retry handling.
+   */
+  @VisibleForTesting
+  static int grpcCodeToHttpStatusCode(Code code) {
+    switch (code) {
+        // 200 Ok
+      case OK:
+        return 200;
+        // 400 Bad Request
+      case INVALID_ARGUMENT:
+      case OUT_OF_RANGE:
+        return 400;
+        // 401 Unauthorized
+      case UNAUTHENTICATED:
+        return 401;
+        // 403 Forbidden
+      case PERMISSION_DENIED:
+        return 403;
+        // 404 Not Found
+      case NOT_FOUND:
+        return 404;
+        // 408 Request Timeout
+        // TODO
+        // 412 Precondition Failed
+      case FAILED_PRECONDITION:
+        return 412;
+        // 409 Conflict
+      case ALREADY_EXISTS:
+        return 409;
+        // 429 Too Many Requests
+      case RESOURCE_EXHAUSTED:
+        return 429;
+        // 500 Internal Server Error
+      case INTERNAL:
+        return 500;
+        // 501 Not Implemented
+      case UNIMPLEMENTED:
+        return 501;
+        // 503 Service Unavailable
+      case UNAVAILABLE:
+        return 503;
+        // 504 Gateway Timeout
+        // TODO
+
+      case ABORTED: // ?
+      case CANCELLED: // ?
+      case UNKNOWN: // ?
+      case DEADLINE_EXCEEDED: // ?
+      case DATA_LOSS: // ?
+      default:
+        return 0;
+    }
+  }
+
   @SuppressWarnings("deprecation")
   private static LifecycleRule deleteRuleEncode(DeleteRule from) {
     if (from instanceof RawDeleteRule) {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java
index 38c1df87..b1811a33 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java
@@ -91,7 +91,7 @@ final class Retrying {
       T result = runWithRetries(c, options.getRetrySettings(), algorithm, options.getClock());
       return result == null ? null : f.decode(result);
     } catch (RetryHelperException e) {
-      throw StorageException.coalesce(e);
+      throw StorageException.coalesce(e.getCause());
     }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
index ed14cac2..07207b07 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
@@ -18,12 +18,18 @@ package com.google.cloud.storage;
 
 import com.google.api.client.googleapis.json.GoogleJsonError;
 import com.google.api.core.InternalApi;
+import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.api.gax.rpc.ApiException;
+import com.google.api.gax.rpc.StatusCode;
 import com.google.cloud.BaseServiceException;
 import com.google.cloud.RetryHelper.RetryHelperException;
 import com.google.cloud.http.BaseHttpServiceException;
 import com.google.common.collect.ImmutableSet;
+import io.grpc.StatusException;
+import io.grpc.StatusRuntimeException;
 import java.io.IOException;
 import java.util.Set;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
  * Storage service exception.
@@ -91,13 +97,39 @@ public final class StorageException extends BaseHttpServiceException {
    * created with an unknown status code.
    */
   static BaseServiceException coalesce(Throwable t) {
-    // TODO: handle com.google.cloud.grpc.BaseGrpcServiceException et. al.
     if (t instanceof BaseServiceException) {
       return (BaseServiceException) t;
     }
     if (t.getCause() instanceof BaseServiceException) {
       return (BaseServiceException) t.getCause();
     }
+    if (t instanceof ApiException) {
+      ApiException apiEx = (ApiException) t;
+
+      // https://cloud.google.com/storage/docs/json_api/v1/status-codes
+      // https://cloud.google.com/apis/design/errors#http_mapping
+      // https://cloud.google.com/apis/design/errors#error_payloads
+      // TODO: flush this out more to wire through "error" and "details" when we're able to get real
+      //   errors from GCS
+      int httpStatusCode = 0;
+      StatusCode statusCode = apiEx.getStatusCode();
+      if (statusCode instanceof GrpcStatusCode) {
+        GrpcStatusCode gsc = (GrpcStatusCode) statusCode;
+        httpStatusCode =
+            BackwardCompatibilityUtils.grpcCodeToHttpStatusCode(gsc.getTransportCode());
+      }
+      // If there is a gRPC exception in our cause change pull it's error message up to be our
+      // message otherwise, create a generic error message with the status code.
+      String message =
+          Utils.firstNonNull(
+              () -> getStatusExceptionMessage(apiEx),
+              () -> String.format("Error: %s", statusCode.getCode().name()));
+
+      // It'd be better to use ExceptionData and BaseServiceException#<init>(ExceptionData) but,
+      // BaseHttpServiceException does not pass that through so we're stuck using this for now.
+      // TODO: When we can break the coupling to BaseHttpServiceException replace this
+      return new StorageException(httpStatusCode, message, apiEx.getReason(), apiEx);
+    }
     return getStorageException(t);
   }
 
@@ -106,7 +138,7 @@ public final class StorageException extends BaseHttpServiceException {
    * defaults to idempotent always being {@code true}. Additionally, this method translates
    * transient issues Connection Closed Prematurely as a retryable error.
    *
-   * @returns {@code StorageException}
+   * @return {@code StorageException}
    */
   public static StorageException translate(IOException exception) {
     String message = exception.getMessage();
@@ -119,4 +151,13 @@ public final class StorageException extends BaseHttpServiceException {
       return new StorageException(exception);
     }
   }
+
+  @Nullable
+  private static String getStatusExceptionMessage(ApiException apiEx) {
+    Throwable cause = apiEx.getCause();
+    if (cause instanceof StatusRuntimeException || cause instanceof StatusException) {
+      return cause.getMessage();
+    }
+    return null;
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java
index 4ae825b0..e3a09e79 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java
@@ -34,7 +34,9 @@ import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.function.Function;
-import javax.annotation.Nullable;
+import java.util.function.Supplier;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
  * A collection of general utility functions providing convenience facilities.
@@ -191,4 +193,22 @@ final class Utils {
   static final <T1, T2> T2 todo(T1 t1) {
     throw new IllegalStateException("Not yet implemented");
   }
+
+  /**
+   * Convenience method to resolve the first non-null {@code T} from an array of suppliers.
+   *
+   * <p>Each supplier will have {@link Supplier#get()} called, and if non-null the value will be
+   * returned.
+   */
+  @NonNull
+  @SafeVarargs
+  static <T> T firstNonNull(Supplier<@Nullable T>... ss) {
+    for (Supplier<T> s : ss) {
+      T t = s.get();
+      if (t != null) {
+        return t;
+      }
+    }
+    throw new IllegalStateException("Unable to resolve non-null value");
+  }
 }


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BackwardCompatibilityUtilsTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BackwardCompatibilityUtilsTest.java
index 46cec8e0..ef3d7985 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BackwardCompatibilityUtilsTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BackwardCompatibilityUtilsTest.java
@@ -30,6 +30,8 @@ import com.google.cloud.storage.BucketInfo.LifecycleRule.LifecycleAction;
 import com.google.cloud.storage.BucketInfo.NumNewerVersionsDeleteRule;
 import com.google.cloud.storage.BucketInfo.RawDeleteRule;
 import com.google.cloud.storage.Conversions.Codec;
+import io.grpc.Status.Code;
+import java.util.EnumMap;
 import org.junit.Test;
 
 public final class BackwardCompatibilityUtilsTest {
@@ -92,6 +94,35 @@ public final class BackwardCompatibilityUtilsTest {
     assertThat(lifecycleRule.getCondition().getAge()).isEqualTo(5);
   }
 
+  @Test
+  public void grpcCodeToHttpStatusCode_expectedMapping() {
+    EnumMap<Code, Integer> expected = new EnumMap<>(Code.class);
+    expected.put(Code.OK, 200);
+    expected.put(Code.INVALID_ARGUMENT, 400);
+    expected.put(Code.OUT_OF_RANGE, 400);
+    expected.put(Code.UNAUTHENTICATED, 401);
+    expected.put(Code.PERMISSION_DENIED, 403);
+    expected.put(Code.NOT_FOUND, 404);
+    expected.put(Code.FAILED_PRECONDITION, 412);
+    expected.put(Code.ALREADY_EXISTS, 409);
+    expected.put(Code.RESOURCE_EXHAUSTED, 429);
+    expected.put(Code.INTERNAL, 500);
+    expected.put(Code.UNIMPLEMENTED, 501);
+    expected.put(Code.UNAVAILABLE, 503);
+    expected.put(Code.ABORTED, 0);
+    expected.put(Code.CANCELLED, 0);
+    expected.put(Code.UNKNOWN, 0);
+    expected.put(Code.DEADLINE_EXCEEDED, 0);
+    expected.put(Code.DATA_LOSS, 0);
+
+    EnumMap<Code, Integer> actual = new EnumMap<>(Code.class);
+    for (Code c : Code.values()) {
+      actual.put(c, BackwardCompatibilityUtils.grpcCodeToHttpStatusCode(c));
+    }
+
+    assertThat(actual).isEqualTo(expected);
+  }
+
   private void verifyConversionRoundTrip(DeleteRule delRule) {
     Rule encode = codec.encode(delRule);
     DeleteRule decode = codec.decode(encode);
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageExceptionGrpcCompatibilityTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageExceptionGrpcCompatibilityTest.java
new file mode 100644
index 00000000..21f6ee39
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageExceptionGrpcCompatibilityTest.java
@@ -0,0 +1,95 @@
+/*
+ * 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.BackwardCompatibilityUtils.grpcCodeToHttpStatusCode;
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.api.gax.rpc.ApiException;
+import com.google.api.gax.rpc.ApiExceptionFactory;
+import com.google.api.gax.rpc.ErrorDetails;
+import com.google.cloud.BaseServiceException;
+import com.google.common.collect.ImmutableList;
+import com.google.protobuf.Any;
+import com.google.rpc.DebugInfo;
+import com.google.rpc.ErrorInfo;
+import io.grpc.Status.Code;
+import io.grpc.StatusRuntimeException;
+import java.util.Arrays;
+import java.util.List;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+@RunWith(Parameterized.class)
+public final class StorageExceptionGrpcCompatibilityTest {
+
+  private final ApiException x;
+
+  public StorageExceptionGrpcCompatibilityTest(Code ignore, ApiException x) {
+    this.x = x;
+  }
+
+  @Test
+  public void testCoalesce() {
+    GrpcStatusCode grpcStatusCode = (GrpcStatusCode) x.getStatusCode();
+    int expectedCode = grpcCodeToHttpStatusCode(grpcStatusCode.getTransportCode());
+
+    BaseServiceException ex = StorageException.coalesce(x);
+    assertThat(ex.getCode()).isEqualTo(expectedCode);
+    assertThat(ex.getReason()).isEqualTo(x.getReason());
+    assertThat(ex.getMessage()).contains(x.getErrorDetails().getDebugInfo().getDetail());
+    assertThat(ex).hasCauseThat().isEqualTo(x);
+  }
+
+  @Parameters(name = "{0}")
+  public static List<Object[]> args() {
+    return Arrays.stream(Code.values())
+        .map(Code::toStatus)
+        .filter(s -> !s.isOk())
+        .map(
+            status -> {
+              Code code = status.getCode();
+
+              GrpcStatusCode statusCode = GrpcStatusCode.of(code);
+              ErrorInfo errorInfo =
+                  ErrorInfo.newBuilder()
+                      .setReason("reason")
+                      .setDomain("global")
+                      .putMetadata("errors", "x")
+                      .build();
+
+              DebugInfo debugInfo =
+                  DebugInfo.newBuilder()
+                      .setDetail(
+                          "bw-storage-dev-region-fine@default-223119.iam.gserviceaccount.com does not have storage.hmacKeys.list access to the Google Cloud project.")
+                      .build();
+
+              ImmutableList<Any> anys = ImmutableList.of(Any.pack(errorInfo), Any.pack(debugInfo));
+              ErrorDetails errorDetails = ErrorDetails.builder().setRawErrorMessages(anys).build();
+
+              StatusRuntimeException cause =
+                  new StatusRuntimeException(status.withDescription(debugInfo.getDetail()));
+              return new Object[] {
+                code, ApiExceptionFactory.createException(cause, statusCode, false, errorDetails)
+              };
+            })
+        .collect(ImmutableList.toImmutableList());
+  }
+}
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 668e0472..936fe800 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
@@ -33,20 +33,26 @@ import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.DataGeneration;
+import com.google.cloud.storage.HttpStorageOptions;
 import com.google.cloud.storage.PackagePrivateMethodWorkarounds;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
 import com.google.cloud.storage.conformance.retry.TestBench;
 import com.google.cloud.storage.conformance.retry.TestBench.RetryTestResource;
+import com.google.cloud.storage.spi.StorageRpcFactory;
 import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.cloud.storage.spi.v1.StorageRpc.Option;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.reflect.AbstractInvocationHandler;
 import com.google.common.reflect.Reflection;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Optional;
 import java.util.Random;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Logger;
 import org.junit.ClassRule;
 import org.junit.Rule;
@@ -101,6 +107,24 @@ public final class ITBlobWriteChannelTest {
     doJsonUnexpectedEOFTest(contentSize, cappedByteCount);
   }
 
+  @Test
+  public void blobWriteChannel_handlesRecoveryOnLastChunkWhenGenerationIsPresent_multipleChunks()
+      throws IOException {
+    int _2MiB = 256 * 1024;
+    int contentSize = 292_617;
+
+    blobWriteChannel_handlesRecoveryOnLastChunkWhenGenerationIsPresent(_2MiB, contentSize);
+  }
+
+  @Test
+  public void blobWriteChannel_handlesRecoveryOnLastChunkWhenGenerationIsPresent_singleChunk()
+      throws IOException {
+    int _4MiB = 256 * 1024 * 2;
+    int contentSize = 292_617;
+
+    blobWriteChannel_handlesRecoveryOnLastChunkWhenGenerationIsPresent(_4MiB, contentSize);
+  }
+
   private void doJsonUnexpectedEOFTest(int contentSize, int cappedByteCount) throws IOException {
     String blobPath = String.format("%s/%s/blob", testName.getMethodName(), NOW_STRING);
 
@@ -193,4 +217,101 @@ public final class ITBlobWriteChannelTest {
     ByteBuffer actual = ByteBuffer.wrap(actualData.toByteArray());
     assertEquals(expected, actual);
   }
+
+  private void blobWriteChannel_handlesRecoveryOnLastChunkWhenGenerationIsPresent(
+      int chunkSize, int contentSize) throws IOException {
+    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();
+
+    ByteBuffer contentGen1 = dataGeneration.randByteBuffer(contentSize);
+    ByteBuffer contentGen2 = dataGeneration.randByteBuffer(contentSize);
+    ByteBuffer contentGen2Expected = contentGen2.duplicate();
+    HttpStorageOptions baseStorageOptions =
+        StorageOptions.http()
+            .setCredentials(NoCredentials.getInstance())
+            .setHost(testBench.getBaseUri())
+            .setProjectId("test-project-id")
+            .build();
+    Storage storage = baseStorageOptions.getService();
+    storage.create(bucketInfo);
+    WriteChannel ww = storage.writer(blobInfo);
+    ww.setChunkSize(chunkSize);
+    ww.write(contentGen1);
+    ww.close();
+
+    Blob blobGen1 = storage.get(blobId);
+
+    final AtomicBoolean exceptionThrown = new AtomicBoolean(false);
+
+    Storage testStorage =
+        baseStorageOptions
+            .toBuilder()
+            .setServiceRpcFactory(
+                new StorageRpcFactory() {
+                  /**
+                   * Here we're creating a proxy of StorageRpc where we can delegate all calls to
+                   * the normal implementation, except in the case of {@link
+                   * StorageRpc#writeWithResponse(String, byte[], int, long, int, boolean)} where
+                   * {@code lastChunk == true}. We allow the call to execute, but instead of
+                   * returning the result we throw an IOException to simulate a prematurely close
+                   * connection. This behavior is to ensure appropriate handling of a completed
+                   * upload where the ACK wasn't received. In particular, if an upload is initiated
+                   * against an object where an {@link Option#IF_GENERATION_MATCH} simply calling
+                   * get on an object can result in a 404 because the object that is created while
+                   * the BlobWriteChannel is executing will be a new generation.
+                   */
+                  @SuppressWarnings("UnstableApiUsage")
+                  @Override
+                  public StorageRpc create(final StorageOptions options) {
+                    return Reflection.newProxy(
+                        StorageRpc.class,
+                        new AbstractInvocationHandler() {
+                          final StorageRpc delegate = (StorageRpc) baseStorageOptions.getRpc();
+
+                          @Override
+                          protected Object handleInvocation(
+                              Object proxy, java.lang.reflect.Method method, Object[] args)
+                              throws Throwable {
+                            if ("writeWithResponse".equals(method.getName())) {
+                              Object result = method.invoke(delegate, args);
+                              boolean lastChunk = (boolean) args[5];
+                              // if we're on the lastChunk simulate a connection failure which
+                              // happens after the request was processed but before response could
+                              // be received by the client.
+                              if (lastChunk) {
+                                exceptionThrown.set(true);
+                                throw StorageException.translate(
+                                    new IOException("simulated Connection closed prematurely"));
+                              } else {
+                                return result;
+                              }
+                            }
+                            return method.invoke(delegate, args);
+                          }
+                        });
+                  }
+                })
+            .build()
+            .getService();
+    try (WriteChannel w = testStorage.writer(blobGen1, BlobWriteOption.generationMatch())) {
+      w.setChunkSize(chunkSize);
+
+      w.write(contentGen2);
+    }
+
+    assertTrue("Expected an exception to be thrown for the last chunk", exceptionThrown.get());
+
+    Blob blobGen2 = storage.get(blobId);
+    assertEquals(contentSize, (long) blobGen2.getSize());
+    assertNotEquals(blobInfo.getGeneration(), blobGen2.getGeneration());
+    ByteArrayOutputStream actualData = new ByteArrayOutputStream();
+    blobGen2.downloadTo(actualData);
+    assertEquals(contentGen2Expected, ByteBuffer.wrap(actualData.toByteArray()));
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITKmsTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITKmsTest.java
new file mode 100644
index 00000000..4a857400
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITKmsTest.java
@@ -0,0 +1,546 @@
+/*
+ * 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 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.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import com.google.api.gax.paging.Page;
+import com.google.auth.Credentials;
+import com.google.cloud.WriteChannel;
+import com.google.cloud.kms.v1.CreateCryptoKeyRequest;
+import com.google.cloud.kms.v1.CreateKeyRingRequest;
+import com.google.cloud.kms.v1.CryptoKey;
+import com.google.cloud.kms.v1.CryptoKeyName;
+import com.google.cloud.kms.v1.GetCryptoKeyRequest;
+import com.google.cloud.kms.v1.GetKeyRingRequest;
+import com.google.cloud.kms.v1.KeyManagementServiceGrpc;
+import com.google.cloud.kms.v1.KeyManagementServiceGrpc.KeyManagementServiceBlockingStub;
+import com.google.cloud.kms.v1.KeyRingName;
+import com.google.cloud.kms.v1.LocationName;
+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.ServiceAccount;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobField;
+import com.google.cloud.storage.Storage.BucketField;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageFixture;
+import com.google.cloud.storage.testing.RemoteStorageHelper;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterators;
+import com.google.common.io.BaseEncoding;
+import com.google.iam.v1.Binding;
+import com.google.iam.v1.IAMPolicyGrpc;
+import com.google.iam.v1.SetIamPolicyRequest;
+import io.grpc.ClientInterceptor;
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import io.grpc.Metadata;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.grpc.auth.MoreCallCredentials;
+import io.grpc.stub.MetadataUtils;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.Key;
+import java.util.Iterator;
+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 javax.crypto.spec.SecretKeySpec;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class ITKmsTest {
+
+  @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 long seed = -7071346537822433445L;
+  @Rule public final DataGeneration dataGeneration = new DataGeneration(new Random(seed));
+
+  private static String kmsKeyOneResourcePath;
+  private static String kmsKeyTwoResourcePath;
+  private static final String KMS_KEY_RING_NAME = "gcs_test_kms_key_ring";
+  private static final String KMS_KEY_RING_LOCATION = "us";
+  private static final String KMS_KEY_ONE_NAME = "gcs_kms_key_one";
+  private static final String KMS_KEY_TWO_NAME = "gcs_kms_key_two";
+  private static Metadata requestParamsHeader = new Metadata();
+  private static Metadata.Key<String> requestParamsKey =
+      Metadata.Key.of("x-goog-request-params", Metadata.ASCII_STRING_MARSHALLER);
+  private static ManagedChannel kmsChannel;
+  private static final Logger log = Logger.getLogger(ITKmsTest.class.getName());
+  private static final byte[] BLOB_BYTE_CONTENT = {0xD, 0xE, 0xA, 0xD};
+  private static final String BLOB_STRING_CONTENT = "Hello Google Cloud Storage!";
+  private static final String BASE64_KEY = "JVzfVl8NLD9FjedFuStegjRfES5ll5zc59CIXw572OA=";
+  private static final Key KEY =
+      new SecretKeySpec(BaseEncoding.base64().decode(BASE64_KEY), "AES256");
+  private static final String CONTENT_TYPE = "text/plain";
+
+  private static Storage storage;
+  private static String bucketName;
+
+  @BeforeClass
+  public static void setup() throws IOException {
+    storage = storageFixture.getInstance();
+    bucketName = bucketFixture.getBucketInfo().getName();
+    // Prepare KMS KeyRing for CMEK tests
+    // https://cloud.google.com/storage/docs/encryption/using-customer-managed-keys
+    String projectId = storage.getOptions().getProjectId();
+    Credentials credentials = storage.getOptions().getCredentials();
+    kmsChannel = ManagedChannelBuilder.forTarget("cloudkms.googleapis.com:443").build();
+    KeyManagementServiceBlockingStub kmsStub =
+        KeyManagementServiceGrpc.newBlockingStub(kmsChannel)
+            .withCallCredentials(MoreCallCredentials.from(credentials));
+    IAMPolicyGrpc.IAMPolicyBlockingStub iamStub =
+        IAMPolicyGrpc.newBlockingStub(kmsChannel)
+            .withCallCredentials(MoreCallCredentials.from(credentials));
+    ensureKmsKeyRingExistsForTests(kmsStub, projectId, KMS_KEY_RING_LOCATION, KMS_KEY_RING_NAME);
+    ensureKmsKeyRingIamPermissionsForTests(
+        iamStub, projectId, KMS_KEY_RING_LOCATION, KMS_KEY_RING_NAME);
+    kmsKeyOneResourcePath =
+        ensureKmsKeyExistsForTests(
+            kmsStub, projectId, KMS_KEY_RING_LOCATION, KMS_KEY_RING_NAME, KMS_KEY_ONE_NAME);
+    kmsKeyTwoResourcePath =
+        ensureKmsKeyExistsForTests(
+            kmsStub, projectId, KMS_KEY_RING_LOCATION, KMS_KEY_RING_NAME, KMS_KEY_TWO_NAME);
+  }
+
+  @AfterClass
+  public static void afterClass() {
+    if (kmsChannel != null) {
+      try {
+        kmsChannel.shutdownNow();
+      } catch (Exception e) {
+        log.log(Level.WARNING, "Error while trying to shutdown kms channel", e);
+      }
+      kmsChannel = null;
+    }
+  }
+
+  private static String ensureKmsKeyRingExistsForTests(
+      KeyManagementServiceBlockingStub kmsStub,
+      String projectId,
+      String location,
+      String keyRingName)
+      throws StatusRuntimeException {
+    String kmsKeyRingResourcePath = KeyRingName.of(projectId, location, keyRingName).toString();
+    try {
+      // Attempt to Get KeyRing
+      GetKeyRingRequest getKeyRingRequest =
+          GetKeyRingRequest.newBuilder().setName(kmsKeyRingResourcePath).build();
+      requestParamsHeader.put(requestParamsKey, "name=" + kmsKeyRingResourcePath);
+      ClientInterceptor headersInterceptor =
+          MetadataUtils.newAttachHeadersInterceptor(requestParamsHeader);
+      KeyManagementServiceBlockingStub stubForGetKeyRing =
+          kmsStub.withInterceptors(headersInterceptor);
+      stubForGetKeyRing.getKeyRing(getKeyRingRequest);
+    } catch (StatusRuntimeException ex) {
+      if (ex.getStatus().getCode() == Status.Code.NOT_FOUND) {
+        // Create KmsKeyRing
+        String keyRingParent = LocationName.of(projectId, location).toString();
+        CreateKeyRingRequest createKeyRingRequest =
+            CreateKeyRingRequest.newBuilder()
+                .setParent(keyRingParent)
+                .setKeyRingId(keyRingName)
+                .build();
+        requestParamsHeader.put(requestParamsKey, "parent=" + keyRingParent);
+        ClientInterceptor headersInterceptor =
+            MetadataUtils.newAttachHeadersInterceptor(requestParamsHeader);
+        KeyManagementServiceBlockingStub stubForCreateKeyRing =
+            kmsStub.withInterceptors(headersInterceptor);
+        stubForCreateKeyRing.createKeyRing(createKeyRingRequest);
+      } else {
+        throw ex;
+      }
+    }
+
+    return kmsKeyRingResourcePath;
+  }
+
+  private static void ensureKmsKeyRingIamPermissionsForTests(
+      IAMPolicyGrpc.IAMPolicyBlockingStub iamStub,
+      String projectId,
+      String location,
+      String keyRingName)
+      throws StatusRuntimeException {
+    ServiceAccount serviceAccount = storage.getServiceAccount(projectId);
+    String kmsKeyRingResourcePath = KeyRingName.of(projectId, location, keyRingName).toString();
+    Binding binding =
+        Binding.newBuilder()
+            .setRole("roles/cloudkms.cryptoKeyEncrypterDecrypter")
+            .addMembers("serviceAccount:" + serviceAccount.getEmail())
+            .build();
+    com.google.iam.v1.Policy policy =
+        com.google.iam.v1.Policy.newBuilder().addBindings(binding).build();
+    SetIamPolicyRequest setIamPolicyRequest =
+        SetIamPolicyRequest.newBuilder()
+            .setResource(kmsKeyRingResourcePath)
+            .setPolicy(policy)
+            .build();
+    requestParamsHeader.put(requestParamsKey, "parent=" + kmsKeyRingResourcePath);
+    ClientInterceptor headersInterceptor =
+        MetadataUtils.newAttachHeadersInterceptor(requestParamsHeader);
+    iamStub = iamStub.withInterceptors(headersInterceptor);
+    try {
+      iamStub.setIamPolicy(setIamPolicyRequest);
+    } catch (StatusRuntimeException e) {
+      if (log.isLoggable(Level.WARNING)) {
+        log.log(Level.WARNING, "Unable to set IAM policy: {0}", e.getMessage());
+      }
+    }
+  }
+
+  private static String ensureKmsKeyExistsForTests(
+      KeyManagementServiceBlockingStub kmsStub,
+      String projectId,
+      String location,
+      String keyRingName,
+      String keyName)
+      throws StatusRuntimeException {
+    String kmsKeyResourcePath =
+        CryptoKeyName.of(projectId, location, keyRingName, keyName).toString();
+    try {
+      // Attempt to Get CryptoKey
+      requestParamsHeader.put(requestParamsKey, "name=" + kmsKeyResourcePath);
+      GetCryptoKeyRequest getCryptoKeyRequest =
+          GetCryptoKeyRequest.newBuilder().setName(kmsKeyResourcePath).build();
+      ClientInterceptor headersInterceptor =
+          MetadataUtils.newAttachHeadersInterceptor(requestParamsHeader);
+      KeyManagementServiceGrpc.KeyManagementServiceBlockingStub stubForGetCryptoKey =
+          kmsStub.withInterceptors(headersInterceptor);
+      stubForGetCryptoKey.getCryptoKey(getCryptoKeyRequest);
+    } catch (StatusRuntimeException ex) {
+      if (ex.getStatus().getCode() == Status.Code.NOT_FOUND) {
+        String kmsKeyRingResourcePath = KeyRingName.of(projectId, location, keyRingName).toString();
+        CryptoKey cryptoKey =
+            CryptoKey.newBuilder().setPurpose(CryptoKey.CryptoKeyPurpose.ENCRYPT_DECRYPT).build();
+        CreateCryptoKeyRequest createCryptoKeyRequest =
+            CreateCryptoKeyRequest.newBuilder()
+                .setCryptoKeyId(keyName)
+                .setParent(kmsKeyRingResourcePath)
+                .setCryptoKey(cryptoKey)
+                .build();
+
+        requestParamsHeader.put(requestParamsKey, "parent=" + kmsKeyRingResourcePath);
+        ClientInterceptor headersInterceptor =
+            MetadataUtils.newAttachHeadersInterceptor(requestParamsHeader);
+        KeyManagementServiceGrpc.KeyManagementServiceBlockingStub stubForCreateCryptoKey =
+            kmsStub.withInterceptors(headersInterceptor);
+        stubForCreateCryptoKey.createCryptoKey(createCryptoKeyRequest);
+      } else {
+        throw ex;
+      }
+    }
+    return kmsKeyResourcePath;
+  }
+
+  @Test
+  public void testClearBucketDefaultKmsKeyName() throws ExecutionException, InterruptedException {
+    String bucketName = dataGeneration.getBucketName();
+    Bucket remoteBucket =
+        storage.create(
+            BucketInfo.newBuilder(bucketName)
+                .setDefaultKmsKeyName(kmsKeyOneResourcePath)
+                .setLocation(KMS_KEY_RING_LOCATION)
+                .build());
+    try {
+      assertEquals(kmsKeyOneResourcePath, remoteBucket.getDefaultKmsKeyName());
+      Bucket updatedBucket = remoteBucket.toBuilder().setDefaultKmsKeyName(null).build().update();
+      assertNull(updatedBucket.getDefaultKmsKeyName());
+    } finally {
+      RemoteStorageHelper.forceDelete(storage, bucketName, 5, TimeUnit.SECONDS);
+    }
+  }
+
+  @Test
+  public void testUpdateBucketDefaultKmsKeyName() throws ExecutionException, InterruptedException {
+    String bucketName = RemoteStorageHelper.generateBucketName();
+    Bucket remoteBucket =
+        storage.create(
+            BucketInfo.newBuilder(bucketName)
+                .setDefaultKmsKeyName(kmsKeyOneResourcePath)
+                .setLocation(KMS_KEY_RING_LOCATION)
+                .build());
+
+    try {
+      assertEquals(kmsKeyOneResourcePath, remoteBucket.getDefaultKmsKeyName());
+      Bucket updatedBucket =
+          remoteBucket.toBuilder().setDefaultKmsKeyName(kmsKeyTwoResourcePath).build().update();
+      assertEquals(kmsKeyTwoResourcePath, updatedBucket.getDefaultKmsKeyName());
+    } finally {
+      RemoteStorageHelper.forceDelete(storage, bucketName, 5, TimeUnit.SECONDS);
+    }
+  }
+
+  @Test
+  public void testCreateBlobWithKmsKeyName() {
+    String blobName = "test-create-with-kms-key-name-blob";
+    BlobInfo blob = BlobInfo.newBuilder(bucketName, blobName).build();
+    Blob remoteBlob =
+        storage.create(
+            blob, BLOB_BYTE_CONTENT, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
+    assertNotNull(remoteBlob);
+    assertEquals(blob.getBucket(), remoteBlob.getBucket());
+    assertEquals(blob.getName(), remoteBlob.getName());
+    assertNotNull(remoteBlob.getKmsKeyName());
+    assertTrue(remoteBlob.getKmsKeyName().startsWith(kmsKeyOneResourcePath));
+    byte[] readBytes = storage.readAllBytes(bucketName, blobName);
+    assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
+  }
+
+  @Test(expected = StorageException.class)
+  public void testCreateBlobWithKmsKeyNameAndCustomerSuppliedKeyFails() {
+    String blobName = "test-create-with-kms-key-name-blob";
+    BlobInfo blob = BlobInfo.newBuilder(bucketName, blobName).build();
+    storage.create(
+        blob,
+        BLOB_BYTE_CONTENT,
+        Storage.BlobTargetOption.encryptionKey(KEY),
+        Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
+  }
+
+  @Test
+  public void testCreateBlobWithDefaultKmsKeyName()
+      throws ExecutionException, InterruptedException {
+    String bucketName = RemoteStorageHelper.generateBucketName();
+    Bucket bucket =
+        storage.create(
+            BucketInfo.newBuilder(bucketName)
+                .setDefaultKmsKeyName(kmsKeyOneResourcePath)
+                .setLocation(KMS_KEY_RING_LOCATION)
+                .build());
+    assertEquals(bucket.getDefaultKmsKeyName(), kmsKeyOneResourcePath);
+
+    try {
+      String blobName = "test-create-with-default-kms-key-name-blob";
+      BlobInfo blob = BlobInfo.newBuilder(bucket, blobName).build();
+      Blob remoteBlob = storage.create(blob, BLOB_BYTE_CONTENT);
+      assertNotNull(remoteBlob);
+      assertEquals(blob.getBucket(), remoteBlob.getBucket());
+      assertEquals(blob.getName(), remoteBlob.getName());
+      assertNotNull(remoteBlob.getKmsKeyName());
+      assertTrue(remoteBlob.getKmsKeyName().startsWith(kmsKeyOneResourcePath));
+      byte[] readBytes = storage.readAllBytes(bucketName, blobName);
+      assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
+    } finally {
+      RemoteStorageHelper.forceDelete(storage, bucketName, 5, TimeUnit.SECONDS);
+    }
+  }
+
+  @Test
+  public void testGetBlobKmsKeyNameField() {
+    String blobName = "test-get-selected-kms-key-name-field-blob";
+    BlobInfo blob = BlobInfo.newBuilder(bucketName, blobName).setContentType(CONTENT_TYPE).build();
+    assertNotNull(storage.create(blob, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath)));
+    Blob remoteBlob =
+        storage.get(blob.getBlobId(), Storage.BlobGetOption.fields(BlobField.KMS_KEY_NAME));
+    assertEquals(blob.getBlobId(), remoteBlob.getBlobId());
+    assertTrue(remoteBlob.getKmsKeyName().startsWith(kmsKeyOneResourcePath));
+    assertNull(remoteBlob.getContentType());
+  }
+
+  @Test(timeout = 5000)
+  public void testListBlobsKmsKeySelectedFields() throws InterruptedException {
+    String[] blobNames = {
+      "test-list-blobs-selected-field-kms-key-name-blob1",
+      "test-list-blobs-selected-field-kms-key-name-blob2"
+    };
+    BlobInfo blob1 =
+        BlobInfo.newBuilder(bucketName, blobNames[0]).setContentType(CONTENT_TYPE).build();
+    BlobInfo blob2 =
+        BlobInfo.newBuilder(bucketName, blobNames[1]).setContentType(CONTENT_TYPE).build();
+    Blob remoteBlob1 =
+        storage.create(blob1, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
+    Blob remoteBlob2 =
+        storage.create(blob2, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
+    assertNotNull(remoteBlob1);
+    assertNotNull(remoteBlob2);
+    Page<Blob> page =
+        storage.list(
+            bucketName,
+            Storage.BlobListOption.prefix("test-list-blobs-selected-field-kms-key-name-blob"),
+            Storage.BlobListOption.fields(BlobField.KMS_KEY_NAME));
+    // Listing blobs is eventually consistent, we loop until the list is of the expected size. The
+    // test fails if timeout is reached.
+    while (Iterators.size(page.iterateAll().iterator()) != 2) {
+      Thread.sleep(500);
+      page =
+          storage.list(
+              bucketName,
+              Storage.BlobListOption.prefix("test-list-blobs-selected-field-kms-key-name-blob"),
+              Storage.BlobListOption.fields(BlobField.KMS_KEY_NAME));
+    }
+    Set<String> blobSet = ImmutableSet.of(blobNames[0], blobNames[1]);
+    Iterator<Blob> iterator = page.iterateAll().iterator();
+    while (iterator.hasNext()) {
+      Blob remoteBlob = iterator.next();
+      assertEquals(bucketName, remoteBlob.getBucket());
+      assertTrue(blobSet.contains(remoteBlob.getName()));
+      assertTrue(remoteBlob.getKmsKeyName().startsWith(kmsKeyOneResourcePath));
+      assertNull(remoteBlob.getContentType());
+    }
+  }
+
+  @Test
+  public void testRotateFromCustomerEncryptionToKmsKey() {
+    String sourceBlobName = "test-copy-blob-encryption-key-source";
+    BlobId source = BlobId.of(bucketName, sourceBlobName);
+    ImmutableMap<String, String> metadata = ImmutableMap.of("k", "v");
+    Blob remoteBlob =
+        storage.create(
+            BlobInfo.newBuilder(source).build(),
+            BLOB_BYTE_CONTENT,
+            Storage.BlobTargetOption.encryptionKey(KEY));
+    assertNotNull(remoteBlob);
+    String targetBlobName = "test-copy-blob-kms-key-target";
+    BlobInfo target =
+        BlobInfo.newBuilder(bucketName, targetBlobName)
+            .setContentType(CONTENT_TYPE)
+            .setMetadata(metadata)
+            .build();
+    Storage.CopyRequest req =
+        Storage.CopyRequest.newBuilder()
+            .setSource(source)
+            .setSourceOptions(Storage.BlobSourceOption.decryptionKey(BASE64_KEY))
+            .setTarget(target, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath))
+            .build();
+    CopyWriter copyWriter = storage.copy(req);
+    assertEquals(bucketName, copyWriter.getResult().getBucket());
+    assertEquals(targetBlobName, copyWriter.getResult().getName());
+    assertEquals(CONTENT_TYPE, copyWriter.getResult().getContentType());
+    assertNotNull(copyWriter.getResult().getKmsKeyName());
+    assertTrue(copyWriter.getResult().getKmsKeyName().startsWith(kmsKeyOneResourcePath));
+    assertArrayEquals(BLOB_BYTE_CONTENT, copyWriter.getResult().getContent());
+    assertEquals(metadata, copyWriter.getResult().getMetadata());
+    assertTrue(copyWriter.isDone());
+    assertTrue(storage.delete(bucketName, targetBlobName));
+  }
+
+  @Test(expected = StorageException.class)
+  public void testRotateFromCustomerEncryptionToKmsKeyWithCustomerEncryption() {
+    String sourceBlobName = "test-copy-blob-encryption-key-source";
+    BlobId source = BlobId.of(bucketName, sourceBlobName);
+    ImmutableMap<String, String> metadata = ImmutableMap.of("k", "v");
+    Blob remoteBlob =
+        storage.create(
+            BlobInfo.newBuilder(source).build(),
+            BLOB_BYTE_CONTENT,
+            Storage.BlobTargetOption.encryptionKey(KEY));
+    assertNotNull(remoteBlob);
+    String targetBlobName = "test-copy-blob-kms-key-target";
+    BlobInfo target =
+        BlobInfo.newBuilder(bucketName, targetBlobName)
+            .setContentType(CONTENT_TYPE)
+            .setMetadata(metadata)
+            .build();
+    Storage.CopyRequest req =
+        Storage.CopyRequest.newBuilder()
+            .setSource(source)
+            .setSourceOptions(Storage.BlobSourceOption.decryptionKey(BASE64_KEY))
+            .setTarget(
+                target,
+                Storage.BlobTargetOption.encryptionKey(KEY),
+                Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath))
+            .build();
+    storage.copy(req);
+  }
+
+  @Test
+  public void testListBucketDefaultKmsKeyName() throws ExecutionException, InterruptedException {
+    String bucketName = RemoteStorageHelper.generateBucketName();
+    Bucket remoteBucket =
+        storage.create(
+            BucketInfo.newBuilder(bucketName)
+                .setDefaultKmsKeyName(kmsKeyOneResourcePath)
+                .setLocation(KMS_KEY_RING_LOCATION)
+                .build());
+    assertNotNull(remoteBucket);
+    assertTrue(remoteBucket.getDefaultKmsKeyName().startsWith(kmsKeyOneResourcePath));
+    try {
+      Iterator<Bucket> bucketIterator =
+          storage
+              .list(
+                  Storage.BucketListOption.prefix(bucketName),
+                  Storage.BucketListOption.fields(BucketField.ENCRYPTION))
+              .iterateAll()
+              .iterator();
+      while (!bucketIterator.hasNext()) {
+        Thread.sleep(500);
+        bucketIterator =
+            storage
+                .list(
+                    Storage.BucketListOption.prefix(bucketName),
+                    Storage.BucketListOption.fields(BucketField.ENCRYPTION))
+                .iterateAll()
+                .iterator();
+      }
+      while (bucketIterator.hasNext()) {
+        Bucket bucket = bucketIterator.next();
+        assertTrue(bucket.getName().startsWith(bucketName));
+        assertNotNull(bucket.getDefaultKmsKeyName());
+        assertTrue(bucket.getDefaultKmsKeyName().startsWith(kmsKeyOneResourcePath));
+        assertNull(bucket.getCreateTimeOffsetDateTime());
+        assertNull(bucket.getSelfLink());
+      }
+    } finally {
+      RemoteStorageHelper.forceDelete(storage, bucketName, 5, TimeUnit.SECONDS);
+    }
+  }
+
+  @Test
+  public void testWriterWithKmsKeyName() throws IOException {
+    // Write an empty object with a kmsKeyName.
+    String blobName = "test-empty-blob";
+    BlobInfo blobInfo = BlobInfo.newBuilder(bucketName, blobName).build();
+    Blob blob =
+        storage.create(blobInfo, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
+
+    // Create a writer using blob that already has metadata received from Storage API.
+    int numberOfBytes;
+    byte[] content = BLOB_STRING_CONTENT.getBytes(UTF_8);
+    try (WriteChannel writer = blob.writer()) {
+      numberOfBytes = writer.write(ByteBuffer.wrap(content, 0, content.length));
+    }
+    assertThat(numberOfBytes).isEqualTo(content.length);
+    assertThat(blob.getKmsKeyName()).isNotNull();
+    assertThat(storage.delete(bucketName, blobName)).isTrue();
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITSignedUrlTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITSignedUrlTest.java
new file mode 100644
index 00000000..24d1d7b9
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITSignedUrlTest.java
@@ -0,0 +1,249 @@
+/*
+ * 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.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeTrue;
+
+import com.google.auth.ServiceAccountSigner;
+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.HttpMethod;
+import com.google.cloud.storage.PostPolicyV4;
+import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageFixture;
+import com.google.cloud.storage.StorageOptions;
+import com.google.common.collect.ImmutableMap;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.mime.MultipartEntityBuilder;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+public class ITSignedUrlTest {
+  @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 Storage storage;
+  private static String bucketName;
+  private static final byte[] BLOB_BYTE_CONTENT = {0xD, 0xE, 0xA, 0xD};
+  private static final String BLOB_STRING_CONTENT = "Hello Google Cloud Storage!";
+
+  @BeforeClass
+  public static void beforeClass() throws IOException {
+    storage = storageFixture.getInstance();
+    bucketName = bucketFixture.getBucketInfo().getName();
+  }
+
+  @Test
+  public void testGetSignedUrl() throws IOException {
+    if (storage.getOptions().getCredentials() != null) {
+      assumeTrue(storage.getOptions().getCredentials() instanceof ServiceAccountSigner);
+    }
+    String blobName = "test-get-signed-url-blob/with/slashes/and?special=!#$&'()*+,:;=?@[]";
+    BlobInfo blob = BlobInfo.newBuilder(bucketName, blobName).build();
+    Blob remoteBlob = storage.create(blob, BLOB_BYTE_CONTENT);
+    assertNotNull(remoteBlob);
+    for (Storage.SignUrlOption urlStyle :
+        Arrays.asList(
+            Storage.SignUrlOption.withPathStyle(),
+            Storage.SignUrlOption.withVirtualHostedStyle())) {
+      URL url = storage.signUrl(blob, 1, TimeUnit.HOURS, urlStyle);
+      URLConnection connection = url.openConnection();
+      byte[] readBytes = new byte[BLOB_BYTE_CONTENT.length];
+      try (InputStream responseStream = connection.getInputStream()) {
+        assertEquals(BLOB_BYTE_CONTENT.length, responseStream.read(readBytes));
+        assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
+      }
+    }
+  }
+
+  @Test
+  public void testGetV2SignedUrlWithAddlQueryParam() throws IOException {
+    if (storage.getOptions().getCredentials() != null) {
+      assumeTrue(storage.getOptions().getCredentials() instanceof ServiceAccountSigner);
+    }
+    String blobName = "test-get-v2-with-generation-param";
+    BlobInfo blob = BlobInfo.newBuilder(bucketName, blobName).build();
+    Blob remoteBlob = storage.create(blob, BLOB_BYTE_CONTENT);
+    assertNotNull(remoteBlob);
+    for (Storage.SignUrlOption urlStyle :
+        Arrays.asList(
+            Storage.SignUrlOption.withPathStyle(),
+            Storage.SignUrlOption.withVirtualHostedStyle())) {
+      String generationStr = remoteBlob.getGeneration().toString();
+      URL url =
+          storage.signUrl(
+              blob,
+              1,
+              TimeUnit.HOURS,
+              urlStyle,
+              Storage.SignUrlOption.withV2Signature(),
+              Storage.SignUrlOption.withQueryParams(ImmutableMap.of("generation", generationStr)));
+      // Finally, verify that the URL works and we can get the object as expected:
+      URLConnection connection = url.openConnection();
+      byte[] readBytes = new byte[BLOB_BYTE_CONTENT.length];
+      try (InputStream responseStream = connection.getInputStream()) {
+        assertEquals(BLOB_BYTE_CONTENT.length, responseStream.read(readBytes));
+        assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
+      }
+    }
+  }
+
+  @Test
+  public void testPostSignedUrl() throws IOException {
+    if (storage.getOptions().getCredentials() != null) {
+      assumeTrue(storage.getOptions().getCredentials() instanceof ServiceAccountSigner);
+    }
+    String blobName = "test-post-signed-url-blob";
+    BlobInfo blob = BlobInfo.newBuilder(bucketName, blobName).build();
+    assertNotNull(storage.create(blob));
+    for (Storage.SignUrlOption urlStyle :
+        Arrays.asList(
+            Storage.SignUrlOption.withPathStyle(),
+            Storage.SignUrlOption.withVirtualHostedStyle())) {
+
+      URL url =
+          storage.signUrl(
+              blob, 1, TimeUnit.HOURS, Storage.SignUrlOption.httpMethod(HttpMethod.POST), urlStyle);
+      URLConnection connection = url.openConnection();
+      connection.setDoOutput(true);
+      connection.connect();
+      Blob remoteBlob = storage.get(bucketName, blobName);
+      assertNotNull(remoteBlob);
+      assertEquals(blob.getBucket(), remoteBlob.getBucket());
+      assertEquals(blob.getName(), remoteBlob.getName());
+    }
+  }
+
+  @Test
+  public void testV4SignedUrl() throws IOException {
+    if (storage.getOptions().getCredentials() != null) {
+      assumeTrue(storage.getOptions().getCredentials() instanceof ServiceAccountSigner);
+    }
+
+    String blobName = "test-get-signed-url-blob/with/slashes/and?special=!#$&'()*+,:;=?@[]";
+    BlobInfo blob = BlobInfo.newBuilder(bucketName, blobName).build();
+    Blob remoteBlob = storage.create(blob, BLOB_BYTE_CONTENT);
+    assertNotNull(remoteBlob);
+    for (Storage.SignUrlOption urlStyle :
+        Arrays.asList(
+            Storage.SignUrlOption.withPathStyle(),
+            Storage.SignUrlOption.withVirtualHostedStyle())) {
+
+      URL url =
+          storage.signUrl(
+              blob, 1, TimeUnit.HOURS, Storage.SignUrlOption.withV4Signature(), urlStyle);
+      URLConnection connection = url.openConnection();
+      byte[] readBytes = new byte[BLOB_BYTE_CONTENT.length];
+      try (InputStream responseStream = connection.getInputStream()) {
+        assertEquals(BLOB_BYTE_CONTENT.length, responseStream.read(readBytes));
+        assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
+      }
+    }
+  }
+
+  @Test
+  public void testSignedPostPolicyV4() throws Exception {
+    PostFieldsV4 fields = PostFieldsV4.newBuilder().setAcl("public-read").build();
+
+    PostPolicyV4 policy =
+        storage.generateSignedPostPolicyV4(
+            BlobInfo.newBuilder(bucketName, "my-object").build(), 7, TimeUnit.DAYS, fields);
+
+    HttpClient client = HttpClientBuilder.create().build();
+    HttpPost request = new HttpPost(policy.getUrl());
+    MultipartEntityBuilder builder = MultipartEntityBuilder.create();
+
+    for (Map.Entry<String, String> entry : policy.getFields().entrySet()) {
+      builder.addTextBody(entry.getKey(), entry.getValue());
+    }
+    File file = File.createTempFile("temp", "file");
+    Files.write(file.toPath(), "hello world".getBytes());
+    builder.addBinaryBody(
+        "file", new FileInputStream(file), ContentType.APPLICATION_OCTET_STREAM, file.getName());
+    request.setEntity(builder.build());
+    client.execute(request);
+
+    assertEquals("hello world", new String(storage.get(bucketName, "my-object").getContent()));
+  }
+
+  @Test
+  public void testUploadUsingSignedURL() throws Exception {
+    String blobName = "test-signed-url-upload";
+    BlobInfo blob = BlobInfo.newBuilder(bucketName, blobName).build();
+    assertNotNull(storage.create(blob));
+    Map<String, String> extensionHeaders = new HashMap<>();
+    extensionHeaders.put("x-goog-resumable", "start");
+    for (Storage.SignUrlOption urlStyle :
+        Arrays.asList(
+            Storage.SignUrlOption.withPathStyle(),
+            Storage.SignUrlOption.withVirtualHostedStyle())) {
+      URL signUrl =
+          storage.signUrl(
+              blob,
+              1,
+              TimeUnit.HOURS,
+              Storage.SignUrlOption.httpMethod(HttpMethod.POST),
+              Storage.SignUrlOption.withExtHeaders(extensionHeaders),
+              urlStyle);
+      byte[] bytesArrayToUpload = BLOB_STRING_CONTENT.getBytes();
+      Storage unauthenticatedStorage = StorageOptions.getUnauthenticatedInstance().getService();
+      try (WriteChannel writer = unauthenticatedStorage.writer(signUrl)) {
+        writer.write(ByteBuffer.wrap(bytesArrayToUpload, 0, bytesArrayToUpload.length));
+      }
+
+      int lengthOfDownLoadBytes = -1;
+      BlobId blobId = BlobId.of(bucketName, blobName);
+      Blob blobToRead = storage.get(blobId);
+      try (ReadChannel reader = blobToRead.reader()) {
+        ByteBuffer bytes = ByteBuffer.allocate(64 * 1024);
+        lengthOfDownLoadBytes = reader.read(bytes);
+      }
+
+      assertEquals(bytesArrayToUpload.length, lengthOfDownLoadBytes);
+      assertTrue(storage.delete(bucketName, blobName));
+    }
+  }
+}
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/ITStorageTest.java
index 600f5045..60249afb 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/ITStorageTest.java
@@ -28,15 +28,11 @@ 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 static org.junit.Assume.assumeTrue;
 
 import com.google.api.client.http.HttpTransport;
 import com.google.api.client.http.apache.ApacheHttpTransport;
-import com.google.api.client.util.DateTime;
 import com.google.api.gax.paging.Page;
-import com.google.auth.ServiceAccountSigner;
 import com.google.auth.http.HttpTransportFactory;
-import com.google.auth.oauth2.GoogleCredentials;
 import com.google.cloud.Condition;
 import com.google.cloud.Identity;
 import com.google.cloud.Policy;
@@ -46,16 +42,6 @@ import com.google.cloud.ServiceOptions;
 import com.google.cloud.TransportOptions;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.http.HttpTransportOptions;
-import com.google.cloud.kms.v1.CreateCryptoKeyRequest;
-import com.google.cloud.kms.v1.CreateKeyRingRequest;
-import com.google.cloud.kms.v1.CryptoKey;
-import com.google.cloud.kms.v1.CryptoKeyName;
-import com.google.cloud.kms.v1.GetCryptoKeyRequest;
-import com.google.cloud.kms.v1.GetKeyRingRequest;
-import com.google.cloud.kms.v1.KeyManagementServiceGrpc;
-import com.google.cloud.kms.v1.KeyManagementServiceGrpc.KeyManagementServiceBlockingStub;
-import com.google.cloud.kms.v1.KeyRingName;
-import com.google.cloud.kms.v1.LocationName;
 import com.google.cloud.pubsub.v1.TopicAdminClient;
 import com.google.cloud.storage.Acl;
 import com.google.cloud.storage.Acl.Role;
@@ -78,8 +64,6 @@ import com.google.cloud.storage.HmacKey.HmacKeyState;
 import com.google.cloud.storage.HttpMethod;
 import com.google.cloud.storage.Notification;
 import com.google.cloud.storage.NotificationInfo;
-import com.google.cloud.storage.PostPolicyV4;
-import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
 import com.google.cloud.storage.Rpo;
 import com.google.cloud.storage.ServiceAccount;
 import com.google.cloud.storage.Storage;
@@ -93,9 +77,6 @@ import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageFixture;
 import com.google.cloud.storage.StorageOptions;
 import com.google.cloud.storage.StorageRoles;
-import com.google.cloud.storage.spi.StorageRpcFactory;
-import com.google.cloud.storage.spi.v1.StorageRpc;
-import com.google.cloud.storage.spi.v1.StorageRpc.Option;
 import com.google.cloud.storage.testing.RemoteStorageHelper;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -104,31 +85,16 @@ 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 com.google.common.reflect.AbstractInvocationHandler;
-import com.google.common.reflect.Reflection;
 import com.google.iam.v1.Binding;
 import com.google.iam.v1.GetIamPolicyRequest;
-import com.google.iam.v1.IAMPolicyGrpc;
 import com.google.iam.v1.SetIamPolicyRequest;
-import io.grpc.ManagedChannel;
-import io.grpc.ManagedChannelBuilder;
-import io.grpc.Metadata;
-import io.grpc.Status;
-import io.grpc.StatusRuntimeException;
-import io.grpc.auth.MoreCallCredentials;
-import io.grpc.stub.MetadataUtils;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.lang.reflect.Method;
-import java.net.URL;
-import java.net.URLConnection;
 import java.nio.ByteBuffer;
-import java.nio.file.Files;
 import java.security.Key;
+import java.time.OffsetDateTime;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -142,17 +108,11 @@ import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 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.client.HttpClient;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.entity.ContentType;
-import org.apache.http.entity.mime.MultipartEntityBuilder;
-import org.apache.http.impl.client.HttpClientBuilder;
 import org.apache.http.impl.client.HttpClients;
 import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
 import org.junit.AfterClass;
@@ -161,22 +121,13 @@ import org.junit.ClassRule;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TestName;
-import org.threeten.bp.Clock;
 import org.threeten.bp.Duration;
 import org.threeten.bp.Instant;
-import org.threeten.bp.ZoneId;
-import org.threeten.bp.ZoneOffset;
-import org.threeten.bp.format.DateTimeFormatter;
 
 public class ITStorageTest {
 
   private static Storage storage;
   private static TopicAdminClient topicAdminClient;
-  private static String kmsKeyOneResourcePath;
-  private static String kmsKeyTwoResourcePath;
-  private static Metadata requestParamsHeader = new Metadata();
-  private static Metadata.Key<String> requestParamsKey =
-      Metadata.Key.of("x-goog-request-params", Metadata.ASCII_STRING_MARSHALLER);
   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();
@@ -203,10 +154,6 @@ public class ITStorageTest {
   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 String KMS_KEY_RING_NAME = "gcs_test_kms_key_ring";
-  private static final String KMS_KEY_RING_LOCATION = "us";
-  private static final String KMS_KEY_ONE_NAME = "gcs_kms_key_one";
-  private static final String KMS_KEY_TWO_NAME = "gcs_kms_key_two";
   private static final boolean IS_VPC_TEST =
       System.getenv("GOOGLE_CLOUD_TESTS_IN_VPCSC") != null
           && System.getenv("GOOGLE_CLOUD_TESTS_IN_VPCSC").equalsIgnoreCase("true");
@@ -241,8 +188,6 @@ public class ITStorageTest {
       Notification.PayloadFormat.JSON_API_V1.JSON_API_V1;
   private static final Map<String, String> CUSTOM_ATTRIBUTES = ImmutableMap.of("label1", "value1");
 
-  private static ManagedChannel kmsChannel;
-
   @BeforeClass
   public static void beforeClass() throws IOException {
     storage = storageFixture.getInstance();
@@ -259,9 +204,6 @@ public class ITStorageTest {
 
     storage.create(BucketInfo.newBuilder(BUCKET_REQUESTER_PAYS).build());
 
-    // Prepare KMS KeyRing for CMEK tests
-    prepareKmsKeys();
-
     // Configure topic admin client for notification.
     topicAdminClient = configureTopicAdminClient();
   }
@@ -284,14 +226,6 @@ public class ITStorageTest {
 
   @AfterClass
   public static void afterClass() throws ExecutionException, InterruptedException {
-    if (kmsChannel != null) {
-      try {
-        kmsChannel.shutdownNow();
-      } catch (Exception e) {
-        log.log(Level.WARNING, "Error while trying to shutdown kms channel", e);
-      }
-      kmsChannel = null;
-    }
 
     /* Delete the Pub/Sub topic */
     if (topicAdminClient != null) {
@@ -330,135 +264,6 @@ public class ITStorageTest {
     }
   }
 
-  private static void prepareKmsKeys() throws IOException {
-    // https://cloud.google.com/storage/docs/encryption/using-customer-managed-keys
-    String projectId = storage.getOptions().getProjectId();
-    GoogleCredentials credentials = GoogleCredentials.getApplicationDefault();
-    kmsChannel = ManagedChannelBuilder.forTarget("cloudkms.googleapis.com:443").build();
-    KeyManagementServiceBlockingStub kmsStub =
-        KeyManagementServiceGrpc.newBlockingStub(kmsChannel)
-            .withCallCredentials(MoreCallCredentials.from(credentials));
-    IAMPolicyGrpc.IAMPolicyBlockingStub iamStub =
-        IAMPolicyGrpc.newBlockingStub(kmsChannel)
-            .withCallCredentials(MoreCallCredentials.from(credentials));
-    ensureKmsKeyRingExistsForTests(kmsStub, projectId, KMS_KEY_RING_LOCATION, KMS_KEY_RING_NAME);
-    ensureKmsKeyRingIamPermissionsForTests(
-        iamStub, projectId, KMS_KEY_RING_LOCATION, KMS_KEY_RING_NAME);
-    kmsKeyOneResourcePath =
-        ensureKmsKeyExistsForTests(
-            kmsStub, projectId, KMS_KEY_RING_LOCATION, KMS_KEY_RING_NAME, KMS_KEY_ONE_NAME);
-    kmsKeyTwoResourcePath =
-        ensureKmsKeyExistsForTests(
-            kmsStub, projectId, KMS_KEY_RING_LOCATION, KMS_KEY_RING_NAME, KMS_KEY_TWO_NAME);
-  }
-
-  private static String ensureKmsKeyRingExistsForTests(
-      KeyManagementServiceBlockingStub kmsStub,
-      String projectId,
-      String location,
-      String keyRingName)
-      throws StatusRuntimeException {
-    String kmsKeyRingResourcePath = KeyRingName.of(projectId, location, keyRingName).toString();
-    try {
-      // Attempt to Get KeyRing
-      GetKeyRingRequest getKeyRingRequest =
-          GetKeyRingRequest.newBuilder().setName(kmsKeyRingResourcePath).build();
-      requestParamsHeader.put(requestParamsKey, "name=" + kmsKeyRingResourcePath);
-      KeyManagementServiceBlockingStub stubForGetKeyRing =
-          MetadataUtils.attachHeaders(kmsStub, requestParamsHeader);
-      stubForGetKeyRing.getKeyRing(getKeyRingRequest);
-    } catch (StatusRuntimeException ex) {
-      if (ex.getStatus().getCode() == Status.Code.NOT_FOUND) {
-        // Create KmsKeyRing
-        String keyRingParent = LocationName.of(projectId, location).toString();
-        CreateKeyRingRequest createKeyRingRequest =
-            CreateKeyRingRequest.newBuilder()
-                .setParent(keyRingParent)
-                .setKeyRingId(keyRingName)
-                .build();
-        requestParamsHeader.put(requestParamsKey, "parent=" + keyRingParent);
-        KeyManagementServiceBlockingStub stubForCreateKeyRing =
-            MetadataUtils.attachHeaders(kmsStub, requestParamsHeader);
-        stubForCreateKeyRing.createKeyRing(createKeyRingRequest);
-      } else {
-        throw ex;
-      }
-    }
-
-    return kmsKeyRingResourcePath;
-  }
-
-  private static void ensureKmsKeyRingIamPermissionsForTests(
-      IAMPolicyGrpc.IAMPolicyBlockingStub iamStub,
-      String projectId,
-      String location,
-      String keyRingName)
-      throws StatusRuntimeException {
-    ServiceAccount serviceAccount = storage.getServiceAccount(projectId);
-    String kmsKeyRingResourcePath = KeyRingName.of(projectId, location, keyRingName).toString();
-    Binding binding =
-        Binding.newBuilder()
-            .setRole("roles/cloudkms.cryptoKeyEncrypterDecrypter")
-            .addMembers("serviceAccount:" + serviceAccount.getEmail())
-            .build();
-    com.google.iam.v1.Policy policy =
-        com.google.iam.v1.Policy.newBuilder().addBindings(binding).build();
-    SetIamPolicyRequest setIamPolicyRequest =
-        SetIamPolicyRequest.newBuilder()
-            .setResource(kmsKeyRingResourcePath)
-            .setPolicy(policy)
-            .build();
-    requestParamsHeader.put(requestParamsKey, "parent=" + kmsKeyRingResourcePath);
-    iamStub = MetadataUtils.attachHeaders(iamStub, requestParamsHeader);
-    try {
-      iamStub.setIamPolicy(setIamPolicyRequest);
-    } catch (StatusRuntimeException e) {
-      if (log.isLoggable(Level.WARNING)) {
-        log.log(Level.WARNING, "Unable to set IAM policy: {0}", e.getMessage());
-      }
-    }
-  }
-
-  private static String ensureKmsKeyExistsForTests(
-      KeyManagementServiceBlockingStub kmsStub,
-      String projectId,
-      String location,
-      String keyRingName,
-      String keyName)
-      throws StatusRuntimeException {
-    String kmsKeyResourcePath =
-        CryptoKeyName.of(projectId, location, keyRingName, keyName).toString();
-    try {
-      // Attempt to Get CryptoKey
-      requestParamsHeader.put(requestParamsKey, "name=" + kmsKeyResourcePath);
-      GetCryptoKeyRequest getCryptoKeyRequest =
-          GetCryptoKeyRequest.newBuilder().setName(kmsKeyResourcePath).build();
-      KeyManagementServiceGrpc.KeyManagementServiceBlockingStub stubForGetCryptoKey =
-          MetadataUtils.attachHeaders(kmsStub, requestParamsHeader);
-      stubForGetCryptoKey.getCryptoKey(getCryptoKeyRequest);
-    } catch (StatusRuntimeException ex) {
-      if (ex.getStatus().getCode() == Status.Code.NOT_FOUND) {
-        String kmsKeyRingResourcePath = KeyRingName.of(projectId, location, keyRingName).toString();
-        CryptoKey cryptoKey =
-            CryptoKey.newBuilder().setPurpose(CryptoKey.CryptoKeyPurpose.ENCRYPT_DECRYPT).build();
-        CreateCryptoKeyRequest createCryptoKeyRequest =
-            CreateCryptoKeyRequest.newBuilder()
-                .setCryptoKeyId(keyName)
-                .setParent(kmsKeyRingResourcePath)
-                .setCryptoKey(cryptoKey)
-                .build();
-
-        requestParamsHeader.put(requestParamsKey, "parent=" + kmsKeyRingResourcePath);
-        KeyManagementServiceGrpc.KeyManagementServiceBlockingStub stubForCreateCryptoKey =
-            MetadataUtils.attachHeaders(kmsStub, requestParamsHeader);
-        stubForCreateCryptoKey.createCryptoKey(createCryptoKeyRequest);
-      } else {
-        throw ex;
-      }
-    }
-    return kmsKeyResourcePath;
-  }
-
   private static TopicAdminClient configureTopicAdminClient() throws IOException {
     TopicAdminClient topicAdminClient = TopicAdminClient.create();
     topicAdminClient.createTopic(TOPIC);
@@ -515,14 +320,6 @@ public class ITStorageTest {
     assertNotNull(remoteBucket.getSelfLink());
   }
 
-  @Test
-  public void testGetBucketEmptyFields() {
-    Bucket remoteBucket = storage.get(BUCKET, Storage.BucketGetOption.fields());
-    assertEquals(BUCKET, remoteBucket.getName());
-    assertNull(remoteBucket.getCreateTime());
-    assertNull(remoteBucket.getSelfLink());
-  }
-
   @Test
   public void testGetBucketLifecycleRules() {
     String lifecycleTestBucketName = RemoteStorageHelper.generateBucketName();
@@ -537,11 +334,11 @@ public class ITStorageTest {
                             .setAge(1)
                             .setNumberOfNewerVersions(3)
                             .setIsLive(false)
-                            .setCreatedBefore(new DateTime(System.currentTimeMillis()))
+                            .setCreatedBeforeOffsetDateTime(OffsetDateTime.now())
                             .setMatchesStorageClass(ImmutableList.of(StorageClass.COLDLINE))
                             .setDaysSinceNoncurrentTime(30)
-                            .setNoncurrentTimeBefore(new DateTime(System.currentTimeMillis()))
-                            .setCustomTimeBefore(new DateTime(System.currentTimeMillis()))
+                            .setNoncurrentTimeBeforeOffsetDateTime(OffsetDateTime.now())
+                            .setCustomTimeBeforeOffsetDateTime(OffsetDateTime.now())
                             .setDaysSinceCustomTime(30)
                             .build())))
             .build());
@@ -555,14 +352,14 @@ public class ITStorageTest {
               .getActionType()
               .equals(LifecycleRule.SetStorageClassLifecycleAction.TYPE));
       assertEquals(3, lifecycleRule.getCondition().getNumberOfNewerVersions().intValue());
-      assertNotNull(lifecycleRule.getCondition().getCreatedBefore());
+      assertNotNull(lifecycleRule.getCondition().getCreatedBeforeOffsetDateTime());
       assertFalse(lifecycleRule.getCondition().getIsLive());
       assertEquals(1, lifecycleRule.getCondition().getAge().intValue());
       assertEquals(1, lifecycleRule.getCondition().getMatchesStorageClass().size());
       assertEquals(30, lifecycleRule.getCondition().getDaysSinceNoncurrentTime().intValue());
-      assertNotNull(lifecycleRule.getCondition().getNoncurrentTimeBefore());
+      assertNotNull(lifecycleRule.getCondition().getNoncurrentTimeBeforeOffsetDateTime());
       assertEquals(30, lifecycleRule.getCondition().getDaysSinceCustomTime().intValue());
-      assertNotNull(lifecycleRule.getCondition().getCustomTimeBefore());
+      assertNotNull(lifecycleRule.getCondition().getCustomTimeBeforeOffsetDateTime());
     } finally {
       storage.delete(lifecycleTestBucketName);
     }
@@ -591,45 +388,6 @@ public class ITStorageTest {
     }
   }
 
-  @Test
-  public void testClearBucketDefaultKmsKeyName() throws ExecutionException, InterruptedException {
-    String bucketName = RemoteStorageHelper.generateBucketName();
-    Bucket remoteBucket =
-        storage.create(
-            BucketInfo.newBuilder(bucketName)
-                .setDefaultKmsKeyName(kmsKeyOneResourcePath)
-                .setLocation(KMS_KEY_RING_LOCATION)
-                .build());
-
-    try {
-      assertEquals(kmsKeyOneResourcePath, remoteBucket.getDefaultKmsKeyName());
-      Bucket updatedBucket = remoteBucket.toBuilder().setDefaultKmsKeyName(null).build().update();
-      assertNull(updatedBucket.getDefaultKmsKeyName());
-    } finally {
-      RemoteStorageHelper.forceDelete(storage, bucketName, 5, TimeUnit.SECONDS);
-    }
-  }
-
-  @Test
-  public void testUpdateBucketDefaultKmsKeyName() throws ExecutionException, InterruptedException {
-    String bucketName = RemoteStorageHelper.generateBucketName();
-    Bucket remoteBucket =
-        storage.create(
-            BucketInfo.newBuilder(bucketName)
-                .setDefaultKmsKeyName(kmsKeyOneResourcePath)
-                .setLocation(KMS_KEY_RING_LOCATION)
-                .build());
-
-    try {
-      assertEquals(kmsKeyOneResourcePath, remoteBucket.getDefaultKmsKeyName());
-      Bucket updatedBucket =
-          remoteBucket.toBuilder().setDefaultKmsKeyName(kmsKeyTwoResourcePath).build().update();
-      assertEquals(kmsKeyTwoResourcePath, updatedBucket.getDefaultKmsKeyName());
-    } finally {
-      RemoteStorageHelper.forceDelete(storage, bucketName, 5, TimeUnit.SECONDS);
-    }
-  }
-
   @Test
   public void testCreateBlob() {
     String blobName = "test-create-blob";
@@ -686,66 +444,6 @@ public class ITStorageTest {
     assertNotNull(remoteBlob.getMd5());
   }
 
-  @Test
-  public void testCreateBlobWithKmsKeyName() {
-    String blobName = "test-create-with-kms-key-name-blob";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
-    Blob remoteBlob =
-        storage.create(
-            blob, BLOB_BYTE_CONTENT, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
-    assertNotNull(remoteBlob);
-    assertEquals(blob.getBucket(), remoteBlob.getBucket());
-    assertEquals(blob.getName(), remoteBlob.getName());
-    assertNotNull(remoteBlob.getKmsKeyName());
-    assertTrue(remoteBlob.getKmsKeyName().startsWith(kmsKeyOneResourcePath));
-    byte[] readBytes = storage.readAllBytes(BUCKET, blobName);
-    assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
-  }
-
-  @Test
-  public void testCreateBlobWithKmsKeyNameAndCustomerSuppliedKey() {
-    try {
-      String blobName = "test-create-with-kms-key-name-blob";
-      BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
-      storage.create(
-          blob,
-          BLOB_BYTE_CONTENT,
-          Storage.BlobTargetOption.encryptionKey(KEY),
-          Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
-      fail("StorageException was expected"); // can't supply both.
-    } catch (StorageException ex) {
-      // expected
-    }
-  }
-
-  @Test
-  public void testCreateBlobWithDefaultKmsKeyName()
-      throws ExecutionException, InterruptedException {
-    String bucketName = RemoteStorageHelper.generateBucketName();
-    Bucket bucket =
-        storage.create(
-            BucketInfo.newBuilder(bucketName)
-                .setDefaultKmsKeyName(kmsKeyOneResourcePath)
-                .setLocation(KMS_KEY_RING_LOCATION)
-                .build());
-    assertEquals(bucket.getDefaultKmsKeyName(), kmsKeyOneResourcePath);
-
-    try {
-      String blobName = "test-create-with-default-kms-key-name-blob";
-      BlobInfo blob = BlobInfo.newBuilder(bucket, blobName).build();
-      Blob remoteBlob = storage.create(blob, BLOB_BYTE_CONTENT);
-      assertNotNull(remoteBlob);
-      assertEquals(blob.getBucket(), remoteBlob.getBucket());
-      assertEquals(blob.getName(), remoteBlob.getName());
-      assertNotNull(remoteBlob.getKmsKeyName());
-      assertTrue(remoteBlob.getKmsKeyName().startsWith(kmsKeyOneResourcePath));
-      byte[] readBytes = storage.readAllBytes(bucketName, blobName);
-      assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
-    } finally {
-      RemoteStorageHelper.forceDelete(storage, bucketName, 5, TimeUnit.SECONDS);
-    }
-  }
-
   @Test
   public void testCreateEmptyBlob() {
     String blobName = "test-create-empty-blob";
@@ -848,18 +546,6 @@ public class ITStorageTest {
     assertNull(remoteBlob.getContentType());
   }
 
-  @Test
-  public void testGetBlobKmsKeyNameField() {
-    String blobName = "test-get-selected-kms-key-name-field-blob";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).setContentType(CONTENT_TYPE).build();
-    assertNotNull(storage.create(blob, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath)));
-    Blob remoteBlob =
-        storage.get(blob.getBlobId(), Storage.BlobGetOption.fields(BlobField.KMS_KEY_NAME));
-    assertEquals(blob.getBlobId(), remoteBlob.getBlobId());
-    assertTrue(remoteBlob.getKmsKeyName().startsWith(kmsKeyOneResourcePath));
-    assertNull(remoteBlob.getContentType());
-  }
-
   @Test
   public void testGetBlobAllSelectedFields() {
     String blobName = "test-get-all-selected-fields-blob";
@@ -954,46 +640,6 @@ public class ITStorageTest {
     }
   }
 
-  @Test(timeout = 5000)
-  public void testListBlobsKmsKeySelectedFields() throws InterruptedException {
-    String[] blobNames = {
-      "test-list-blobs-selected-field-kms-key-name-blob1",
-      "test-list-blobs-selected-field-kms-key-name-blob2"
-    };
-    BlobInfo blob1 = BlobInfo.newBuilder(BUCKET, blobNames[0]).setContentType(CONTENT_TYPE).build();
-    BlobInfo blob2 = BlobInfo.newBuilder(BUCKET, blobNames[1]).setContentType(CONTENT_TYPE).build();
-    Blob remoteBlob1 =
-        storage.create(blob1, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
-    Blob remoteBlob2 =
-        storage.create(blob2, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
-    assertNotNull(remoteBlob1);
-    assertNotNull(remoteBlob2);
-    Page<Blob> page =
-        storage.list(
-            BUCKET,
-            Storage.BlobListOption.prefix("test-list-blobs-selected-field-kms-key-name-blob"),
-            Storage.BlobListOption.fields(BlobField.KMS_KEY_NAME));
-    // Listing blobs is eventually consistent, we loop until the list is of the expected size. The
-    // test fails if timeout is reached.
-    while (Iterators.size(page.iterateAll().iterator()) != 2) {
-      Thread.sleep(500);
-      page =
-          storage.list(
-              BUCKET,
-              Storage.BlobListOption.prefix("test-list-blobs-selected-field-kms-key-name-blob"),
-              Storage.BlobListOption.fields(BlobField.KMS_KEY_NAME));
-    }
-    Set<String> blobSet = ImmutableSet.of(blobNames[0], blobNames[1]);
-    Iterator<Blob> iterator = page.iterateAll().iterator();
-    while (iterator.hasNext()) {
-      Blob remoteBlob = iterator.next();
-      assertEquals(BUCKET, remoteBlob.getBucket());
-      assertTrue(blobSet.contains(remoteBlob.getName()));
-      assertTrue(remoteBlob.getKmsKeyName().startsWith(kmsKeyOneResourcePath));
-      assertNull(remoteBlob.getContentType());
-    }
-  }
-
   @Test(timeout = 5000)
   public void testListBlobsEmptySelectedFields() throws InterruptedException {
     String[] blobNames = {
@@ -1517,75 +1163,6 @@ public class ITStorageTest {
     assertTrue(storage.delete(BUCKET, targetBlobName));
   }
 
-  @Test
-  public void testRotateFromCustomerEncryptionToKmsKey() {
-    String sourceBlobName = "test-copy-blob-encryption-key-source";
-    BlobId source = BlobId.of(BUCKET, sourceBlobName);
-    ImmutableMap<String, String> metadata = ImmutableMap.of("k", "v");
-    Blob remoteBlob =
-        storage.create(
-            BlobInfo.newBuilder(source).build(),
-            BLOB_BYTE_CONTENT,
-            Storage.BlobTargetOption.encryptionKey(KEY));
-    assertNotNull(remoteBlob);
-    String targetBlobName = "test-copy-blob-kms-key-target";
-    BlobInfo target =
-        BlobInfo.newBuilder(BUCKET, targetBlobName)
-            .setContentType(CONTENT_TYPE)
-            .setMetadata(metadata)
-            .build();
-    Storage.CopyRequest req =
-        Storage.CopyRequest.newBuilder()
-            .setSource(source)
-            .setSourceOptions(Storage.BlobSourceOption.decryptionKey(BASE64_KEY))
-            .setTarget(target, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath))
-            .build();
-    CopyWriter copyWriter = storage.copy(req);
-    assertEquals(BUCKET, copyWriter.getResult().getBucket());
-    assertEquals(targetBlobName, copyWriter.getResult().getName());
-    assertEquals(CONTENT_TYPE, copyWriter.getResult().getContentType());
-    assertNotNull(copyWriter.getResult().getKmsKeyName());
-    assertTrue(copyWriter.getResult().getKmsKeyName().startsWith(kmsKeyOneResourcePath));
-    assertArrayEquals(BLOB_BYTE_CONTENT, copyWriter.getResult().getContent());
-    assertEquals(metadata, copyWriter.getResult().getMetadata());
-    assertTrue(copyWriter.isDone());
-    assertTrue(storage.delete(BUCKET, targetBlobName));
-  }
-
-  @Test
-  public void testRotateFromCustomerEncryptionToKmsKeyWithCustomerEncryption() {
-    String sourceBlobName = "test-copy-blob-encryption-key-source";
-    BlobId source = BlobId.of(BUCKET, sourceBlobName);
-    ImmutableMap<String, String> metadata = ImmutableMap.of("k", "v");
-    Blob remoteBlob =
-        storage.create(
-            BlobInfo.newBuilder(source).build(),
-            BLOB_BYTE_CONTENT,
-            Storage.BlobTargetOption.encryptionKey(KEY));
-    assertNotNull(remoteBlob);
-    String targetBlobName = "test-copy-blob-kms-key-target";
-    BlobInfo target =
-        BlobInfo.newBuilder(BUCKET, targetBlobName)
-            .setContentType(CONTENT_TYPE)
-            .setMetadata(metadata)
-            .build();
-    try {
-      Storage.CopyRequest req =
-          Storage.CopyRequest.newBuilder()
-              .setSource(source)
-              .setSourceOptions(Storage.BlobSourceOption.decryptionKey(BASE64_KEY))
-              .setTarget(
-                  target,
-                  Storage.BlobTargetOption.encryptionKey(KEY),
-                  Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath))
-              .build();
-      storage.copy(req);
-      fail("StorageException was expected");
-    } catch (StorageException ex) {
-      // expected
-    }
-  }
-
   @Test
   public void testCopyBlobUpdateMetadata() {
     String sourceBlobName = "test-copy-blob-update-metadata-source";
@@ -2065,149 +1642,6 @@ public class ITStorageTest {
     }
   }
 
-  @Test
-  public void testGetSignedUrl() throws IOException {
-    if (storage.getOptions().getCredentials() != null) {
-      assumeTrue(storage.getOptions().getCredentials() instanceof ServiceAccountSigner);
-    }
-    String blobName = "test-get-signed-url-blob/with/slashes/and?special=!#$&'()*+,:;=?@[]";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
-    Blob remoteBlob = storage.create(blob, BLOB_BYTE_CONTENT);
-    assertNotNull(remoteBlob);
-    for (Storage.SignUrlOption urlStyle :
-        Arrays.asList(
-            Storage.SignUrlOption.withPathStyle(),
-            Storage.SignUrlOption.withVirtualHostedStyle())) {
-      URL url = storage.signUrl(blob, 1, TimeUnit.HOURS, urlStyle);
-      URLConnection connection = url.openConnection();
-      byte[] readBytes = new byte[BLOB_BYTE_CONTENT.length];
-      try (InputStream responseStream = connection.getInputStream()) {
-        assertEquals(BLOB_BYTE_CONTENT.length, responseStream.read(readBytes));
-        assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
-      }
-    }
-  }
-
-  @Test
-  public void testGetV2SignedUrlWithAddlQueryParam() throws IOException {
-    if (storage.getOptions().getCredentials() != null) {
-      assumeTrue(storage.getOptions().getCredentials() instanceof ServiceAccountSigner);
-    }
-    String blobName = "test-get-v2-with-generation-param";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
-    Blob remoteBlob = storage.create(blob, BLOB_BYTE_CONTENT);
-    assertNotNull(remoteBlob);
-    for (Storage.SignUrlOption urlStyle :
-        Arrays.asList(
-            Storage.SignUrlOption.withPathStyle(),
-            Storage.SignUrlOption.withVirtualHostedStyle())) {
-      String generationStr = remoteBlob.getGeneration().toString();
-      URL url =
-          storage.signUrl(
-              blob,
-              1,
-              TimeUnit.HOURS,
-              urlStyle,
-              Storage.SignUrlOption.withV2Signature(),
-              Storage.SignUrlOption.withQueryParams(
-                  ImmutableMap.<String, String>of("generation", generationStr)));
-      // Finally, verify that the URL works and we can get the object as expected:
-      URLConnection connection = url.openConnection();
-      byte[] readBytes = new byte[BLOB_BYTE_CONTENT.length];
-      try (InputStream responseStream = connection.getInputStream()) {
-        assertEquals(BLOB_BYTE_CONTENT.length, responseStream.read(readBytes));
-        assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
-      }
-    }
-  }
-
-  // TODO(b/144304815): Remove this test once all conformance tests contain query param test cases.
-  @Test
-  public void testGetV4SignedUrlWithAddlQueryParam() throws IOException {
-    if (storage.getOptions().getCredentials() != null) {
-      assumeTrue(storage.getOptions().getCredentials() instanceof ServiceAccountSigner);
-    }
-    String blobName = "test-get-v4-with-generation-param";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
-    Blob remoteBlob = storage.create(blob, BLOB_BYTE_CONTENT);
-    assertNotNull(remoteBlob);
-    for (Storage.SignUrlOption urlStyle :
-        Arrays.asList(
-            Storage.SignUrlOption.withPathStyle(),
-            Storage.SignUrlOption.withVirtualHostedStyle())) {
-      String generationStr = remoteBlob.getGeneration().toString();
-      URL url =
-          storage.signUrl(
-              blob,
-              1,
-              TimeUnit.HOURS,
-              urlStyle,
-              Storage.SignUrlOption.withV4Signature(),
-              Storage.SignUrlOption.withQueryParams(
-                  ImmutableMap.<String, String>of("generation", generationStr)));
-      // Finally, verify that the URL works and we can get the object as expected:
-      URLConnection connection = url.openConnection();
-      byte[] readBytes = new byte[BLOB_BYTE_CONTENT.length];
-      try (InputStream responseStream = connection.getInputStream()) {
-        assertEquals(BLOB_BYTE_CONTENT.length, responseStream.read(readBytes));
-        assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
-      }
-    }
-  }
-
-  @Test
-  public void testPostSignedUrl() throws IOException {
-    if (storage.getOptions().getCredentials() != null) {
-      assumeTrue(storage.getOptions().getCredentials() instanceof ServiceAccountSigner);
-    }
-    String blobName = "test-post-signed-url-blob";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
-    assertNotNull(storage.create(blob));
-    for (Storage.SignUrlOption urlStyle :
-        Arrays.asList(
-            Storage.SignUrlOption.withPathStyle(),
-            Storage.SignUrlOption.withVirtualHostedStyle())) {
-
-      URL url =
-          storage.signUrl(
-              blob, 1, TimeUnit.HOURS, Storage.SignUrlOption.httpMethod(HttpMethod.POST), urlStyle);
-      URLConnection connection = url.openConnection();
-      connection.setDoOutput(true);
-      connection.connect();
-      Blob remoteBlob = storage.get(BUCKET, blobName);
-      assertNotNull(remoteBlob);
-      assertEquals(blob.getBucket(), remoteBlob.getBucket());
-      assertEquals(blob.getName(), remoteBlob.getName());
-    }
-  }
-
-  @Test
-  public void testV4SignedUrl() throws IOException {
-    if (storage.getOptions().getCredentials() != null) {
-      assumeTrue(storage.getOptions().getCredentials() instanceof ServiceAccountSigner);
-    }
-
-    String blobName = "test-get-signed-url-blob/with/slashes/and?special=!#$&'()*+,:;=?@[]";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
-    Blob remoteBlob = storage.create(blob, BLOB_BYTE_CONTENT);
-    assertNotNull(remoteBlob);
-    for (Storage.SignUrlOption urlStyle :
-        Arrays.asList(
-            Storage.SignUrlOption.withPathStyle(),
-            Storage.SignUrlOption.withVirtualHostedStyle())) {
-
-      URL url =
-          storage.signUrl(
-              blob, 1, TimeUnit.HOURS, Storage.SignUrlOption.withV4Signature(), urlStyle);
-      URLConnection connection = url.openConnection();
-      byte[] readBytes = new byte[BLOB_BYTE_CONTENT.length];
-      try (InputStream responseStream = connection.getInputStream()) {
-        assertEquals(BLOB_BYTE_CONTENT.length, responseStream.read(readBytes));
-        assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
-      }
-    }
-  }
-
   @Test
   public void testGetBlobs() {
     String sourceBlobName1 = "test-get-blobs-1";
@@ -2938,48 +2372,6 @@ public class ITStorageTest {
     }
   }
 
-  @Test
-  public void testListBucketDefaultKmsKeyName() throws ExecutionException, InterruptedException {
-    String bucketName = RemoteStorageHelper.generateBucketName();
-    Bucket remoteBucket =
-        storage.create(
-            BucketInfo.newBuilder(bucketName)
-                .setDefaultKmsKeyName(kmsKeyOneResourcePath)
-                .setLocation(KMS_KEY_RING_LOCATION)
-                .build());
-    assertNotNull(remoteBucket);
-    assertTrue(remoteBucket.getDefaultKmsKeyName().startsWith(kmsKeyOneResourcePath));
-    try {
-      Iterator<Bucket> bucketIterator =
-          storage
-              .list(
-                  Storage.BucketListOption.prefix(bucketName),
-                  Storage.BucketListOption.fields(BucketField.ENCRYPTION))
-              .iterateAll()
-              .iterator();
-      while (!bucketIterator.hasNext()) {
-        Thread.sleep(500);
-        bucketIterator =
-            storage
-                .list(
-                    Storage.BucketListOption.prefix(bucketName),
-                    Storage.BucketListOption.fields(BucketField.ENCRYPTION))
-                .iterateAll()
-                .iterator();
-      }
-      while (bucketIterator.hasNext()) {
-        Bucket bucket = bucketIterator.next();
-        assertTrue(bucket.getName().startsWith(bucketName));
-        assertNotNull(bucket.getDefaultKmsKeyName());
-        assertTrue(bucket.getDefaultKmsKeyName().startsWith(kmsKeyOneResourcePath));
-        assertNull(bucket.getCreateTime());
-        assertNull(bucket.getSelfLink());
-      }
-    } finally {
-      RemoteStorageHelper.forceDelete(storage, bucketName, 5, TimeUnit.SECONDS);
-    }
-  }
-
   @Test
   public void testRetentionPolicyNoLock() throws ExecutionException, InterruptedException {
     String bucketName = RemoteStorageHelper.generateBucketName();
@@ -3483,44 +2875,6 @@ public class ITStorageTest {
     }
   }
 
-  @Test
-  public void testUploadUsingSignedURL() throws Exception {
-    String blobName = "test-signed-url-upload";
-    BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
-    assertNotNull(storage.create(blob));
-    Map<String, String> extensionHeaders = new HashMap<>();
-    extensionHeaders.put("x-goog-resumable", "start");
-    for (Storage.SignUrlOption urlStyle :
-        Arrays.asList(
-            Storage.SignUrlOption.withPathStyle(),
-            Storage.SignUrlOption.withVirtualHostedStyle())) {
-      URL signUrl =
-          storage.signUrl(
-              blob,
-              1,
-              TimeUnit.HOURS,
-              Storage.SignUrlOption.httpMethod(HttpMethod.POST),
-              Storage.SignUrlOption.withExtHeaders(extensionHeaders),
-              urlStyle);
-      byte[] bytesArrayToUpload = BLOB_STRING_CONTENT.getBytes();
-      Storage unauthenticatedStorage = StorageOptions.getUnauthenticatedInstance().getService();
-      try (WriteChannel writer = unauthenticatedStorage.writer(signUrl)) {
-        writer.write(ByteBuffer.wrap(bytesArrayToUpload, 0, bytesArrayToUpload.length));
-      }
-
-      int lengthOfDownLoadBytes = -1;
-      BlobId blobId = BlobId.of(BUCKET, blobName);
-      Blob blobToRead = storage.get(blobId);
-      try (ReadChannel reader = blobToRead.reader()) {
-        ByteBuffer bytes = ByteBuffer.allocate(64 * 1024);
-        lengthOfDownLoadBytes = reader.read(bytes);
-      }
-
-      assertEquals(bytesArrayToUpload.length, lengthOfDownLoadBytes);
-      assertTrue(storage.delete(BUCKET, blobName));
-    }
-  }
-
   @Test
   public void testBucketLocationType() throws ExecutionException, InterruptedException {
     String bucketName = RemoteStorageHelper.generateBucketName();
@@ -3604,31 +2958,6 @@ public class ITStorageTest {
     }
   }
 
-  @Test
-  public void testSignedPostPolicyV4() throws Exception {
-    PostFieldsV4 fields = PostFieldsV4.newBuilder().setAcl("public-read").build();
-
-    PostPolicyV4 policy =
-        storage.generateSignedPostPolicyV4(
-            BlobInfo.newBuilder(BUCKET, "my-object").build(), 7, TimeUnit.DAYS, fields);
-
-    HttpClient client = HttpClientBuilder.create().build();
-    HttpPost request = new HttpPost(policy.getUrl());
-    MultipartEntityBuilder builder = MultipartEntityBuilder.create();
-
-    for (Map.Entry<String, String> entry : policy.getFields().entrySet()) {
-      builder.addTextBody(entry.getKey(), entry.getValue());
-    }
-    File file = File.createTempFile("temp", "file");
-    Files.write(file.toPath(), "hello world".getBytes());
-    builder.addBinaryBody(
-        "file", new FileInputStream(file), ContentType.APPLICATION_OCTET_STREAM, file.getName());
-    request.setEntity(builder.build());
-    client.execute(request);
-
-    assertEquals("hello world", new String(storage.get(BUCKET, "my-object").getContent()));
-  }
-
   @Test
   public void testBlobReload() throws Exception {
     String blobName = "test-blob-reload";
@@ -3909,133 +3238,6 @@ public class ITStorageTest {
     assertThat(updatedBlob2.delete()).isTrue();
   }
 
-  @Test
-  public void testWriterWithKmsKeyName() throws IOException {
-    // Write an empty object with a kmsKeyName.
-    String blobName = "test-empty-blob";
-    BlobInfo blobInfo = BlobInfo.newBuilder(BUCKET, blobName).build();
-    Blob blob =
-        storage.create(blobInfo, Storage.BlobTargetOption.kmsKeyName(kmsKeyOneResourcePath));
-
-    // Create a writer using blob that already has metadata received from Storage API.
-    int numberOfBytes;
-    try (WriteChannel writer = blob.writer()) {
-      byte[] content = BLOB_STRING_CONTENT.getBytes(UTF_8);
-      numberOfBytes = writer.write(ByteBuffer.wrap(content, 0, content.length));
-    }
-    assertThat(numberOfBytes).isEqualTo(27);
-    assertThat(blob.getKmsKeyName()).isNotNull();
-    assertThat(storage.delete(BUCKET, blobName)).isTrue();
-  }
-
-  @Test
-  public void blobWriteChannel_handlesRecoveryOnLastChunkWhenGenerationIsPresent_multipleChunks()
-      throws IOException {
-    int _2MiB = 256 * 1024;
-    int contentSize = 292_617;
-
-    blobWriteChannel_handlesRecoveryOnLastChunkWhenGenerationIsPresent(_2MiB, contentSize);
-  }
-
-  @Test
-  public void blobWriteChannel_handlesRecoveryOnLastChunkWhenGenerationIsPresent_singleChunk()
-      throws IOException {
-    int _4MiB = 256 * 1024 * 2;
-    int contentSize = 292_617;
-
-    blobWriteChannel_handlesRecoveryOnLastChunkWhenGenerationIsPresent(_4MiB, contentSize);
-  }
-
-  private void blobWriteChannel_handlesRecoveryOnLastChunkWhenGenerationIsPresent(
-      int chunkSize, int contentSize) throws IOException {
-    Instant now = Clock.systemUTC().instant();
-    DateTimeFormatter formatter =
-        DateTimeFormatter.ISO_LOCAL_DATE_TIME.withZone(ZoneId.from(ZoneOffset.UTC));
-    String nowString = formatter.format(now);
-
-    String blobPath = String.format("%s/%s/blob", testName.getMethodName(), nowString);
-    BlobId blobId = BlobId.of(BUCKET, blobPath);
-    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();
-
-    ByteBuffer contentGen1 = dataGeneration.randByteBuffer(contentSize);
-    ByteBuffer contentGen2 = dataGeneration.randByteBuffer(contentSize);
-    ByteBuffer contentGen2Expected = contentGen2.duplicate();
-    Storage storage = StorageOptions.getDefaultInstance().getService();
-    WriteChannel ww = storage.writer(blobInfo);
-    ww.setChunkSize(chunkSize);
-    ww.write(contentGen1);
-    ww.close();
-
-    Blob blobGen1 = storage.get(blobId);
-
-    final AtomicBoolean exceptionThrown = new AtomicBoolean(false);
-
-    Storage testStorage =
-        StorageOptions.http()
-            .setServiceRpcFactory(
-                new StorageRpcFactory() {
-                  /**
-                   * Here we're creating a proxy of StorageRpc where we can delegate all calls to
-                   * the normal implementation, except in the case of {@link
-                   * StorageRpc#writeWithResponse(String, byte[], int, long, int, boolean)} where
-                   * {@code lastChunk == true}. We allow the call to execute, but instead of
-                   * returning the result we throw an IOException to simulate a prematurely close
-                   * connection. This behavior is to ensure appropriate handling of a completed
-                   * upload where the ACK wasn't received. In particular, if an upload is initiated
-                   * against an object where an {@link Option#IF_GENERATION_MATCH} simply calling
-                   * get on an object can result in a 404 because the object that is created while
-                   * the BlobWriteChannel is executing will be a new generation.
-                   */
-                  @SuppressWarnings("UnstableApiUsage")
-                  @Override
-                  public StorageRpc create(final StorageOptions options) {
-                    return Reflection.newProxy(
-                        StorageRpc.class,
-                        new AbstractInvocationHandler() {
-                          final StorageRpc delegate =
-                              (StorageRpc) StorageOptions.getDefaultInstance().getRpc();
-
-                          @Override
-                          protected Object handleInvocation(
-                              Object proxy, Method method, Object[] args) throws Throwable {
-                            if ("writeWithResponse".equals(method.getName())) {
-                              Object result = method.invoke(delegate, args);
-                              boolean lastChunk = (boolean) args[5];
-                              // if we're on the lastChunk simulate a connection failure which
-                              // happens after the request was processed but before response could
-                              // be received by the client.
-                              if (lastChunk) {
-                                exceptionThrown.set(true);
-                                throw StorageException.translate(
-                                    new IOException("simulated Connection closed prematurely"));
-                              } else {
-                                return result;
-                              }
-                            }
-                            return method.invoke(delegate, args);
-                          }
-                        });
-                  }
-                })
-            .build()
-            .getService();
-
-    try (WriteChannel w = testStorage.writer(blobGen1, BlobWriteOption.generationMatch())) {
-      w.setChunkSize(chunkSize);
-
-      w.write(contentGen2);
-    }
-
-    assertTrue("Expected an exception to be thrown for the last chunk", exceptionThrown.get());
-
-    Blob blobGen2 = storage.get(blobId);
-    assertEquals(contentSize, (long) blobGen2.getSize());
-    assertNotEquals(blobInfo.getGeneration(), blobGen2.getGeneration());
-    ByteArrayOutputStream actualData = new ByteArrayOutputStream();
-    blobGen2.downloadTo(actualData);
-    assertEquals(contentGen2Expected, ByteBuffer.wrap(actualData.toByteArray()));
-  }
-
   @Test
   public void testRpoConfig() {
     String rpoBucket = RemoteStorageHelper.generateBucketName();
@@ -4052,16 +3254,4 @@ public class ITStorageTest {
       storage.delete(rpoBucket);
     }
   }
-
-  private static String randString(Random rand, int length) {
-    final StringBuilder sb = new StringBuilder();
-    while (sb.length() < length) {
-      int i = rand.nextInt('z');
-      char c = (char) i;
-      if (Character.isLetter(c) || Character.isDigit(c)) {
-        sb.append(c);
-      }
-    }
-    return sb.toString();
-  }
 }