ISSUE-3222: SchemaRegistrySerde checkSchemaExistenceForDeserialize property added (#3235)

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serdes/builtin/sr/SchemaRegistrySerde.java

@@ -46,6 +46,7 @@ public class SchemaRegistrySerde implements BuiltInSerde {
   private List<String> schemaRegistryUrls;
   private String valueSchemaNameTemplate;
   private String keySchemaNameTemplate;
+  private boolean checkSchemaExistenceForDeserialize;
 
   private Map<SchemaType, MessageFormatter> schemaRegistryFormatters;
 
@@ -75,7 +76,9 @@ public class SchemaRegistrySerde implements BuiltInSerde {
             kafkaClusterProperties.getProperty("schemaRegistrySSL.truststorePassword", String.class).orElse(null)
         ),
         kafkaClusterProperties.getProperty("schemaRegistryKeySchemaNameTemplate", String.class).orElse("%s-key"),
-        kafkaClusterProperties.getProperty("schemaRegistrySchemaNameTemplate", String.class).orElse("%s-value")
+        kafkaClusterProperties.getProperty("schemaRegistrySchemaNameTemplate", String.class).orElse("%s-value"),
+        kafkaClusterProperties.getProperty("schemaRegistryCheckSchemaExistenceForDeserialize", Boolean.class)
+            .orElse(false)
     );
   }
 
@@ -99,7 +102,9 @@ public class SchemaRegistrySerde implements BuiltInSerde {
             serdeProperties.getProperty("truststorePassword", String.class).orElse(null)
         ),
         serdeProperties.getProperty("keySchemaNameTemplate", String.class).orElse("%s-key"),
-        serdeProperties.getProperty("schemaNameTemplate", String.class).orElse("%s-value")
+        serdeProperties.getProperty("schemaNameTemplate", String.class).orElse("%s-value"),
+        kafkaClusterProperties.getProperty("checkSchemaExistenceForDeserialize", Boolean.class)
+            .orElse(false)
     );
   }
 
