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

Issue:
KTB fails in validation with topology/config parsing not using schemas...
I found this bug on master while testing some changes. This is a showstopper, not possible to run KTB for these cases (not using schemas feature) really. Makes it difficult to contribute...

The problem is that validation of topology and config fails when there are no config for schema.registry.url, regardless of topics having schemas defined or not.

Exception in thread "main" com.purbon.kafka.topology.exceptions.ConfigurationException: Configuration key schema.registry.url should not have the default value mock://
	at com.purbon.kafka.topology.TopologyBuilderConfig.raiseIfDefault(TopologyBuilderConfig.java:202)
	at com.purbon.kafka.topology.TopologyBuilderConfig.validateGeneralConfiguration(TopologyBuilderConfig.java:149)
	at com.purbon.kafka.topology.TopologyBuilderConfig.validateWith(TopologyBuilderConfig.java:124)
	at com.purbon.kafka.topology.KafkaTopologyBuilder.build(KafkaTopologyBuilder.java:113)
	at com.purbon.kafka.topology.KafkaTopologyBuilder.build(KafkaTopologyBuilder.java:68)
	at com.purbon.kafka.topology.BuilderCLI.processTopology(BuilderCLI.java:185)
	at com.purbon.kafka.topology.BuilderCLI.run(BuilderCLI.java:140)
	at com.purbon.kafka.topology.BuilderCLI.main(BuilderCLI.java:130)

The problem is that inTopologyBuilderConfig.raiseIfDefault() you check for existence of schemas on topic. But because of changes to TopicImpl in #127, the topic always have a TopicSchemas object.

Pull request description:
Fix bug with a topology/config not using schemas
This fixes #142  

