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

Issue:
KTB should support not only YAML descriptors, but as well JSON
As a user of the Kafka Topology Builder I don't have special love for YAML and I would love to have it understand as well JSON format for the descriptor files. 

Pull request description:
Enable JSON and YAML as Topology descriptor file formats.
This pull request add code to handle other file descriptor formats such as:

- [x] JSON

*xml* for a future pull request

(*work in progress*, xml needs to redefine the custom parser)

*NOTE*: This future is for the upcoming 2.0 branch

closes #107 

Patch:
diff --git a/docs/the-descriptor-files.rst b/docs/the-descriptor-files.rst
index f4c3892..a82475e 100644
--- a/docs/the-descriptor-files.rst
+++ b/docs/the-descriptor-files.rst
@@ -8,7 +8,15 @@ Because users might have complex deployments, the structure is flexible enough t
 The File format
 -----------
 
-Currently the tool only supports reading YAML files. In the future this might change.
+Currently the tool supports reading YAML and JSON files. **Note** that all topologies will need to be using the same file format, a mixture is not supported.
+
+In this guide we will use yaml as the core format, for JSON examples please refer to the examples directory.
+
+The file type is configured using the *topology.file.type* configuration variable.
+
+```
+topology.file.type=JSON
+```
 
 Getting started
 -----------
diff --git a/example/descriptor.json b/example/descriptor.json
new file mode 100644
index 0000000..3b4abdb
--- /dev/null
+++ b/example/descriptor.json
@@ -0,0 +1,20 @@
+{
+  "context": "team",
+  "projects": [{
+    "name": "foo",
+    "topics": [{
+      "dataType": "json",
+      "name": "foo",
+      "config": {
+        "replication.factor": "2",
+        "num.partitions": "3"
+      }
+    }, {
+      "name": "topic2",
+      "config": {
+        "replication.factor": "2",
+        "num.partitions": "3"
+      }
+    }]
+  }]
+}
\ No newline at end of file
diff --git a/src/main/conf/topology-builder.properties b/src/main/conf/topology-builder.properties
index 95d5c4d..671d62b 100644
--- a/src/main/conf/topology-builder.properties
+++ b/src/main/conf/topology-builder.properties
@@ -24,6 +24,8 @@
 #ssl.keystore.type=PKCS12
 #ssl.key.password=test1234
 
+## Topology files type, the default value is YAML
+# topology.file.type=JSON
 
 ## SASL connection configuration
 #sasl.mechanism=PLAIN
diff --git a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
index 5e5139a..2cc79ce 100644
--- a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
+++ b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
@@ -28,9 +28,7 @@ import com.purbon.kafka.topology.model.users.KStream;
 import com.purbon.kafka.topology.model.users.Schemas;
 import com.purbon.kafka.topology.model.users.platform.ControlCenterInstance;
 import com.purbon.kafka.topology.model.users.platform.SchemaRegistryInstance;
-import com.purbon.kafka.topology.roles.SimpleAclsProvider;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
-import java.io.IOException;
 import java.io.PrintStream;
 import java.util.ArrayList;
 import java.util.List;
@@ -65,8 +63,26 @@ public class AccessControlManager {
     this.config = config;
   }
 
