make the datatype field as optional

kafka-ops · Dec 20, 2019 · 15d7cb9 · 15d7cb9
1 parent 28c3f7c
commit 15d7cb9
Show file tree

Hide file tree

Showing 5 changed files with 86 additions and 29 deletions.
diff --git a/example/descriptor.yaml b/example/descriptor.yaml
@@ -1,18 +1,18 @@
-
 ---
 team: "team"
 source: "source"
 projects:
   - name: "foo"
     zookeepers: []
     consumers:
-      - principal: "User:app0"
-      - principal: "User:app1"
+      - principal: "User:App0"
+      - principal: "User:App1"
     producers: []
     streams:
       - principal: "User:App0"
         topics:
           read:
+            - "topicA"
             - "topicB"
           write:
             - "topicC"
@@ -29,12 +29,11 @@ projects:
             - "topicC"
             - "topicD"
     topics:
-      - dataType: "unknown"
-        name: "foo"
+      - name: "foo"
         config:
           replication.factor: "1"
           num.partitions: "1"
-      - dataType: "unknown"
+      - dataType: "avro"
         name: "bar"
         config:
           replication.factor: "1"
@@ -46,7 +45,7 @@ projects:
     streams: []
     connectors: []
     topics:
-      - dataType: "unknown"
+      - dataType: "avro"
         name: "bar"
         config:
           replication.factor: "1"

diff --git a/pom.xml b/pom.xml
@@ -169,6 +169,11 @@
   </reporting>
 
 
+  <properties>
+    <jackson.version>2.10.0</jackson.version>
+    <kafka.version>2.3.0</kafka.version>
+    <log4j.version>2.12.1</log4j.version>
+  </properties>
   <dependencies>
 
     <!-- https://mvnrepository.com/artifact/org.apache.maven.plugins/maven-pmd-plugin -->
@@ -189,36 +194,48 @@
     <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka-clients</artifactId>
-      <version>2.3.0</version>
+      <version>${kafka.version}</version>
     </dependency>
 
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-databind</artifactId>
-      <version>2.10.0</version>
+      <version>${jackson.version}</version>
     </dependency>
 
     <dependency>
       <groupId>com.fasterxml.jackson.dataformat</groupId>
       <artifactId>jackson-dataformat-yaml</artifactId>
-      <version>2.10.0</version>
+      <version>${jackson.version}</version>
     </dependency>
 
     <dependency>
       <groupId>com.fasterxml.jackson.datatype</groupId>
       <artifactId>jackson-datatype-jsr310</artifactId>
