fix: KMS Bad Key error when using existing Blob context to overwrite object

google-cloud-storage/src/main/java/com/google/cloud/storage/BlobInfo.java

@@ -72,7 +72,6 @@ public StorageObject apply(BlobInfo blobInfo) {
           return blobInfo.toPb();
         }
       };
-
   private static final long serialVersionUID = -5625857076205028976L;
   private final BlobId blobId;
   private final String generatedId;
@@ -323,6 +322,23 @@ public Builder setTimeStorageClassUpdated(Long timeStorageClassUpdated) {
 
     abstract Builder setCustomerEncryption(CustomerEncryption customerEncryption);
 
+    /**
+     * Sets a customer-managed key for server-side encryption of the blob. Note that when a KMS key
+     * is used to encrypt Cloud Storage object, object resource metadata will store the version of
+     * the KMS cryptographic. If a {@code Blob} with KMS Key metadata is used to upload a new
+     * version of the object then the existing kmsKeyName version value can't be used in the upload
+     * request and the client instead ignores it.
+     *
+     * <p>Example of setting the KMS key name
+     *
+     * <pre>{@code
+     * String bucketName = "my-unique-bucket";
+     * String blobName = "my-blob-name";
+     * String kmsKeyName = "projects/project-id/locations/us/keyRings/lab1/cryptoKeys/test-key"
+     * BlobInfo blobInfo = BlobInfo.newBuilder(bucketName, blobName).build();
+     * Blob blob = storage.create(blobInfo, Storage.BlobTargetOption.kmsKeyName(kmsKeyName));
+     * }</pre>
+     */
     abstract Builder setKmsKeyName(String kmsKeyName);
 
     /** Sets the blob's event-based hold. */
@@ -1095,7 +1111,6 @@ public ObjectAccessControl apply(Acl acl) {
     if (retentionExpirationTime != null) {
       storageObject.setRetentionExpirationTime(new DateTime(retentionExpirationTime));
     }
-
     storageObject.setKmsKeyName(kmsKeyName);
     storageObject.setEventBasedHold(eventBasedHold);
     storageObject.setTemporaryHold(temporaryHold);

google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java

@@ -832,6 +832,10 @@ public String open(StorageObject object, Map<Option, ?> options) {
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_OPEN);
     Scope scope = tracer.withSpan(span);
     try {
+      String kmsKeyName = object.getKmsKeyName();
+      if (kmsKeyName != null && kmsKeyName.contains("cryptoKeyVersions")) {
+        object.setKmsKeyName("");
+      }
       Insert req = storage.objects().insert(object.getBucket(), object);
       GenericUrl url = req.buildHttpRequest().getUrl();
       String scheme = url.getScheme();

google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITStorageTest.java

@@ -3605,4 +3605,23 @@ public void testBlobTimeStorageClassUpdated() {
         .isEqualTo(updatedBlob1.getTimeStorageClassUpdated());
     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();
+  }
 }