@@ -108,12 +113,14 @@ public class SchemaRegistrySerde implements BuiltInSerde {
       List<String> schemaRegistryUrls,
       SchemaRegistryClient schemaRegistryClient,
       String keySchemaNameTemplate,
-      String valueSchemaNameTemplate) {
+      String valueSchemaNameTemplate,
+      boolean checkTopicSchemaExistenceForDeserialize) {
     this.schemaRegistryUrls = schemaRegistryUrls;
     this.schemaRegistryClient = schemaRegistryClient;
     this.keySchemaNameTemplate = keySchemaNameTemplate;
     this.valueSchemaNameTemplate = valueSchemaNameTemplate;
     this.schemaRegistryFormatters = MessageFormatter.createMap(schemaRegistryClient);
+    this.checkSchemaExistenceForDeserialize = checkTopicSchemaExistenceForDeserialize;
   }
 
   private static SchemaRegistryClient createSchemaRegistryClient(List<String> urls,
@@ -122,8 +129,7 @@ public class SchemaRegistrySerde implements BuiltInSerde {
                                                                  @Nullable String keyStoreLocation,
                                                                  @Nullable String keyStorePassword,
                                                                  @Nullable String trustStoreLocation,
-                                                                 @Nullable String trustStorePassword
-                                                                 ) {
+                                                                 @Nullable String trustStorePassword) {
     Map<String, String> configs = new HashMap<>();
     if (username != null && password != null) {
       configs.put(BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO");
@@ -169,7 +175,8 @@ public class SchemaRegistrySerde implements BuiltInSerde {
   @Override
   public boolean canDeserialize(String topic, Target type) {
     String subject = schemaSubject(topic, type);
-    return getSchemaBySubject(subject).isPresent();
+    return !checkSchemaExistenceForDeserialize
+        || getSchemaBySubject(subject).isPresent();
   }
 
   @Override

kafka-ui-api/src/test/java/com/provectus/kafka/ui/serdes/builtin/sr/SchemaRegistrySerdeTest.java

@@ -22,6 +22,7 @@ import org.apache.avro.generic.GenericDatumWriter;
 import org.apache.avro.io.Encoder;
 import org.apache.avro.io.EncoderFactory;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
@@ -35,7 +36,7 @@ class SchemaRegistrySerdeTest {
   @BeforeEach
   void init() {
     serde = new SchemaRegistrySerde();
-    serde.configure(List.of("wontbeused"), registryClient, "%s-key", "%s-value");
+    serde.configure(List.of("wontbeused"), registryClient, "%s-key", "%s-value", true);
   }
 
   @ParameterizedTest
@@ -129,24 +130,61 @@ class SchemaRegistrySerdeTest {
         .contains(Map.entry("schemaId", schemaId));
   }
 
+  @Nested
+  class SerdeWithDisabledSubjectExistenceCheck {
+
+    @BeforeEach
+    void init() {
+      serde.configure(List.of("wontbeused"), registryClient, "%s-key", "%s-value", false);
+    }
+
+    @Test
+    void canDeserializeAlwaysReturnsTrue() {
+      String topic = RandomString.make(10);
+      assertThat(serde.canDeserialize(topic, Serde.Target.KEY)).isTrue();
+      assertThat(serde.canDeserialize(topic, Serde.Target.VALUE)).isTrue();
+    }
+  }
+
+  @Nested
+  class SerdeWithEnabledSubjectExistenceCheck {
+
+    @BeforeEach
+    void init() {
+      serde.configure(List.of("wontbeused"), registryClient, "%s-key", "%s-value", true);
+    }
+
+    @Test
+    void canDeserializeReturnsTrueIfSubjectExists() throws Exception {
+      String topic = RandomString.make(10);
+      registryClient.register(topic + "-key", new AvroSchema("\"int\""));
+      registryClient.register(topic + "-value", new AvroSchema("\"int\""));
+
+      assertThat(serde.canDeserialize(topic, Serde.Target.KEY)).isTrue();
+      assertThat(serde.canDeserialize(topic, Serde.Target.VALUE)).isTrue();
+    }
+
+    @Test
+    void canDeserializeReturnsFalseIfSubjectDoesNotExist() {
+      String topic = RandomString.make(10);
+      assertThat(serde.canDeserialize(topic, Serde.Target.KEY)).isFalse();
+      assertThat(serde.canDeserialize(topic, Serde.Target.VALUE)).isFalse();
+    }
+  }
+
   @Test
   void canDeserializeAndCanSerializeReturnsTrueIfSubjectExists() throws Exception {
     String topic = RandomString.make(10);
     registryClient.register(topic + "-key", new AvroSchema("\"int\""));
     registryClient.register(topic + "-value", new AvroSchema("\"int\""));
 
-    assertThat(serde.canDeserialize(topic, Serde.Target.KEY)).isTrue();
-    assertThat(serde.canDeserialize(topic, Serde.Target.VALUE)).isTrue();
-
     assertThat(serde.canSerialize(topic, Serde.Target.KEY)).isTrue();
     assertThat(serde.canSerialize(topic, Serde.Target.VALUE)).isTrue();
   }
 
   @Test
-  void canDeserializeAndCanSerializeReturnsFalseIfSubjectDoesNotExist() {
+  void canSerializeReturnsFalseIfSubjectDoesNotExist() {
     String topic = RandomString.make(10);
-    assertThat(serde.canDeserialize(topic, Serde.Target.KEY)).isFalse();
-    assertThat(serde.canDeserialize(topic, Serde.Target.VALUE)).isFalse();
     assertThat(serde.canSerialize(topic, Serde.Target.KEY)).isFalse();
     assertThat(serde.canSerialize(topic, Serde.Target.VALUE)).isFalse();
   }
@@ -178,4 +216,4 @@ class SchemaRegistrySerdeTest {
     return output.toByteArray();
   }
 
-}
+}