Merge branch '7.5.x' into 7.6.x by rayokota

confluentinc · Oct 3, 2024 · 6e2c2b6 · 6e2c2b6
2 parents a312c5e + b4f6a17
commit 6e2c2b6
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 5 deletions.
diff --git a/...c/main/java/io/confluent/kafka/schemaregistry/storage/encoder/MetadataEncoderService.java b/...c/main/java/io/confluent/kafka/schemaregistry/storage/encoder/MetadataEncoderService.java
@@ -286,7 +286,9 @@ public void decodeMetadata(SchemaValue schema) {
         byte[] plaintext = aead.decrypt(Base64.getDecoder().decode(value), EMPTY_AAD);
         return new String(plaintext, StandardCharsets.UTF_8);
       } catch (GeneralSecurityException e) {
-        throw new IllegalStateException("Could not encrypt sensitive metadata", e);
+        log.error("Could not decrypt sensitive metadata for schema {}", schema, e);
+        // Just return the value as-is if we can't decrypt it
+        return value;
       }
     });
   }

diff --git a/...st/java/io/confluent/kafka/schemaregistry/storage/encoder/MetadataEncoderServiceTest.java b/...st/java/io/confluent/kafka/schemaregistry/storage/encoder/MetadataEncoderServiceTest.java
@@ -32,6 +32,8 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
+import java.util.SortedMap;
+import java.util.TreeMap;
 import org.junit.Test;
 
 public class MetadataEncoderServiceTest {
@@ -61,10 +63,26 @@ public void testEncoding() throws Exception {
     assertNotEquals(schema.getMetadata().getProperties().get("sensitive"), "foo");
     assertNotNull(schema.getMetadata().getProperties().get(SchemaValue.ENCODED_PROPERTY));
 
-    encoderService.decodeMetadata(schema);
-    assertEquals(schema.getMetadata().getProperties().get("nonsensitive"), "foo");
+    SchemaValue schema2 = new SchemaValue(
+        "mysubject", null, null, null, null, null,
+        new io.confluent.kafka.schemaregistry.storage.Metadata(
+            schema.getMetadata().toMetadataEntity()), null, "true", false);
+    encoderService.decodeMetadata(schema2);
+    assertEquals(schema2.getMetadata().getProperties().get("nonsensitive"), "foo");
     // the value of "sensitive" is decrypted
-    assertEquals(schema.getMetadata().getProperties().get("sensitive"), "foo");
-    assertNull(schema.getMetadata().getProperties().get(SchemaValue.ENCODED_PROPERTY));
+    assertEquals(schema2.getMetadata().getProperties().get("sensitive"), "foo");
+    assertNull(schema2.getMetadata().getProperties().get(SchemaValue.ENCODED_PROPERTY));
+
+    SortedMap<String, String> badProperties = new TreeMap<>(schema.getMetadata().getProperties());
+    badProperties.put("sensitive", "badValue");
+    SchemaValue schema3 = new SchemaValue(
+        "mysubject", null, null, null, null, null,
+        new io.confluent.kafka.schemaregistry.storage.Metadata(
+            new Metadata(null, badProperties, Collections.singleton("sensitive"))), null, "true", false);
+    encoderService.decodeMetadata(schema3);
+    assertEquals(schema3.getMetadata().getProperties().get("nonsensitive"), "foo");
+    // the value of "sensitive" is not decrypted
+    assertEquals(schema3.getMetadata().getProperties().get("sensitive"), "badValue");
+    assertNull(schema3.getMetadata().getProperties().get(SchemaValue.ENCODED_PROPERTY));
   }
 }