Repository: googleapis/java-storage
Pull request #1696 (2022-10-06 21:53:56)

fix: update Object Write handling to be able to specify an explicit crc32c value
Storage#createFrom and Storage#writer allow specifying an explicit crc32c value which should be passed to GCS for validation. Update our uploads to allow these explicitly defined values to be passed in the `finish_write` message to GCS.

Update UnifiedOpts.Crc32cValue to represent its value as an int rather than the b64 string from apiary. Internally when the value is necessary for apiary it will be converted to the b64 string representation.

Map gRPC Status `DATA_LOSS` to HTTP 400 Bad Request to mirror the behavior of JSON when a checksum validation fails.

Add a new Hasher.constant instant which will function to specify a constant value for the cumulative value of a write.

Various plumbing tweaks to allow passing this value all the way down and also attempting to "nullSafeConcat"

Add new integration tests to try the negative case to ensue we are getting checksum failures when we pass mismatch values and content.


Patch:
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index 012b7560..9005d64d 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -150,6 +150,11 @@
       <groupId>io.grpc</groupId>
       <artifactId>grpc-stub</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.grpc</groupId>
+      <artifactId>grpc-googleapis</artifactId>
+      <scope>runtime</scope>
+    </dependency>
     <dependency>
       <groupId>org.checkerframework</groupId>
       <artifactId>checker-qual</artifactId>
@@ -261,6 +266,11 @@
             <ignoredDependencies>
               <dependency>io.grpc:grpc-netty-shaded</dependency>
               <!--