Patch:
diff --git a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
index 89901b6..0371066 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
@@ -13,7 +13,6 @@ import com.typesafe.config.ConfigFactory;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Properties;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -144,7 +143,7 @@ public class TopologyBuilderConfig {
   }
 
   private void validateGeneralConfiguration(Topology topology) throws ConfigurationException {
-    if (countOfSchemas(topology) > 0) {
+    if (hasSchemas(topology)) {
       raiseIfDefault(CONFLUENT_SCHEMA_REGISTRY_URL_CONFIG, "mock://");
     }
 
@@ -188,12 +187,10 @@ public class TopologyBuilderConfig {
     }
   }
 
-  private static long countOfSchemas(Topology topology) {
+  private static boolean hasSchemas(Topology topology) {
     return topology.getProjects().stream()
         .flatMap((Function<Project, Stream<Topic>>) project -> project.getTopics().stream())
-        .map(Topic::getSchemas)
-        .filter(Objects::nonNull)
-        .count();
+        .anyMatch(topic -> topic.getSchemas().isPresent());
   }
 
   private void raiseIfDefault(String key, String _default) throws ConfigurationException {
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 a24b022..563f4fc 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
@@ -54,8 +54,8 @@ public class SyncTopicAction extends BaseAction {
       adminClient.createTopic(topic, fullTopicName);
     }
 
-    if (topic.getSchemas() != null) {
-      final TopicSchemas schemas = topic.getSchemas();
+    if (topic.getSchemas().isPresent()) {
+      final TopicSchemas schemas = topic.getSchemas().get();
 
       schemas
           .getKeySchemaFile()
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 774e582..3547d5c 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
@@ -24,7 +24,7 @@ public class TopicImpl implements Topic, Cloneable {
   private Optional<String> dataType;
 
   @JsonInclude(Include.NON_EMPTY)
-  private TopicSchemas schemas;
+  private Optional<TopicSchemas> schemas;
 
   private String name;
 
@@ -89,7 +89,7 @@ public class TopicImpl implements Topic, Cloneable {
     this.consumers = consumers;
     this.dataType = dataType;
     this.config = config;
-    this.schemas = new TopicSchemas();
+    this.schemas = Optional.empty();
     this.replicationFactor = 0;
     this.partitionCount = 0;
     this.appConfig = appConfig;
@@ -103,16 +103,16 @@ public class TopicImpl implements Topic, Cloneable {
     return name;
   }
 
-  public TopicSchemas getSchemas() {
-    return schemas;
-  }
-
   @Override
   public String getPlan() {
     return plan;
   }
 
-  public void setSchemas(TopicSchemas schemas) {
+  public Optional<TopicSchemas> getSchemas() {
+    return schemas;
+  }
+
+  public void setSchemas(Optional<TopicSchemas> schemas) {
     this.schemas = schemas;
   }
 
diff --git a/src/main/java/com/purbon/kafka/topology/model/Topic.java b/src/main/java/com/purbon/kafka/topology/model/Topic.java
index fe2e25f..2d71fe1 100644
--- a/src/main/java/com/purbon/kafka/topology/model/Topic.java
+++ b/src/main/java/com/purbon/kafka/topology/model/Topic.java
@@ -15,11 +15,11 @@ public interface Topic {
 
   String getName();
 
-  TopicSchemas getSchemas();
+  Optional<TopicSchemas> getSchemas();
 
   String getPlan();
 
-  void setSchemas(TopicSchemas schemas);
+  void setSchemas(Optional<TopicSchemas> schemas);
 
   Map<String, String> getConfig();
 
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 2eed45f..d71387a 100644
--- a/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
+++ b/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
@@ -90,14 +90,16 @@ public class TopicCustomDeserializer extends StdDeserializer<TopicImpl> {
         .ifPresent(
             node -> {
               topic.setSchemas(
-                  new TopicSchemas(
-                      Optional.ofNullable(node.get("key.schema.file")),
-                      Optional.ofNullable(node.get("value.schema.file"))));
+                  Optional.of(
+                      new TopicSchemas(
+                          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");
-      }
+    if (topic.getSchemas().isPresent()
+        && !topic.getSchemas().get().getValueSchemaFile().isPresent()) {
+      throw new IOException(
+          String.format(
+              "Missing required value.schema.file on schemas for topic %s", topic.getName()));
     }
     LOGGER.debug(
         String.format("Topic %s with config %s has been created", topic.getName(), config));


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/TopologyBuilderConfigTest.java b/src/test/java/com/purbon/kafka/topology/TopologyBuilderConfigTest.java
index 8b8b133..474f711 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologyBuilderConfigTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologyBuilderConfigTest.java
@@ -26,6 +26,7 @@ import java.nio.file.Paths;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Properties;
 import org.junit.Before;
 import org.junit.Test;
@@ -62,7 +63,7 @@ public class TopologyBuilderConfigTest {
     Project project = new ProjectImpl();
     Topic topic = new TopicImpl();
     TopicSchemas schema = new TopicSchemas("foo", "bar");
-    topic.setSchemas(schema);
+    topic.setSchemas(Optional.of(schema));
     project.addTopic(topic);
     topology.addProject(project);
     props.put(CONFLUENT_SCHEMA_REGISTRY_URL_CONFIG, "mock://");
@@ -77,7 +78,7 @@ public class TopologyBuilderConfigTest {
     Project project = new ProjectImpl();
     Topic topic = new TopicImpl();
     TopicSchemas schema = new TopicSchemas("foo", "bar");
-    topic.setSchemas(schema);
+    topic.setSchemas(Optional.of(schema));
     project.addTopic(topic);
     topology.addProject(project);
 
@@ -101,6 +102,18 @@ public class TopologyBuilderConfigTest {
     config.validateWith(topology);
   }
 
+  @Test
+  public void testNoSchemaRegistry() throws ConfigurationException {
+    Topology topology = new TopologyImpl();
+    Project project = new ProjectImpl("project");
+    Topic topic = new TopicImpl("topic", "json");
+    project.addTopic(topic);
+    topology.addProject(project);
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    config.validateWith(topology);
+  }
+
   @Test
   public void testPrefixValidConfigFields() throws ConfigurationException {
     Topology topology = new TopologyImpl();