-      <version>2.10.0</version>
+      <version>${jackson.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>com.fasterxml.jackson.datatype</groupId>
+      <artifactId>jackson-datatype-jdk8</artifactId>
+      <version>${jackson.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-core</artifactId>
+      <version>${jackson.version}</version>
     </dependency>
 
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-api</artifactId>
-      <version>2.12.1</version>
+      <version>${log4j.version}</version>
     </dependency>
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-core</artifactId>
-      <version>2.12.1</version>
+      <version>${log4j.version}</version>
     </dependency>
 
     <dependency>

diff --git a/src/main/java/com/purbon/kafka/topology/TopologySerdes.java b/src/main/java/com/purbon/kafka/topology/TopologySerdes.java
@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
 import com.purbon.kafka.topology.model.Project;
 import com.purbon.kafka.topology.model.Topic;
 import com.purbon.kafka.topology.model.Topology;
@@ -16,6 +17,7 @@ public class TopologySerdes {
 
   public TopologySerdes() {
     mapper = new ObjectMapper(new YAMLFactory());
+    mapper.registerModule(new Jdk8Module());
     mapper.findAndRegisterModules();
   }
 

diff --git a/src/main/java/com/purbon/kafka/topology/model/Topic.java b/src/main/java/com/purbon/kafka/topology/model/Topic.java
@@ -1,35 +1,38 @@
 package com.purbon.kafka.topology.model;
 
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
 import com.purbon.kafka.topology.TopicManager;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 
 public class Topic {
 
-  public static final String UNKNOWN_DATATYPE = "unknown";
   public static final String DEFAULT_TOPIC_NAME = "default";
-  private String dataType;
+  @JsonInclude(Include.NON_EMPTY)
+  private Optional<String> dataType;
   private String name;
   private HashMap<String, String> config;
 
   private Project project;
 
   public Topic(String name) {
-    this(name, UNKNOWN_DATATYPE, new HashMap<>());
+    this(name, Optional.empty(), new HashMap<>());
   }
 
   public Topic(String name, String dataType) {
-    this(name, dataType, new HashMap<>());
+    this(name, Optional.of(dataType), new HashMap<>());
   }
 
-  public Topic(String name, String dataType, HashMap<String, String> config) {
+  public Topic(String name, Optional<String> dataType, HashMap<String, String> config) {
     this.name = name;
     this.dataType = dataType;
     this.config = config;
   }
 
   public Topic() {
-    this(DEFAULT_TOPIC_NAME, UNKNOWN_DATATYPE, new HashMap<>());
+    this(DEFAULT_TOPIC_NAME, Optional.empty(), new HashMap<>());
   }
 
   public String getName() {
@@ -42,9 +45,9 @@ private String toString(Project project) {
         .append(".")
         .append(getName());
 
-    if (!getDataType().equals(UNKNOWN_DATATYPE)) {
+    if (getDataType().isPresent()) {
       sb.append(".")
-          .append(getDataType());
+          .append(getDataType().get());
     }
 
     return sb.toString();
@@ -73,7 +76,7 @@ public Map<String, String> rawConfig() {
     return getConfig();
   }
 
-  public String getDataType() {
+  public Optional<String> getDataType() {
     return dataType;
   }
 

diff --git a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
@@ -53,15 +53,14 @@ public void testTopicConfigSerdes() throws IOException {
     Topic topic = new Topic();
     topic.setName("foo");
     HashMap<String, String> topicConfig = new HashMap<>();
-    topicConfig.put("num.partitions", "3");
-    topicConfig.put("replication.factor", "2");
+    topicConfig.put("num.partitions", "1");
+    topicConfig.put("replication.factor", "1");
     topic.setConfig(topicConfig);
 
-    Topic topicBar = new Topic();
-    topicBar.setName("bar");
+    Topic topicBar = new Topic("bar", "avro");
     HashMap<String, String> topicBarConfig = new HashMap<>();
-    topicBarConfig.put("num.partitions", "3");
-    topicBarConfig.put("replication.factor", "2");
+    topicBarConfig.put("num.partitions", "1");
+    topicBarConfig.put("replication.factor", "1");
     topicBar.setConfig(topicBarConfig);
 
     Project project = new Project("foo");
@@ -99,6 +98,7 @@ public void testTopicConfigSerdes() throws IOException {
     topology.setProjects(Arrays.asList(project, project2));
 
     String topologyYamlString = parser.serialise(topology);
+    System.out.println(topologyYamlString);
 
     Topology deserTopology = parser.deserialise(topologyYamlString);
 
@@ -107,9 +107,45 @@ public void testTopicConfigSerdes() throws IOException {
 
     Assert.assertEquals(topic.getName(), serdesTopic.getName());
     Assert.assertEquals(topic.getConfig().get("num.partitions"), serdesTopic.getConfig().get("num.partitions"));
-
   }
 
+  @Test
+  public void testTopicWithDataType() throws IOException {
+
+    Project project = new Project("foo");
+
+    Topology topology = new Topology();
+    topology.setTeam("team");
+    topology.setSource("source");
+
+    project.setTopology(topology);
+    topology.addProject(project);
+
+    Topic topic = new Topic("foo", "json");
+    HashMap<String, String> topicConfig = new HashMap<>();
+    topicConfig.put("num.partitions", "3");
+    topicConfig.put("replication.factor", "2");
+    topic.setConfig(topicConfig);
+
+    project.addTopic(topic);
+
+    Topic topic2 = new Topic("topic2");
+    topic.setConfig(topicConfig);
+    project.addTopic(topic2);
+
+    String topologyYamlString = parser.serialise(topology);
+    Topology deserTopology = parser.deserialise(topologyYamlString);
+
+    Project serdesProject = deserTopology.getProjects().get(0);
+    Topic serdesTopic = serdesProject.getTopics().get(0);
+
+    Assert.assertEquals(topic.getDataType(), serdesTopic.getDataType());
+    Assert.assertEquals(topic.getDataType().get(), serdesTopic.getDataType().get());
+
+    Topic serdesTopic2 = serdesProject.getTopics().get(1);
+    Assert.assertEquals(topic2.getDataType(), serdesTopic2.getDataType());
+
+  }
 
 
   private List<Project> buildProjects() {