+              defining a runtime dependency on this so it is on the default classpath for customers wanting
+              to use direct path
+              -->
+              <dependency>io.grpc:grpc-googleapis</dependency>
+              <!--
               We depend on `net.jqwik:jqwik` which itself has no classes, but depends on each of the other jars.
               In order to not churn on which of the sub apis we happen to be using at this time, we're flagging
               each of them as ignored for dependency checking.
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 af30e950..6dc5575d 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BackwardCompatibilityUtils.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BackwardCompatibilityUtils.java
@@ -93,6 +93,7 @@ final class BackwardCompatibilityUtils {
       case OK:
         return 200;
         // 400 Bad Request
+      case DATA_LOSS:
       case INVALID_ARGUMENT:
       case OUT_OF_RANGE:
         return 400;
@@ -133,7 +134,6 @@ final class BackwardCompatibilityUtils {
       case ABORTED: // ?
       case CANCELLED: // ?
       case UNKNOWN: // ?
-      case DATA_LOSS: // ?
       default:
         return 0;
     }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
index c4e15733..007e4802 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
@@ -88,11 +88,12 @@ final class GapicUnbufferedWritableByteChannel<
       ByteString b = datum.getB();
       int contentSize = b.size();
       long offset = writeCtx.getTotalSentBytes().getAndAdd(contentSize);
+      Crc32cLengthKnown cumulative =
+          writeCtx
+              .getCumulativeCrc32c()
+              .accumulateAndGet(crc32c, chunkSegmenter.getHasher()::nullSafeConcat);
       ChecksummedData.Builder checksummedData = ChecksummedData.newBuilder().setContent(b);
       if (crc32c != null) {
-        writeCtx
-            .getCumulativeCrc32c()
-            .getAndAccumulate(crc32c, chunkSegmenter.getHasher()::nullSafeConcat);
         checksummedData.setCrc32C(crc32c.getValue());
       }
       WriteObjectRequest.Builder builder =
@@ -102,11 +103,9 @@ final class GapicUnbufferedWritableByteChannel<
               .setChecksummedData(checksummedData.build());
       if (!datum.isOnlyFullBlocks()) {
         builder.setFinishWrite(true);
-        if (crc32c != null) {
+        if (cumulative != null) {
           builder.setObjectChecksums(
-              ObjectChecksums.newBuilder()
-                  .setCrc32C(writeCtx.getCumulativeCrc32c().get().getValue())
-                  .build());
+              ObjectChecksums.newBuilder().setCrc32C(cumulative.getValue()).build());
         }
         finished = true;
       }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java
index 4c85726d..1204a542 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java
@@ -53,7 +53,8 @@ final class GrpcBlobWriteChannel implements WriteChannel {
       ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
       RetryingDependencies deps,
       ResultRetryAlgorithm<?> alg,
-      Supplier<ApiFuture<ResumableWrite>> start) {
+      Supplier<ApiFuture<ResumableWrite>> start,
+      Hasher hasher) {
     lazyWriteChannel =
         new LazyWriteChannel(
             Suppliers.memoize(
@@ -61,7 +62,7 @@ final class GrpcBlobWriteChannel implements WriteChannel {
                     ResumableMedia.gapic()
                         .write()
                         .byteChannel(write)
-                        .setHasher(Hasher.noop())
+                        .setHasher(hasher)
                         .setByteStringStrategy(ByteStringStrategy.copy())
                         .resumable()
                         .withRetryConfig(deps, alg)
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
index 5f9f6e0c..4d571091 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
@@ -39,7 +39,6 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.io.BaseEncoding;
-import com.google.common.primitives.Ints;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Timestamp;
 import com.google.storage.v2.Bucket;
@@ -100,9 +99,6 @@ final class GrpcConversions {
   private final Codec<Condition, Expr> iamConditionCodec =
       Codec.of(this::conditionEncode, this::conditionDecode);
 
-  private final Codec<Integer, String> crc32cCodec =
-      Codec.of(this::crc32cEncode, this::crc32cDecode);
-
   @VisibleForTesting
   final Codec<OffsetDateTime, Timestamp> timestampCodec =
       Codec.of(
@@ -750,7 +746,7 @@ final class GrpcConversions {
             ByteString.copyFrom(BaseEncoding.base64().decode(from.getMd5())));
       }
       if (from.getCrc32c() != null) {
-        objectChecksums.setCrc32C(crc32cCodec.decode(from.getCrc32c()));
+        objectChecksums.setCrc32C(Utils.crc32cCodec.decode(from.getCrc32c()));
       }
       toBuilder.setChecksums(objectChecksums.build());
     }
@@ -801,7 +797,7 @@ final class GrpcConversions {
     if (from.hasChecksums()) {
       ObjectChecksums checksums = from.getChecksums();
       if (checksums.hasCrc32C()) {
-        toBuilder.setCrc32c(crc32cCodec.encode(checksums.getCrc32C()));
+        toBuilder.setCrc32c(Utils.crc32cCodec.encode(checksums.getCrc32C()));
       }
       ByteString md5Hash = checksums.getMd5Hash();
       if (!md5Hash.isEmpty()) {
@@ -922,15 +918,6 @@ final class GrpcConversions {
     return to.build();
   }
 
-  private int crc32cDecode(String from) {
-    byte[] decodeCrc32c = BaseEncoding.base64().decode(from);
-    return Ints.fromByteArray(decodeCrc32c);
-  }
-
-  private String crc32cEncode(int from) {
-    return BaseEncoding.base64().encode(Ints.toByteArray(from));
-  }
-
   private Map<String, String> removeNullValues(Map<String, String> from) {
     Map<String, String> to = new HashMap<>(from);
     to.values().removeAll(Collections.singleton(null));
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index a513baeb..0855a0ce 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -279,12 +279,16 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
 
+    Hasher hasher = Hasher.enabled();
+    if (req.hasObjectChecksums() && req.getObjectChecksums().hasCrc32C()) {
+      hasher = Hasher.constant(req.getObjectChecksums().getCrc32C());
+    }
     GapicWritableByteChannelSessionBuilder channelSessionBuilder =
         ResumableMedia.gapic()
             .write()
             .byteChannel(
                 storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
-            .setHasher(Hasher.enabled())
+            .setHasher(hasher)
             .setByteStringStrategy(ByteStringStrategy.noCopy());
 
     long size = Files.size(path);
@@ -348,12 +352,16 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
     ApiFuture<ResumableWrite> start = startResumableWrite(grpcCallContext, req);
 
+    Hasher hasher = Hasher.enabled();
+    if (req.hasObjectChecksums() && req.getObjectChecksums().hasCrc32C()) {
+      hasher = Hasher.constant(req.getObjectChecksums().getCrc32C());
+    }
     BufferedWritableByteChannelSession<WriteObjectResponse> session =
         ResumableMedia.gapic()
             .write()
             .byteChannel(
                 storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
-            .setHasher(Hasher.enabled())
+            .setHasher(hasher)
             .setByteStringStrategy(ByteStringStrategy.noCopy())
             .resumable()
             .withRetryConfig(getOptions(), retryAlgorithmManager.idempotent())
@@ -729,11 +737,16 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
+    Hasher hasher = Hasher.noop();
+    if (req.hasObjectChecksums() && req.getObjectChecksums().hasCrc32C()) {
+      hasher = Hasher.constant(req.getObjectChecksums().getCrc32C());
+    }
     return new GrpcBlobWriteChannel(
         storageClient.writeObjectCallable(),
         getOptions(),
         retryAlgorithmManager.idempotent(),
-        () -> startResumableWrite(grpcCallContext, req));
+        () -> startResumableWrite(grpcCallContext, req),
+        hasher);
   }
 
   @Override
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
index 02e502dc..2a969ef4 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
@@ -64,6 +64,7 @@ public final class GrpcStorageOptions extends StorageOptions
 
   private final GrpcRetryAlgorithmManager retryAlgorithmManager;
   private final Duration terminationAwaitDuration;
+  private final boolean attemptDirectPath;
 
   @BetaApi
   public GrpcStorageOptions(Builder builder, GrpcStorageDefaults serviceDefaults) {
@@ -75,6 +76,7 @@ public final class GrpcStorageOptions extends StorageOptions
     this.terminationAwaitDuration =
         MoreObjects.firstNonNull(
             builder.terminationAwaitDuration, serviceDefaults.getTerminationAwaitDuration());
+    this.attemptDirectPath = builder.attemptDirectPath;
   }
 
   @Override
@@ -94,10 +96,20 @@ public final class GrpcStorageOptions extends StorageOptions
 
   @InternalApi
   StorageSettings getStorageSettings() throws IOException {
-    URI uri = URI.create(getHost());
+    String endpoint = getHost();
+    URI uri = URI.create(endpoint);
     String scheme = uri.getScheme();
-    int port = uri.getPort() > 0 ? uri.getPort() : scheme.equals("http") ? 80 : 443;
-    String endpoint = String.format("%s:%d", uri.getHost(), port);
+    int port = uri.getPort();
+    // Gax routes the endpoint into a method which can't handle schemes, unless for direct path
+    // try and strip here if we can
+    switch (scheme) {
+      case "http":
+        endpoint = String.format("%s:%s", uri.getHost(), port > 0 ? port : 80);
+        break;
+      case "https":
+        endpoint = String.format("%s:%s", uri.getHost(), port > 0 ? port : 443);
+        break;
+    }
 
     CredentialsProvider credentialsProvider;
     if (credentials instanceof NoCredentials) {
@@ -112,13 +124,15 @@ public final class GrpcStorageOptions extends StorageOptions
             .setCredentialsProvider(credentialsProvider)
             .setClock(getClock());
 
+    InstantiatingGrpcChannelProvider.Builder channelProviderBuilder =
+        InstantiatingGrpcChannelProvider.newBuilder()
+            .setEndpoint(endpoint)
+            .setAttemptDirectPath(attemptDirectPath);
+
     if (scheme.equals("http")) {
-      builder.setTransportChannelProvider(
-          InstantiatingGrpcChannelProvider.newBuilder()
-              .setEndpoint(endpoint)
-              .setChannelConfigurator(ManagedChannelBuilder::usePlaintext)
-              .build());
+      channelProviderBuilder.setChannelConfigurator(ManagedChannelBuilder::usePlaintext);
     }
+    builder.setTransportChannelProvider(channelProviderBuilder.build());
     RetrySettings baseRetrySettings = getRetrySettings();
     RetrySettings readRetrySettings =
         baseRetrySettings
@@ -215,6 +229,7 @@ public final class GrpcStorageOptions extends StorageOptions
 
     private StorageRetryStrategy storageRetryStrategy;
     private Duration terminationAwaitDuration;
+    private boolean attemptDirectPath = GrpcStorageDefaults.INSTANCE.isAttemptDirectPath();
 
     Builder() {}
 
@@ -236,6 +251,22 @@ public final class GrpcStorageOptions extends StorageOptions
       return this;
     }
 
+    /**
+     * Option which signifies the client should attempt to connect to gcs via Direct Path.
+     *
+     * <p>In order to use direct path, both this option must be true and the environment variable
+     * (not system property) {@code GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS} must be true.
+     *
+     * <p><i>NOTE</i>There is no need to specify a new endpoint via {@link #setHost(String)} as the
+     * underlying code will translate the normal {@code https://storage.googleapis.com:443} into the
+     * proper Direct Path URI for you.
+     */
+    @BetaApi
+    public GrpcStorageOptions.Builder setAttemptDirectPath(boolean attemptDirectPath) {
+      this.attemptDirectPath = attemptDirectPath;
+      return this;
+    }
+
     @BetaApi
     @Override
     public GrpcStorageOptions.Builder setTransportOptions(TransportOptions transportOptions) {
@@ -380,6 +411,11 @@ public final class GrpcStorageOptions extends StorageOptions
     public Duration getTerminationAwaitDuration() {
       return Duration.ofMinutes(1);
     }
+
+    @BetaApi
+    public boolean isAttemptDirectPath() {
+      return false;
+    }
   }
 
   /**
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Hasher.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Hasher.java
index 06fca041..5c29ce26 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Hasher.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Hasher.java
@@ -47,6 +47,17 @@ interface Hasher {
     return GuavaHasher.INSTANCE;
   }
 
+  /**
+   * Create a Hasher which will always yield the specified value when {@link
+   * #nullSafeConcat(Crc32cLengthKnown, Crc32cLengthKnown)} is invoked.
+   */
+  // Not perfect, and not a great approach for a public API. However, this is the most pragmatic way
+  // right now to wire an externally defined value all the way down to the last write message of a
+  // resumable upload session.
+  static Hasher constant(int crc32c) {
+    return new ConstantConcatValueHasher(Crc32cValue.of(crc32c, -1));
+  }
+
   @Immutable
   class NoOpHasher implements Hasher {
     private static final NoOpHasher INSTANCE = new NoOpHasher();
@@ -101,4 +112,26 @@ interface Hasher {
       }
     }
   }
+
+  @Immutable
+  class ConstantConcatValueHasher implements Hasher {
+    private final Crc32cLengthKnown value;
+
+    private ConstantConcatValueHasher(Crc32cLengthKnown value) {
+      this.value = value;
+    }
+
+    @Override
+    public @Nullable Crc32cLengthKnown hash(ByteBuffer b) {
+      return null;
+    }
+
+    @Override
+    public void validate(Crc32cValue<?> expected, Supplier<ByteBuffer> b) {}
+
+    @Override
+    public @Nullable Crc32cLengthKnown nullSafeConcat(Crc32cLengthKnown r1, Crc32cLengthKnown r2) {
+      return value;
+    }
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
index c8eddbb5..09148838 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
@@ -721,7 +721,11 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
     /**
      * Returns an option for blob's data MD5 hash match. If this option is used the request will
      * fail if blobs' data MD5 hash does not match.
+     *
+     * @deprecated Please compute and use a crc32c checksum instead. {@link #crc32cMatch()}
      */
+    @Deprecated
+    @TransportCompatibility(Transport.HTTP)
     public static BlobWriteOption md5Match() {
       return new BlobWriteOption(UnifiedOpts.md5MatchExtractor());
     }
@@ -730,6 +734,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * Returns an option for blob's data CRC32C checksum match. If this option is used the request
      * will fail if blobs' data CRC32C checksum does not match.
      */
+    @TransportCompatibility({Transport.HTTP, Transport.GRPC})
     public static BlobWriteOption crc32cMatch() {
       return new BlobWriteOption(UnifiedOpts.crc32cMatchExtractor());
     }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
index d2d01487..156d5e31 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
@@ -331,10 +331,14 @@ final class UnifiedOpts {
   compatibility overloads
   -- */
 
-  static Crc32cMatch crc32cMatch(String i) {
+  static Crc32cMatch crc32cMatch(int i) {
     return new Crc32cMatch(i);
   }
 
+  static Crc32cMatch crc32cMatch(String i) {
+    return new Crc32cMatch(Utils.crc32cCodec.decode(i));
+  }
+
   static Delimiter currentDirectory() {
     return new Delimiter("/");
   }
@@ -495,18 +499,17 @@ final class UnifiedOpts {
     return Md5MatchExtractor.INSTANCE;
   }
 
-  @Deprecated
   static final class Crc32cMatch implements ObjectTargetOpt {
     private static final long serialVersionUID = -8680237667319155418L;
-    private final String val;
+    private final int val;
 
-    private Crc32cMatch(String val) {
+    private Crc32cMatch(int val) {
       this.val = val;
     }
 
     @Override
     public Mapper<BlobInfo.Builder> blobInfo() {
-      return b -> b.setCrc32c(val);
+      return b -> b.setCrc32c(Utils.crc32cCodec.encode(val));
     }
 
     @Override
@@ -521,6 +524,14 @@ final class UnifiedOpts {
       return Objects.equals(val, that.val);
     }
 
+    @Override
+    public Mapper<WriteObjectRequest.Builder> writeObject() {
+      return b -> {
+        b.getObjectChecksumsBuilder().setCrc32C(val);
+        return b;
+      };
+    }
+
     @Override
     public int hashCode() {
       return Objects.hash(val);
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 2182f1f2..9a378556 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
@@ -23,6 +23,8 @@ import com.google.api.core.InternalApi;
 import com.google.cloud.storage.Conversions.Codec;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
+import com.google.common.io.BaseEncoding;
+import com.google.common.primitives.Ints;
 import com.google.storage.v2.BucketName;
 import com.google.storage.v2.ProjectName;
 import java.time.Duration;
@@ -147,6 +149,9 @@ final class Utils {
             }
           });
 
+  static final Codec<Integer, String> crc32cCodec =
+      Codec.of(Utils::crc32cEncode, Utils::crc32cDecode);
+
   private Utils() {}
 
   /**
@@ -228,4 +233,13 @@ final class Utils {
     }
     throw new IllegalStateException("Unable to resolve non-null value");
   }
+
+  private static int crc32cDecode(String from) {
+    byte[] decodeCrc32c = BaseEncoding.base64().decode(from);
+    return Ints.fromByteArray(decodeCrc32c);
+  }
+
+  private static String crc32cEncode(int from) {
+    return BaseEncoding.base64().encode(Ints.toByteArray(from));
+  }
 }


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 b84a59a0..43e2dda0 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
@@ -113,7 +113,7 @@ public final class BackwardCompatibilityUtilsTest {
     expected.put(Code.CANCELLED, 0);
     expected.put(Code.UNKNOWN, 0);
     expected.put(Code.DEADLINE_EXCEEDED, 504);
-    expected.put(Code.DATA_LOSS, 0);
+    expected.put(Code.DATA_LOSS, 400);
 
     EnumMap<Code, Integer> actual = new EnumMap<>(Code.class);
     for (Code c : Code.values()) {
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java
index 99e7bb17..4bdb3d7f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java
@@ -45,6 +45,7 @@ import com.google.common.base.Function;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
+import com.google.common.hash.Hashing;
 import com.google.common.io.BaseEncoding;
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
@@ -484,6 +485,7 @@ public class BucketTest {
     BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
     Blob expectedBlob = info.asBlob(serviceMockReturnsOptions);
     byte[] content = {0xD, 0xE, 0xA, 0xD};
+    String crc32c = Utils.crc32cCodec.encode(Hashing.crc32c().hashBytes(content).asInt());
     Storage.PredefinedAcl acl = Storage.PredefinedAcl.ALL_AUTHENTICATED_USERS;
     InputStream streamContent = new ByteArrayInputStream(content);
     expect(storage.getOptions()).andReturn(mockOptions);
@@ -494,7 +496,7 @@ public class BucketTest {
                 new BlobWriteOption(UnifiedOpts.generationMatch(42L)),
                 new BlobWriteOption(UnifiedOpts.metagenerationMatch(24L)),
                 Storage.BlobWriteOption.predefinedAcl(acl),
-                new BlobWriteOption(UnifiedOpts.crc32cMatch("crc")),
+                new BlobWriteOption(UnifiedOpts.crc32cMatch(crc32c)),
                 new BlobWriteOption(UnifiedOpts.md5Match("md5")),
                 Storage.BlobWriteOption.encryptionKey(BASE64_KEY),
                 Storage.BlobWriteOption.userProject(USER_PROJECT)))
@@ -509,7 +511,7 @@ public class BucketTest {
             Bucket.BlobWriteOption.generationMatch(42L),
             Bucket.BlobWriteOption.metagenerationMatch(24L),
             Bucket.BlobWriteOption.predefinedAcl(acl),
-            Bucket.BlobWriteOption.crc32cMatch("crc"),
+            Bucket.BlobWriteOption.crc32cMatch(crc32c),
             Bucket.BlobWriteOption.md5Match("md5"),
             Bucket.BlobWriteOption.encryptionKey(BASE64_KEY),
             Bucket.BlobWriteOption.userProject(USER_PROJECT));
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
index bbf18c40..e525ad77 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
@@ -24,6 +24,7 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.junit.Assume.assumeTrue;
@@ -53,12 +54,17 @@ import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterators;
 import com.google.common.collect.Lists;
+import com.google.common.hash.Hashing;
 import com.google.common.io.BaseEncoding;
+import com.google.common.io.ByteStreams;
+import com.google.common.primitives.Ints;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
 import java.security.Key;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -72,7 +78,9 @@ import java.util.concurrent.TimeUnit;
 import javax.crypto.spec.SecretKeySpec;
 import org.junit.AfterClass;
 import org.junit.ClassRule;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TestName;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
@@ -82,6 +90,11 @@ public class ITObjectTest {
   private static final String CONTENT_TYPE = "text/plain";
   private static final byte[] BLOB_BYTE_CONTENT = {0xD, 0xE, 0xA, 0xD};
   private static final String BLOB_STRING_CONTENT = "Hello Google Cloud Storage!";
+  private static final String BLOB_STRING_CONTENT_CRC32C =
+      BaseEncoding.base64()
+          .encode(
+              Ints.toByteArray(
+                  Hashing.crc32c().hashBytes(BLOB_STRING_CONTENT.getBytes(UTF_8)).asInt()));
   private static final String BASE64_KEY = "JVzfVl8NLD9FjedFuStegjRfES5ll5zc59CIXw572OA=";
   private static final String OTHER_BASE64_KEY = "IcOIQGlliNr5pr3vJb63l+XMqc7NjXqjfw/deBoNxPA=";
   private static final Key KEY =
@@ -121,6 +134,8 @@ public class ITObjectTest {
           .setHandle(storageFixtureHttp::getInstance)
           .build();
 
+  @Rule public final TestName testName = new TestName();
+
   private final BucketFixture bucketFixture;
   private final BucketFixture requesterPaysBucketFixture;
   private final String clientName;
@@ -1499,6 +1514,83 @@ public class ITObjectTest {
     assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
   }
 
+  @Test
+  public void testCrc32cValidated_createFrom_expectFailure() {
+    String blobName = testName.getMethodName();
+    BlobId blobId = BlobId.of(bucketFixture.getBucketInfo().getName(), blobName);
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(BLOB_STRING_CONTENT_CRC32C).build();
+
+    byte[] bytes = (BLOB_STRING_CONTENT + "x").getBytes(UTF_8);
+    StorageException expected =
+        assertThrows(
+            StorageException.class,
+            () ->
+                storage.createFrom(
+                    blobInfo,
+                    new ByteArrayInputStream(bytes),
+                    BlobWriteOption.doesNotExist(),
+                    BlobWriteOption.crc32cMatch()));
+    assertThat(expected.getCode()).isEqualTo(400);
+  }
+
+  @Test
+  public void testCrc32cValidated_createFrom_expectSuccess() throws IOException {
+    String blobName = testName.getMethodName();
+    BlobId blobId = BlobId.of(bucketFixture.getBucketInfo().getName(), blobName);
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(BLOB_STRING_CONTENT_CRC32C).build();
+
+    byte[] bytes = BLOB_STRING_CONTENT.getBytes(UTF_8);
+    Blob blob =
+        storage.createFrom(
+            blobInfo,
+            new ByteArrayInputStream(bytes),
+            BlobWriteOption.doesNotExist(),
+            BlobWriteOption.crc32cMatch());
+    assertThat(blob.getCrc32c()).isEqualTo(BLOB_STRING_CONTENT_CRC32C);
+  }
+
+  @Test
+  public void testCrc32cValidated_writer_expectFailure() {
+    String blobName = testName.getMethodName();
+    BlobId blobId = BlobId.of(bucketFixture.getBucketInfo().getName(), blobName);
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(BLOB_STRING_CONTENT_CRC32C).build();
+
+    byte[] bytes = (BLOB_STRING_CONTENT + "x").getBytes(UTF_8);
+    StorageException expected =
+        assertThrows(
+            StorageException.class,
+            () -> {
+              try (ReadableByteChannel src = Channels.newChannel(new ByteArrayInputStream(bytes));
+                  WriteChannel dst =
+                      storage.writer(
+                          blobInfo,
+                          BlobWriteOption.doesNotExist(),
+                          BlobWriteOption.crc32cMatch())) {
+                ByteStreams.copy(src, dst);
+              }
+            });
+    assertThat(expected.getCode()).isEqualTo(400);
+  }
+
+  @Test
+  public void testCrc32cValidated_writer_expectSuccess() throws IOException {
+    String blobName = testName.getMethodName();
+    BlobId blobId = BlobId.of(bucketFixture.getBucketInfo().getName(), blobName);
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(BLOB_STRING_CONTENT_CRC32C).build();
+
+    byte[] bytes = BLOB_STRING_CONTENT.getBytes(UTF_8);
+
+    try (ReadableByteChannel src = Channels.newChannel(new ByteArrayInputStream(bytes));
+        WriteChannel dst =
+            storage.writer(
+                blobInfo, BlobWriteOption.doesNotExist(), BlobWriteOption.crc32cMatch())) {
+      ByteStreams.copy(src, dst);
+    }
+
+    Blob blob = storage.get(blobId);
+    assertThat(blob.getCrc32c()).isEqualTo(BLOB_STRING_CONTENT_CRC32C);
+  }
+
   private Blob createBlob(String method, BlobInfo blobInfo, boolean detectType) throws IOException {
     switch (method) {
       case "create":