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

Issue:
Metadata on topics and principals
This feature request is overlapping with #100, but expands on the subject.

**Is your feature request related to a problem? Please describe.**

When looking at what is deployed to a secure Kafka cluster there is often a need to know something more than topic names, schemas and which users (often represented by certificates) can read/write the topics. Important questions to answer can be:
- Who is the owner of this topic?
- What system is behind (the owner) of this principal? (Including consumer, producer, streams and connector principals)
- What topics exists within a given sub domain of our information model? 
- Which systems are consuming data from a given topic? Which system is the producer? 


**Describe the solution you'd like**

All of the questions above can be answered by adding your own metadata to the topology. By annotating topics and principals you can either find the information by looking into the topology files or maybe also by writing your own tools to parse the topology and produce documentation based on metadata.

In addition the metadata should be loaded into the Java model so that validations based on metadata values can be implemented.

@purbon: I have created a draft PR already to give you a clear idea of what we want. We really need this feature now and I'm ready to contribute! :-)

Pull request description:
Add topology metadata support for topics and users
This implement support for #199.  

* **Please check if the PR fulfills these requirements**
- [x] The commit messages are descriptive
- [x] Tests for the changes have been added (for bug fixes / features)
- [x] Docs have been added / updated (for bug fixes / features)
- [x] An issue has been created for the pull requests. Some issues might require previous discussion.

* **What kind of change does this PR introduce?** (Bug fix, feature, docs update, ...)
New feature

* **What is the current behavior?** (You can also link to an open issue here)
Not supported.

* **Does this PR introduce a breaking change?** 
No

closes #199 



Patch:
diff --git a/docs/the-descriptor-files.rst b/docs/the-descriptor-files.rst
index a82475e..a130b5c 100644
--- a/docs/the-descriptor-files.rst
+++ b/docs/the-descriptor-files.rst
@@ -65,6 +65,59 @@ The relevant properties are:
 
 more details can be found in the :doc:`config-values` section where the most important configuration details are explained.
 
+What can you do in the descriptor files?
+-----------
+
+The `example` directory in the project contains a set of example topologies that you can use to get an idea of how to
+use the different features. For example:
+* RBAC roles with Confluent Platform
+* Administration of schemas for the topics
+* Using plans to create common topic configurations
+* Add metadata to different elements in the topology
+
+Add metadata to your topology
+-----------
+
+In addition to definitions needed for creating the required topics, acls, rbac bindings etc in your Kafka cluster KTB
+also supports annotating your topology through metadata. This can allow you to make the descriptor files easier to read,
+e.g. for someone not into the details of what a given topic is all about.
+
+But you can also utilise it to do the following:
+* Add validations that also use metadata. E.g. combine topic metadata and topic name to enforce more refined topic naming rules.
+* Generate documentation from the your descriptor files (with some tool) and include metadata create better documentation on topics and users.
+
+You can add metadata to the following elements in the topology:
+* topics
+* consumers
+* producers
+* streams
+* connectors
+
+A short example is given below, have a look in the `example` directory for a more complete example.
+
+.. code-block:: YAML
+
+    ---
+    context: "context"
+    projects:
+      - name: "foo with metadata"
+        consumers:
+          - principal: "User:App0"
+            metadata:
+              system: "System0"
+        producers:
+          - principal: "User:App1"
+            metadata:
+              system: "System1"
+        topics:
+          - name: "topicA"
+            metadata:
+              domain: "Sales"
+              owner: "DepartmentA"
+            config:
+              replication.factor: "3"
+              num.partitions: "3"
+
 Manage only topics, the optional files
 -----------
 