-  public void apply(final Topology topology, ExecutionPlan plan) throws IOException {
+  /**
+   * Main apply method, append to the execution plan the necessary bindings to update the access
+   * control
+   *
+   * @param topology A topology file descriptor
+   * @param plan An Execution plan
+   */
+  public void apply(final Topology topology, ExecutionPlan plan) {
+    List<Action> actions = buildProjectActions(topology);
+    actions.addAll(buildPlatformLevelActions(topology));
+    buildUpdateBindingsActions(actions, plan.getBindings()).forEach(plan::add);
+  }
 
+  /**
+   * Build the core list of actions builders for creating access control rules
+   *
+   * @param topology A topology file
+   * @return List<Action> A list of actions required based on the parameters
+   */
+  public List<Action> buildProjectActions(Topology topology) {
     List<Action> actions = new ArrayList<>();
 
     for (Project project : topology.getProjects()) {
@@ -91,14 +107,17 @@ public class AccessControlManager {
       // Setup global Kafka Stream Access control lists
       String topicPrefix = project.namePrefix();
       for (KStream app : project.getStreams()) {
-        Action action = syncApplicationAcls(app, topicPrefix);
-        actions.add(action);
+        syncApplicationAcls(app, topicPrefix).ifPresent(actions::add);
       }
       for (Connector connector : project.getConnectors()) {
-        Action action = syncApplicationAcls(connector, topicPrefix);
-        actions.add(action);
-        if (connector.getConnectors().isPresent())
-          actions.add(new BuildBindingsForConnectorAuthorization(bindingsBuilder, connector));
+        syncApplicationAcls(connector, topicPrefix).ifPresent(actions::add);
+        connector
+            .getConnectors()
+            .ifPresent(
+                (list) -> {
+                  actions.add(
+                      new BuildBindingsForConnectorAuthorization(bindingsBuilder, connector));
+                });
       }
 
       for (Schemas schemaAuthorization : project.getSchemas()) {
@@ -107,50 +126,62 @@ public class AccessControlManager {
 
       syncRbacRawRoles(project.getRbacRawRoles(), topicPrefix, actions);
     }
+    return actions;
+  }
 
-    syncPlatformAcls(topology, actions);
+  /**
+   * Build a list of actions required to create or delete necessary bindings
+   *
+   * @param actions List of pre computed actions based on a topology
+   * @param bindings List of current bindings available in the cluster
+   * @return List<Action> list of actions necessary to update the cluster
+   */
+  public List<Action> buildUpdateBindingsActions(
+      List<Action> actions, Set<TopologyAclBinding> bindings) {
 
-    // Main actions now should be setup to create low level bindings
+    List<Action> updateActions = new ArrayList<>();
 
     Set<TopologyAclBinding> allFinalBindings =
-        actions.stream().flatMap(executeToFunction()).collect(Collectors.toSet());
+        actions.stream().flatMap(actionApplyFunction()).collect(Collectors.toSet());
 
     // Diff of bindings, so we only create what is not already created in the cluster.
     Set<TopologyAclBinding> bindingsToBeCreated =
         allFinalBindings.stream()
-            .filter(binding -> !plan.getBindings().contains(binding))
+            .filter(binding -> !bindings.contains(binding))
             .collect(Collectors.toSet());
 
     CreateBindings createBindings = new CreateBindings(controlProvider, bindingsToBeCreated);
-    plan.add(createBindings);
+    updateActions.add(createBindings);
 
     if (config.allowDelete()) {
       // clear acls that does not appear anymore in the new generated list,
       // but where previously created
       Set<TopologyAclBinding> bindingsToDelete =
-          plan.getBindings().stream()
+          bindings.stream()
               .filter(binding -> !allFinalBindings.contains(binding))
               .collect(Collectors.toSet());
 
       ClearBindings clearBindings = new ClearBindings(controlProvider, bindingsToDelete);
-      plan.add(clearBindings);
+      updateActions.add(clearBindings);
     }
+    return updateActions;
   }
 
-  private Function<Action, Stream<TopologyAclBinding>> executeToFunction() {
+  private Function<Action, Stream<TopologyAclBinding>> actionApplyFunction() {
     return action -> {
       try {
         action.run();
         return action.getBindings().stream();
       } catch (Exception ex) {
         LOGGER.error(ex);
-        return new ArrayList<TopologyAclBinding>().stream();
+        return Stream.empty();
       }
     };
   }
 
-  private void syncPlatformAcls(final Topology topology, List<Action> actions) throws IOException {
-    // Sync platform relevant Access Control List.
+  // Sync platform relevant Access Control List.
+  public List<Action> buildPlatformLevelActions(final Topology topology) {
+    List<Action> actions = new ArrayList<>();
     Platform platform = topology.getPlatform();
 
     // Set cluster level ACLs
@@ -165,6 +196,7 @@ public class AccessControlManager {
     for (ControlCenterInstance controlCenter : platform.getControlCenter().getInstances()) {
       actions.add(new BuildBindingsForControlCenter(bindingsBuilder, controlCenter));
     }
+    return actions;
   }
 
   private void syncClusterLevelRbac(
@@ -190,14 +222,14 @@ public class AccessControlManager {
                             bindingsBuilder, principal, predefinedRole, topicPrefix))));
   }
 
-  private Action syncApplicationAcls(DynamicUser app, String topicPrefix) throws IOException {
+  private Optional<Action> syncApplicationAcls(DynamicUser app, String topicPrefix) {
+    Action action = null;
     if (app instanceof KStream) {
-      return new BuildBindingsForKStreams(bindingsBuilder, (KStream) app, topicPrefix);
+      action = new BuildBindingsForKStreams(bindingsBuilder, (KStream) app, topicPrefix);
     } else if (app instanceof Connector) {
-      return new BuildBindingsForKConnect(bindingsBuilder, (Connector) app, topicPrefix);
-    } else {
-      throw new IOException("Wrong dynamic app used.");
+      action = new BuildBindingsForKConnect(bindingsBuilder, (Connector) app, topicPrefix);
     }
+    return Optional.ofNullable(action);
   }
 
   public void printCurrentState(PrintStream out) {
@@ -207,11 +239,7 @@ public class AccessControlManager {
         .forEach(
             (topic, aclBindings) -> {
               out.println(topic);
-              aclBindings.forEach(binding -> out.println(binding));
+              aclBindings.forEach(out::println);
             });
   }
-
-  public void setAclsProvider(SimpleAclsProvider aclsProvider) {
-    this.controlProvider = aclsProvider;
-  }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
index 2449941..a8f700a 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
@@ -7,6 +7,7 @@ import com.purbon.kafka.topology.exceptions.ConfigurationException;
 import com.purbon.kafka.topology.model.Project;
 import com.purbon.kafka.topology.model.Topic;
 import com.purbon.kafka.topology.model.Topology;
+import com.purbon.kafka.topology.serdes.TopologySerdes.FileType;
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
 import java.util.HashMap;
@@ -59,6 +60,8 @@ public class TopologyBuilderConfig {
   static final String TOPIC_PREFIX_SEPARATOR_CONFIG = "topology.topic.prefix.separator";
   static final String TOPOLOGY_VALIDATIONS_CONFIG = "topology.validations";
 
+  static final String TOPOLOGY_FILE_TYPE = "topology.file.type";
+
   private final Map<String, String> cliParams;
   private Config config;
 
@@ -270,4 +273,8 @@ public class TopologyBuilderConfig {
   public boolean isDryRun() {
     return Boolean.parseBoolean(cliParams.getOrDefault(DRY_RUN_OPTION, "false"));
   }
+
+  public FileType getTopologyFileType() {
+    return config.getEnum(FileType.class, TOPOLOGY_FILE_TYPE);
+  }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/TopologyDescriptorBuilder.java b/src/main/java/com/purbon/kafka/topology/TopologyDescriptorBuilder.java
index 79f6c1b..6ca812b 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyDescriptorBuilder.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyDescriptorBuilder.java
@@ -9,9 +9,13 @@ 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 TopologyDescriptorBuilder {
 
+  private static final Logger LOGGER = LogManager.getLogger(TopologyDescriptorBuilder.class);
+
   public static Topology build(String fileOrDir) throws IOException {
     return build(fileOrDir, new TopologyBuilderConfig());
   }
@@ -44,7 +48,7 @@ public class TopologyDescriptorBuilder {
                 try {
                   return parser.deserialise(path.toFile());
                 } catch (IOException e) {
-                  e.printStackTrace();
+                  LOGGER.error(e);
                   return new TopologyImpl();
                 }
               })
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 825e6fb..9542a14 100644
--- a/src/main/java/com/purbon/kafka/topology/serdes/TopologySerdes.java
+++ b/src/main/java/com/purbon/kafka/topology/serdes/TopologySerdes.java
@@ -15,18 +15,21 @@ public class TopologySerdes {
 
   private ObjectMapper mapper;
 
+  public enum FileType {
+    JSON,
+    YAML
+  }
+
   public TopologySerdes() {
-    this(new TopologyBuilderConfig());
+    this(new TopologyBuilderConfig(), FileType.YAML);
   }
 
   public TopologySerdes(TopologyBuilderConfig config) {
-    mapper = new ObjectMapper(new YAMLFactory());
-    SimpleModule module = new SimpleModule();
-    module.addDeserializer(Topology.class, new TopologyCustomDeserializer(config));
-    module.addDeserializer(TopicImpl.class, new TopicCustomDeserializer(config));
-    mapper.registerModule(module);
-    mapper.registerModule(new Jdk8Module());
-    mapper.findAndRegisterModules();
+    this(config, config.getTopologyFileType());
+  }
+
+  public TopologySerdes(TopologyBuilderConfig config, FileType type) {
+    mapper = ObjectMapperFactory.build(type, config);
   }
 
   public Topology deserialise(File file) throws IOException {
@@ -40,4 +43,24 @@ public class TopologySerdes {
   public String serialise(Topology topology) throws JsonProcessingException {
     return mapper.writeValueAsString(topology);
   }
+
+  private static class ObjectMapperFactory {
+
+    public static ObjectMapper build(FileType type, TopologyBuilderConfig config) {
+      ObjectMapper mapper;
+      if (type.equals(FileType.JSON)) {
+        mapper = new ObjectMapper();
+      } else {
+        mapper = new ObjectMapper(new YAMLFactory());
+      }
+
+      SimpleModule module = new SimpleModule();
+      module.addDeserializer(Topology.class, new TopologyCustomDeserializer(config));
+      module.addDeserializer(TopicImpl.class, new TopicCustomDeserializer(config));
+      mapper.registerModule(module);
+      mapper.registerModule(new Jdk8Module());
+      mapper.findAndRegisterModules();
+      return mapper;
+    }
+  }
 }
diff --git a/src/main/resources/reference.conf b/src/main/resources/reference.conf
index e1831ad..be1b6b0 100644
--- a/src/main/resources/reference.conf
+++ b/src/main/resources/reference.conf
@@ -1,4 +1,7 @@
 topology {
+  file {
+    type = "YAML"
+  }
   validations = []
   builder {
     access.control.class = "com.purbon.kafka.topology.roles.SimpleAclsProvider"


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 070e6a1..cbfad5f 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
@@ -21,6 +21,7 @@ import com.purbon.kafka.topology.model.users.Producer;
 import com.purbon.kafka.topology.model.users.platform.ControlCenterInstance;
 import com.purbon.kafka.topology.model.users.platform.SchemaRegistryInstance;
 import com.purbon.kafka.topology.serdes.TopologySerdes;
+import com.purbon.kafka.topology.serdes.TopologySerdes.FileType;
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.net.URL;
@@ -304,6 +305,18 @@ public class TopologySerdesTest {
     assertEquals("source.contextOrg.foo.bar", p.getTopics().get(1).toString());
   }
 
+  @Test
+  public void testJsonDescriptorFileSerdes() throws IOException, URISyntaxException {
+
+    TopologySerdes parser = new TopologySerdes(new TopologyBuilderConfig(), FileType.JSON);
+    URL descriptorWithOptionals = getClass().getResource("/descriptor.json");
+    Topology topology = parser.deserialise(Paths.get(descriptorWithOptionals.toURI()).toFile());
+
+    assertEquals(1, topology.getProjects().size());
+    assertEquals("foo", topology.getProjects().get(0).getName());
+    assertEquals(2, topology.getProjects().get(0).getTopics().size());
+  }
+
   private List<Project> buildProjects() {
 
     Project project = new ProjectImpl("project");
diff --git a/src/test/resources/descriptor.json b/src/test/resources/descriptor.json
new file mode 100644
index 0000000..3b4abdb
--- /dev/null
+++ b/src/test/resources/descriptor.json
@@ -0,0 +1,20 @@
+{
+  "context": "team",
+  "projects": [{
+    "name": "foo",
+    "topics": [{
+      "dataType": "json",
+      "name": "foo",
+      "config": {
+        "replication.factor": "2",
+        "num.partitions": "3"
+      }
+    }, {
+      "name": "topic2",
+      "config": {
+        "replication.factor": "2",
+        "num.partitions": "3"
+      }
+    }]
+  }]
+}
\ No newline at end of file