Repository: kafka-ops/kafka-topology-builder (java)

Issue:
Schemas for values without corresponding schema for key should be supported
Currently KTB throws a null-pointer exception when a schema-file for values is specified without providing a schema file for the corresponding key.

It should be possible to just specify a value schema.

Pull request description:
Make usage of topic key schemas optional, values are always required
*NOTE*:  Throw an exception if the key schema is present but not value one.

closes #122 

Patch:
diff --git a/src/main/java/com/purbon/kafka/topology/actions/topics/SyncTopicAction.java b/src/main/java/com/purbon/kafka/topology/actions/topics/SyncTopicAction.java
index 3ead643..a24b022 100644
--- a/src/main/java/com/purbon/kafka/topology/actions/topics/SyncTopicAction.java
+++ b/src/main/java/com/purbon/kafka/topology/actions/topics/SyncTopicAction.java
@@ -9,7 +9,6 @@ import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -58,13 +57,17 @@ public class SyncTopicAction extends BaseAction {
     if (topic.getSchemas() != null) {
       final TopicSchemas schemas = topic.getSchemas();
 
-      if (StringUtils.isNotBlank(schemas.getKeySchemaFile())) {
-        schemaRegistryManager.register(fullTopicName + "-key", schemas.getKeySchemaFile());
-      }
+      schemas
+          .getKeySchemaFile()
+          .ifPresent(
+              keySchemaFile ->
+                  schemaRegistryManager.register(fullTopicName + "-key", keySchemaFile));
 
-      if (StringUtils.isNotBlank(schemas.getValueSchemaFile())) {
-        schemaRegistryManager.register(fullTopicName + "-value", schemas.getValueSchemaFile());
-      }
+      schemas
+          .getValueSchemaFile()
+          .ifPresent(
+              valueSchemaFile ->
+                  schemaRegistryManager.register(fullTopicName + "-value", valueSchemaFile));
     }
   }
 
diff --git a/src/main/java/com/purbon/kafka/topology/model/Impl/TopicImpl.java b/src/main/java/com/purbon/kafka/topology/model/Impl/TopicImpl.java
index 46be415..22ecb8f 100644
--- a/src/main/java/com/purbon/kafka/topology/model/Impl/TopicImpl.java
+++ b/src/main/java/com/purbon/kafka/topology/model/Impl/TopicImpl.java
@@ -86,6 +86,7 @@ public class TopicImpl implements Topic, Cloneable {
     this.consumers = consumers;
     this.dataType = dataType;
     this.config = config;
+    this.schemas = new TopicSchemas();
     this.replicationFactor = 0;
     this.partitionCount = 0;
     this.appConfig = appConfig;
diff --git a/src/main/java/com/purbon/kafka/topology/model/TopicSchemas.java b/src/main/java/com/purbon/kafka/topology/model/TopicSchemas.java
index 604294c..85cdfb7 100644
--- a/src/main/java/com/purbon/kafka/topology/model/TopicSchemas.java
+++ b/src/main/java/com/purbon/kafka/topology/model/TopicSchemas.java
@@ -1,24 +1,37 @@
 package com.purbon.kafka.topology.model;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.PropertyNamingStrategy;
 import com.fasterxml.jackson.databind.annotation.JsonNaming;
+import java.util.Optional;
 
 @JsonNaming(PropertyNamingStrategy.LowerDotCaseStrategy.class)
 public class TopicSchemas {
 
-  private String keySchemaFile;
-  private String valueSchemaFile;
+  private Optional<String> keySchemaFile;
+  private Optional<String> valueSchemaFile;
+
+  public TopicSchemas() {
+    this(Optional.empty(), Optional.empty());
+  }
+
+  public TopicSchemas(Optional<JsonNode> keyJsonNode, Optional<JsonNode> valueJsonNode) {
+    this.keySchemaFile = Optional.empty();
+    this.valueSchemaFile = Optional.empty();
+    keyJsonNode.ifPresent(node -> keySchemaFile = Optional.of(node.asText()));
+    valueJsonNode.ifPresent(node -> valueSchemaFile = Optional.of(node.asText()));
+  }
 
   public TopicSchemas(String keySchemaFile, String valueSchemaFile) {
-    this.keySchemaFile = keySchemaFile;
-    this.valueSchemaFile = valueSchemaFile;
+    this.keySchemaFile = Optional.of(keySchemaFile);
+    this.valueSchemaFile = Optional.of(valueSchemaFile);
   }
 
-  public String getKeySchemaFile() {
+  public Optional<String> getKeySchemaFile() {
     return keySchemaFile;
   }
 
-  public String getValueSchemaFile() {
+  public Optional<String> getValueSchemaFile() {
     return valueSchemaFile;
   }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java b/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
index cbb7b64..1ab210b 100644
--- a/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
+++ b/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
@@ -67,9 +67,14 @@ public class TopicCustomDeserializer extends StdDeserializer<TopicImpl> {
             node -> {
               topic.setSchemas(
                   new TopicSchemas(
-                      node.get("key.schema.file").asText(),
-                      node.get("value.schema.file").asText()));
+                      Optional.ofNullable(node.get("key.schema.file")),
+                      Optional.ofNullable(node.get("value.schema.file"))));
             });
+    if (rootNode.get("schemas") != null) {
+      if (!topic.getSchemas().getValueSchemaFile().isPresent()) {
+        throw new IOException("Missing required value.schema.file if key file is present");
+      }
+    }
     LOGGER.debug(
         String.format("Topic %s with config %s has been created", topic.getName(), config));
     return topic;


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
index beaf987..b47e3bc 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
@@ -201,6 +201,12 @@ public class TopologySerdesTest {
     assertThat(project.getProducers().get(2).getIdempotence()).isNotEmpty();
   }
 
+  @Test(expected = IllegalArgumentException.class)
+  public void testSchemaSerdes() throws URISyntaxException, IOException {
+    URL topologyDescriptor = getClass().getResource("/descriptor-wrong-schemas.yaml");
+    parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
+  }
+
   @Test
   public void testPlaformProcessing() throws IOException, URISyntaxException {
 
diff --git a/src/test/resources/descriptor-wrong-schemas.yaml b/src/test/resources/descriptor-wrong-schemas.yaml
new file mode 100644
index 0000000..f9414b5
--- /dev/null
+++ b/src/test/resources/descriptor-wrong-schemas.yaml
@@ -0,0 +1,22 @@
+---
+context: "contextOrg"
+projects:
+  - name: "foo"
+    topics:
+      - name: "foo"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: "bar"
+        dataType: "avro"
+        schemas:
+          key.schema.file: "schemas/bar-key.avsc"
+  - name: "bar"
+    topics:
+      - dataType: "avro"
+        schemas:
+          value.schema.file: "schemas/bar-value.avsc"
+        name: "bar"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
diff --git a/src/test/resources/descriptor.yaml b/src/test/resources/descriptor.yaml
index 82eb3f9..5692542 100644
--- a/src/test/resources/descriptor.yaml
+++ b/src/test/resources/descriptor.yaml
@@ -51,6 +51,8 @@ projects:
   - name: "bar"
     topics:
       - dataType: "avro"
+        schemas:
+          value.schema.file: "schemas/bar-value.avsc"
         name: "bar"
         config:
           replication.factor: "1"