diff --git a/example/descriptor-with-metadata.yaml b/example/descriptor-with-metadata.yaml
new file mode 100644
index 0000000..f06114e
--- /dev/null
+++ b/example/descriptor-with-metadata.yaml
@@ -0,0 +1,51 @@
+---
+context: "context"
+projects:
+  - name: "foo with metadata"
+    consumers:
+      - principal: "User:App0"
+        metadata:
+          system: "System0"
+    producers:
+      - principal: "User:App1"
+        metadata:
+          system: "System1"
+    streams:
+      - principal: "User:StreamsApp1"
+        metadata:
+          contactInfo: "app1@company.com"
+        topics:
+          read:
+            - "topicA"
+          write:
+            - "topicB"
+    connectors:
+      - principal: "User:Connect1"
+        metadata:
+          system: "System3"
+          contactInfo: "system3@company.com"
+        status_topic: "status"
+        offset_topic: "offset"
+        configs_topic: "configs"
+        topics:
+          read:
+            - "topicB"
+    topics:
+      - name: "topicA"
+        metadata:
+          domain: "Sales"
+          owner: "DepartmentA"
+        consumers:
+          - principal: "User:App4"
+            metadata:
+              system: "System4"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: "topicB"
+        metadata:
+          domain: "ProductionStatistics"
+          owner: "DepartmentB"
+        dataType: "avro"
+        schemas:
+          value.schema.file: "schemas/bar-value.avsc"
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 9ab6874..d4b7cd9 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
@@ -35,6 +35,8 @@ public class TopicImpl implements Topic, Cloneable {
   @JsonInclude(Include.NON_EMPTY)
   private String plan;
 
+  private Map<String, String> metadata;
+
   private Map<String, String> config;
   @JsonIgnore private TopologyBuilderConfig appConfig;
   @JsonIgnore private Map<String, Object> context;
@@ -230,6 +232,14 @@ public class TopicImpl implements Topic, Cloneable {
     this.appConfig = appConfig;
   }
 
+  public Map<String, String> getMetadata() {
+    return metadata;
+  }
+
+  public void setMetadata(Map<String, String> metadata) {
+    this.metadata = metadata;
+  }
+
   @Override
   public Topic clone() {
     try {
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 65b933c..5e0ba10 100644
--- a/src/main/java/com/purbon/kafka/topology/model/Topic.java
+++ b/src/main/java/com/purbon/kafka/topology/model/Topic.java
@@ -45,5 +45,9 @@ public interface Topic {
 
   void setProducers(List<Producer> producers);
 
+  Map<String, String> getMetadata();
+
+  void setMetadata(Map<String, String> metadata);
+
   SubjectNameStrategy getSubjectNameStrategy();
 }
diff --git a/src/main/java/com/purbon/kafka/topology/model/User.java b/src/main/java/com/purbon/kafka/topology/model/User.java
index 03a3e4c..d74fb99 100644
--- a/src/main/java/com/purbon/kafka/topology/model/User.java
+++ b/src/main/java/com/purbon/kafka/topology/model/User.java
@@ -1,8 +1,11 @@
 package com.purbon.kafka.topology.model;
 
+import java.util.Map;
+
 public class User {
 
   private String principal;
+  private Map<String, String> metadata;
 
   public User() {
     this("");
@@ -19,4 +22,12 @@ public class User {
   public void setPrincipal(String principal) {
     this.principal = principal;
   }
+
+  public Map<String, String> getMetadata() {
+    return metadata;
+  }
+
+  public void setMetadata(Map<String, String> metadata) {
+    this.metadata = metadata;
+  }
 }
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 f7a1011..fed7a18 100644
--- a/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
+++ b/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
@@ -74,20 +74,7 @@ public class TopicCustomDeserializer extends StdDeserializer<TopicImpl> {
     Optional<JsonNode> optionalDataTypeNode = Optional.ofNullable(rootNode.get("dataType"));
     Optional<String> optionalDataType = optionalDataTypeNode.map(JsonNode::asText);
 
-    Optional<JsonNode> optionalConfigNode = Optional.ofNullable(rootNode.get("config"));
-    Map<String, String> config =
-        optionalConfigNode
-            .map(
-                node -> {
-                  Map<String, String> map = new HashMap<>();
-                  Iterator<Map.Entry<String, JsonNode>> it = node.fields();
-                  while (it.hasNext()) {
-                    Map.Entry<String, JsonNode> entry = it.next();
-                    map.put(entry.getKey(), entry.getValue().asText());
-                  }
-                  return map;
-                })
-            .orElse(new HashMap<>());
+    Map<String, String> config = getMap(rootNode.get("config"));
 
     Optional<JsonNode> optionalPlanLabel = Optional.ofNullable(rootNode.get("plan"));
     if (optionalPlanLabel.isPresent() && plans.size() == 0) {
@@ -154,11 +141,30 @@ public class TopicCustomDeserializer extends StdDeserializer<TopicImpl> {
 
     topic.setSchemas(schemas);
 
+    Map<String, String> metadata = getMap(rootNode.get("metadata"));
+    topic.setMetadata(metadata);
+
     LOGGER.debug(
         String.format("Topic %s with config %s has been created", topic.getName(), config));
     return topic;
   }
 
+  private Map<String, String> getMap(JsonNode jsonNode) {
+    Optional<JsonNode> optionalNode = Optional.ofNullable(jsonNode);
+    return optionalNode
+        .map(
+            node -> {
+              Map<String, String> map = new HashMap<>();
+              Iterator<Map.Entry<String, JsonNode>> it = node.fields();
+              while (it.hasNext()) {
+                Map.Entry<String, JsonNode> entry = it.next();
+                map.put(entry.getKey(), entry.getValue().asText());
+              }
+              return map;
+            })
+        .orElse(new HashMap<>());
+  }
+
   private Function<JsonNode, Either<ValidationException, TopicSchemas>> validateAndBuildSchemas(
       Topic topic) {
     return node -> {


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 458de90..2f26641 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
@@ -42,6 +42,23 @@ public class TopologySerdesTest {
     parser = new TopologySerdes();
   }
 
+  @Test
+  public void testMetadata() {
+    Topology topology =
+        parser.deserialise(TestUtils.getResourceFile("/descriptor-with-metadata.yaml"));
+    Project project = topology.getProjects().get(0);
+
+    assertThat(project.getConsumers().get(0).getMetadata()).containsKey("system");
+    assertThat(project.getProducers().get(0).getMetadata()).containsKey("contactInfo");
+    assertThat(project.getStreams().get(0).getMetadata()).containsKey("system");
+    assertThat(project.getConnectors().get(0).getMetadata()).containsKey("system");
+    assertThat(project.getTopics().get(0).getMetadata()).containsKey("domain");
+    assertThat(project.getTopics().get(1).getMetadata()).containsKey("domain");
+    assertThat(project.getTopics().get(1).getMetadata()).containsKey("owner");
+    assertThat(project.getTopics().get(0).getConsumers().get(0).getMetadata())
+        .containsKey("system");
+  }
+
   @Test
   public void testDynamicFirstLevelAttributes() {
     Topology topology =
diff --git a/src/test/resources/descriptor-with-metadata.yaml b/src/test/resources/descriptor-with-metadata.yaml
new file mode 100644
index 0000000..4c6c320
--- /dev/null
+++ b/src/test/resources/descriptor-with-metadata.yaml
@@ -0,0 +1,50 @@
+---
+context: "contextOrg"
+source: "source"
+projects:
+  - name: "foo with metadata"
+    consumers:
+      - principal: "User:App0"
+        metadata:
+          system: "OwnerSystem0"
+    producers:
+      - principal: "User:App1"
+        metadata:
+          contactInfo: "app1@company.com"
+    streams:
+      - principal: "User:StreamsApp1"
+        metadata:
+          system: "OwnerSystemS1"
+        topics:
+          read:
+            - "topicA"
+          write:
+            - "topicB"
+    connectors:
+      - principal: "User:Connect1"
+        metadata:
+          system: "OwnerSystemC1"
+        status_topic: "status"
+        offset_topic: "offset"
+        configs_topic: "configs"
+        topics:
+          read:
+            - "topicB"
+    topics:
+      - name: "topicA"
+        metadata:
+          domain: "SomeInformationModelDomain"
+        consumers:
+          - principal: "User:App4"
+            metadata:
+              system: "OwnerSystem4"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: "topicB"
+        metadata:
+          domain: "SomeInformationModelDomain"
+          owner: "DataOwnerDepartment"
+        dataType: "avro"
+        schemas:
+          value.schema.file: "schemas/bar-value.avsc"