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

Issue:
Should fail with exception when deserializing invalid topology
When a specific file path is given to KTB for the topology an exception will be thrown in case of invalid topology. So then also a CI pipeline will fail, as it should. 

However in the case when a directory (of topology files) is specified then the error is only logged and processing continues, so the CI pipeline will wrongly look like it succeeded.

Pull request description:
Fail on parsing of invalid topologies
This fixes #148 

Patch:
diff --git a/src/main/java/com/purbon/kafka/topology/TopologyObjectBuilder.java b/src/main/java/com/purbon/kafka/topology/TopologyObjectBuilder.java
index a531a58..ff20077 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyObjectBuilder.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyObjectBuilder.java
@@ -1,6 +1,5 @@
 package com.purbon.kafka.topology;
 
-import com.purbon.kafka.topology.model.Impl.TopologyImpl;
 import com.purbon.kafka.topology.model.PlanMap;
 import com.purbon.kafka.topology.model.Topology;
 import com.purbon.kafka.topology.serdes.PlanMapSerdes;
@@ -11,13 +10,9 @@ import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.List;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 
 public class TopologyObjectBuilder {
 
-  private static final Logger LOGGER = LogManager.getLogger(TopologyObjectBuilder.class);
-
   public static Topology build(String fileOrDir) throws IOException {
     return build(fileOrDir, "", new TopologyBuilderConfig());
   }
@@ -60,15 +55,7 @@ public class TopologyObjectBuilder {
     if (isDir) {
       Files.list(Paths.get(fileOrDir))
           .sorted()
-          .map(
-              path -> {
-                try {
-                  return parser.deserialise(path.toFile());
-                } catch (IOException e) {
-                  LOGGER.error(e);
-                  return new TopologyImpl();
-                }
-              })
+          .map(path -> parser.deserialise(path.toFile()))
           .forEach(subTopology -> topologies.add(subTopology));
     } else {
       Topology firstTopology = parser.deserialise(new File(fileOrDir));
diff --git a/src/main/java/com/purbon/kafka/topology/exceptions/TopologyParsingException.java b/src/main/java/com/purbon/kafka/topology/exceptions/TopologyParsingException.java
new file mode 100644
index 0000000..19d999f
--- /dev/null
+++ b/src/main/java/com/purbon/kafka/topology/exceptions/TopologyParsingException.java
@@ -0,0 +1,7 @@
+package com.purbon.kafka.topology.exceptions;
+
+public class TopologyParsingException extends RuntimeException {
+  public TopologyParsingException(String msg, Throwable e) {
+    super(msg, e);
+  }
+}
diff --git a/src/main/java/com/purbon/kafka/topology/serdes/TopologySerdes.java b/src/main/java/com/purbon/kafka/topology/serdes/TopologySerdes.java
index 6ee51c7..c9fc528 100644
--- a/src/main/java/com/purbon/kafka/topology/serdes/TopologySerdes.java
+++ b/src/main/java/com/purbon/kafka/topology/serdes/TopologySerdes.java
@@ -6,6 +6,7 @@ import com.fasterxml.jackson.databind.module.SimpleModule;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
 import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
 import com.purbon.kafka.topology.TopologyBuilderConfig;
+import com.purbon.kafka.topology.exceptions.TopologyParsingException;
 import com.purbon.kafka.topology.model.Impl.TopicImpl;
 import com.purbon.kafka.topology.model.PlanMap;
 import com.purbon.kafka.topology.model.Topology;
@@ -41,12 +42,21 @@ public class TopologySerdes {
     mapper = ObjectMapperFactory.build(type, config, plans);
   }
 
-  public Topology deserialise(File file) throws IOException {
-    return mapper.readValue(file, Topology.class);
+  public Topology deserialise(File file) {
+    try {
+      return mapper.readValue(file, Topology.class);
+    } catch (IOException e) {
+      throw new TopologyParsingException(
+          "Failed to deserialize topology from " + file.getPath(), e);
+    }
   }
 
-  public Topology deserialise(String content) throws IOException {
-    return mapper.readValue(content, Topology.class);
+  public Topology deserialise(String content) {
+    try {
+      return mapper.readValue(content, Topology.class);
+    } catch (IOException e) {
+      throw new TopologyParsingException("Failed to deserialize topology from " + content, e);
+    }
   }
 
   public String serialise(Topology topology) throws JsonProcessingException {


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java b/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
index f24bc17..20d7b37 100644
--- a/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
@@ -8,6 +8,7 @@ import static org.mockito.Mockito.*;
 import com.purbon.kafka.topology.BackendController.Mode;
 import com.purbon.kafka.topology.api.adminclient.TopologyBuilderAdminClient;
 import com.purbon.kafka.topology.backend.RedisBackend;
+import com.purbon.kafka.topology.exceptions.TopologyParsingException;
 import com.purbon.kafka.topology.utils.TestUtils;
 import java.io.IOException;
 import java.util.HashMap;
@@ -69,7 +70,7 @@ public class KafkaTopologyBuilderTest {
     verify(topologyAdminClient, times(1)).close();
   }
 
-  @Test(expected = IOException.class)
+  @Test(expected = TopologyParsingException.class)
   public void verifyProblematicParametersTest() throws Exception {
     String file = "fileThatDoesNotExist.yaml";
     TopologyBuilderConfig builderConfig = new TopologyBuilderConfig(cliOps, props);
diff --git a/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java b/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java
index 7cc0f3c..7964bd3 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java
@@ -3,6 +3,7 @@ package com.purbon.kafka.topology;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 
+import com.purbon.kafka.topology.exceptions.TopologyParsingException;
 import com.purbon.kafka.topology.model.Topic;
 import com.purbon.kafka.topology.model.Topology;
 import com.purbon.kafka.topology.utils.TestUtils;
@@ -58,4 +59,16 @@ public class TopologyObjectBuilderTest {
     String descriptorFile = TestUtils.getResourceFilename("/descriptor-with-plans.yaml");
     TopologyObjectBuilder.build(descriptorFile);
   }
+
+  @Test(expected = TopologyParsingException.class)
+  public void testInvalidTopology() throws IOException {
+    String descriptorFile = TestUtils.getResourceFilename("/errors_dir/descriptor-with-errors.yaml");
+    TopologyObjectBuilder.build(descriptorFile);
+  }
+
+  @Test(expected = TopologyParsingException.class)
+  public void testInvalidTopologyFromDir() throws IOException {
+    String dirPath = TestUtils.getResourceFilename("/errors_dir");
+    TopologyObjectBuilder.build(dirPath);
+  }
 }
diff --git a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
index f4011a3..87dfa4e 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
@@ -8,6 +8,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
+import com.purbon.kafka.topology.exceptions.TopologyParsingException;
 import com.purbon.kafka.topology.model.Impl.ProjectImpl;
 import com.purbon.kafka.topology.model.Impl.TopicImpl;
 import com.purbon.kafka.topology.model.Impl.TopologyImpl;
@@ -40,7 +41,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testDynamicFirstLevelAttributes() throws IOException {
+  public void testDynamicFirstLevelAttributes() {
     Topology topology =
         parser.deserialise(TestUtils.getResourceFile("/descriptor-with-others.yml"));
     Project project = topology.getProjects().get(0);
@@ -154,18 +155,18 @@ public class TopologySerdesTest {
     assertEquals(topic2.getDataType(), serdesTopic2.getDataType());
   }
 
-  @Test(expected = IOException.class)
-  public void testTopologyWithNoTeam() throws IOException {
+  @Test(expected = TopologyParsingException.class)
+  public void testTopologyWithNoTeam() {
     parser.deserialise(TestUtils.getResourceFile("/descriptor-with-no-context.yaml"));
   }
 
-  @Test(expected = IOException.class)
-  public void testTopologyWithNoProject() throws IOException {
+  @Test(expected = TopologyParsingException.class)
+  public void testTopologyWithNoProject() {
     parser.deserialise(TestUtils.getResourceFile("/descriptor-with-no-project.yaml"));
   }
 
   @Test
-  public void testCoreElementsProcessing() throws IOException {
+  public void testCoreElementsProcessing() {
     Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
 
     Project project = topology.getProjects().get(0);
@@ -180,7 +181,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testStreamsApps() throws IOException {
+  public void testStreamsApps() {
     Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
 
     Project project1 = topology.getProjects().get(0);
@@ -194,12 +195,12 @@ public class TopologySerdesTest {
   }
 
   @Test(expected = IllegalArgumentException.class)
-  public void testSchemaSerdes() throws IOException {
+  public void testSchemaSerdes() {
     parser.deserialise(TestUtils.getResourceFile("/descriptor-wrong-schemas.yaml"));
   }
 
   @Test
-  public void testPlaformProcessing() throws IOException {
+  public void testPlaformProcessing() {
     Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
 
     assertEquals("contextOrg", topology.getContext());
@@ -220,7 +221,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testOnlyTopics() throws IOException {
+  public void testOnlyTopics() {
     Topology topology =
         parser.deserialise(TestUtils.getResourceFile("/descriptor-only-topics.yaml"));
 
@@ -232,7 +233,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testRBACTopics() throws IOException {
+  public void testRBACTopics() {
     Topology topology =
         parser.deserialise(TestUtils.getResourceFile("/descriptor-with-rbac-topics.yaml"));
 
@@ -255,7 +256,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testWithRBACDescriptor() throws IOException {
+  public void testWithRBACDescriptor() {
     Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor-with-rbac.yaml"));
     Project myProject = topology.getProjects().get(0);
 
@@ -288,7 +289,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testTopicNameWithCustomSeparator() throws IOException {
+  public void testTopicNameWithCustomSeparator() {
     Map<String, String> cliOps = new HashMap<>();
     cliOps.put(BROKERS_OPTION, "");
     cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, "/fooBar");
@@ -312,7 +313,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testTopicNameWithCustomPattern() throws IOException {
+  public void testTopicNameWithCustomPattern() {
     Map<String, String> cliOps = new HashMap<>();
     cliOps.put(BROKERS_OPTION, "");
     cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, "/fooBar");
@@ -336,7 +337,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testJsonDescriptorFileSerdes() throws IOException {
+  public void testJsonDescriptorFileSerdes() {
     TopologySerdes parser = new TopologySerdes(new TopologyBuilderConfig(), FileType.JSON);
     Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.json"));
 
diff --git a/src/test/resources/errors_dir/descriptor-with-errors.yaml b/src/test/resources/errors_dir/descriptor-with-errors.yaml
new file mode 100644
index 0000000..aeb2c72
--- /dev/null
+++ b/src/test/resources/errors_dir/descriptor-with-errors.yaml
@@ -0,0 +1,22 @@
+---
+context: "contextOrg"
+source: "source"
+projects:
+  - name: "foo"
+    streams:
+      - principal: "User:streams"
+        unknownProperty: "catchMe"
+        topics:
+          read:
+            - foo
+          write:
+            - bar
+    topics:
+      - name: "foo"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: "bar"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
diff --git a/src/test/resources/errors_dir/descriptor.yaml b/src/test/resources/errors_dir/descriptor.yaml
new file mode 100644
index 0000000..0cb7005
--- /dev/null
+++ b/src/test/resources/errors_dir/descriptor.yaml
@@ -0,0 +1,11 @@
+---
+context: "contextOrg"
+source: "source"
+projects:
+  - name: "bar"
+    topics:
+      - dataType: "avro"
+        name: "bar"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
\ No newline at end of file