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

Issue:
Support custom plans as part of the topic configuration
As a user of Kafka I should not be familiar for sure with the purpose of each configuration option. In a similar way as what the OpenServiceBrokers does we would like to have support for plans.
This configurations option would abstract a default configuration for topics in labels, users should be able to select a plan and if desired customize it using the config section.

With this addition, topics and acls will be fully abstracted in KTB!

Pull request:
Add custom plans to help standardise topic config
closes #103 

Patch:
diff --git a/docs/futures/use-of-custom-config-plans.rst b/docs/futures/use-of-custom-config-plans.rst
new file mode 100644
index 0000000..9475698
--- /dev/null
+++ b/docs/futures/use-of-custom-config-plans.rst
@@ -0,0 +1,86 @@
+Use of custom configuration plans
+*******************************
+
+It is possible with Kafka Topology Builder to utilise custom configuration plans.
+These plans will allow you to summarize a set of default configuration properties in a reusable label, making easier for users and as well operators to configure each topic.
+
+You might wonder, why are Plans an interesting thing, this are a few ideas where you can use it:
+
+* To define custom service levels, where you have topics with different retentions or that accept different message sizes.
+* To group default configuration, allowing you as Kafka Topology user and operator to have an smaller file size.
+
+And I am sure there is going to be more.
+
+How can you take advantage of Plans?
+
+Defining the your plans
+-----------
+
+As an operator of the Kafka Topology Builder you can define a set of custom plans, this would be a file that look like this:
+
+.. code-block:: YAML
+  ---
+  plans:
+    gold:
+      alias: "gold"
+      config:
+        retention.ms: "5000"
+        max.message.bytes: "7340116"
+    silver:
+      alias: "silver"
+      config:
+        retention.ms: "6000"
+        max.message.bytes: "524294"
+
+In this file, the reader can see two plans, *gold* and *silver* with different default configuration values.
+
+
+Using plans in the Topology
+-----------
+
+Once the plan is defined, you can use it in your Topology files. This process would look like this:
+
+.. code-block:: YAML
+  ---
+  context: "contextOrg"
+  source: "source"
+  projects:
+    - name: "foo"
+      topics:
+        - name: "foo"
+          config:
+            replication.factor: "1"
+            num.partitions: "1"
+        - name: "fooBar"
+          plan: "silver"
+          config:
+            replication.factor: "1"
+        - name: "barFoo"
+          plan: "gold"
+          config:
+            replication.factor: "1"
+        - name: "barFooBar"
+          plan: "gold"
+          config:
+            replication.factor: "1"
+
+In this Topology the topics *fooBar*, *barFoo* and *barFooBar* will be using custom plans.
+
+**Things to consider**:
+
+* Configuration values defined in the plan take precedence over the config values defined in the topic.
+* A topic can have only a plan, without providing any configuration.
+* The Plan label is optional, topics can use, or not, plans.
+* If plans are defined in the Topology, but no proper definition file is passed, the tool will complain.
+
+What about from the CLI
+-----------
+
+As a user of the Kafka Topology Builder CLI, if interested to use Plans you can pass a file using the dedicated parameter.
+An example call will look like:
+
+.. code-block:: BASH
+  $>  kafka-topology-builder.sh --brokers localhost:9092  \
+                --clientConfig example/topology-builder.properties \
+                --topology example/descriptor-with-plans.yaml  \
+                --allowDelete
\ No newline at end of file
diff --git a/example/descriptor-with-plans.yaml b/example/descriptor-with-plans.yaml
new file mode 100644
index 0000000..5cb7406
--- /dev/null
+++ b/example/descriptor-with-plans.yaml
@@ -0,0 +1,29 @@
+---
+context: "contextOrg"
+source: "source"
+projects:
+  - name: "foo"
+    topics:
+      - name: "foo"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: "fooBar"
+        plan: "silver"
+        config:
+          replication.factor: "1"
+      - name: "barFoo"
+        plan: "gold"
+        config:
+          replication.factor: "1"
+      - name: "barFooBar"
+        plan: "gold"
+        config:
+          replication.factor: "1"
+  - name: "bar"
+    topics:
+      - dataType: "avro"
+        name: "bar"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
diff --git a/example/plans.yaml b/example/plans.yaml
new file mode 100644
index 0000000..333aaaf
--- /dev/null
+++ b/example/plans.yaml
@@ -0,0 +1,12 @@
+---
+plans:
+  gold:
+    alias: "gold"
+    config:
+      retention.ms: "5000"
+      max.message.bytes: "7340116"
+  silver:
+    alias: "silver"
+    config:
+      retention.ms: "6000"
+      max.message.bytes: "524294"
diff --git a/src/main/java/com/purbon/kafka/topology/BuilderCLI.java b/src/main/java/com/purbon/kafka/topology/BuilderCLI.java
index ff88c88..7d28fa8 100644
--- a/src/main/java/com/purbon/kafka/topology/BuilderCLI.java
+++ b/src/main/java/com/purbon/kafka/topology/BuilderCLI.java
@@ -13,6 +13,9 @@ public class BuilderCLI {
   public static final String TOPOLOGY_OPTION = "topology";
   public static final String TOPOLOGY_DESC = "Topology config file.";
 
+  public static final String PLANS_OPTION = "plans";
+  public static final String PLANS_DESC = "File describing the predefined plans";
+
   public static final String BROKERS_OPTION = "brokers";
   public static final String BROKERS_DESC = "The Apache Kafka server(s) to connect to.";
 
@@ -52,6 +55,9 @@ public class BuilderCLI {
     final Option topologyFileOption =
         Option.builder().longOpt(TOPOLOGY_OPTION).hasArg().desc(TOPOLOGY_DESC).required().build();
 
+    final Option plansFileOption =
+        Option.builder().longOpt(PLANS_OPTION).hasArg().desc(PLANS_DESC).required(false).build();
+
     final Option brokersListOption =
         Option.builder()
             .longOpt(BROKERS_OPTION)
@@ -106,6 +112,7 @@ public class BuilderCLI {
     final Options options = new Options();
 
     options.addOption(topologyFileOption);
+    options.addOption(plansFileOption);
     options.addOption(brokersListOption);
     options.addOption(adminClientConfigFileOption);
 
@@ -130,7 +137,8 @@ public class BuilderCLI {
 
     Map<String, String> config = parseConfig(cmd);
 
-    processTopology(cmd.getOptionValue(TOPOLOGY_OPTION), config);
+    processTopology(
+        cmd.getOptionValue(TOPOLOGY_OPTION), cmd.getOptionValue(PLANS_OPTION, "default"), config);
     System.out.println("Kafka Topology updated");
   }
 
@@ -171,8 +179,10 @@ public class BuilderCLI {
     return cmd;
   }
 
-  void processTopology(String topologyFile, Map<String, String> config) throws Exception {
-    try (KafkaTopologyBuilder builder = KafkaTopologyBuilder.build(topologyFile, config)) {
+  void processTopology(String topologyFile, String plansFile, Map<String, String> config)
+      throws Exception {
+    try (KafkaTopologyBuilder builder =
+        KafkaTopologyBuilder.build(topologyFile, plansFile, config)) {
       builder.run();
     }
   }
diff --git a/src/main/java/com/purbon/kafka/topology/KafkaTopologyBuilder.java b/src/main/java/com/purbon/kafka/topology/KafkaTopologyBuilder.java
index f1491e0..c889775 100644
--- a/src/main/java/com/purbon/kafka/topology/KafkaTopologyBuilder.java
+++ b/src/main/java/com/purbon/kafka/topology/KafkaTopologyBuilder.java
@@ -51,6 +51,11 @@ public class KafkaTopologyBuilder implements AutoCloseable {
 
   public static KafkaTopologyBuilder build(String topologyFile, Map<String, String> config)
       throws Exception {
+    return build(topologyFile, "default", config);
+  }
+
+  public static KafkaTopologyBuilder build(
+      String topologyFile, String plansFile, Map<String, String> config) throws Exception {
 
     TopologyBuilderConfig builderConfig = TopologyBuilderConfig.build(config);
     TopologyBuilderAdminClient adminClient =
@@ -60,8 +65,10 @@ public class KafkaTopologyBuilder implements AutoCloseable {
             builderConfig, adminClient, new MDSApiClientBuilder(builderConfig));
 
     KafkaTopologyBuilder builder =
-        build(topologyFile, builderConfig, adminClient, factory.get(), factory.builder());
+        build(
+            topologyFile, plansFile, builderConfig, adminClient, factory.get(), factory.builder());
     builder.verifyRequiredParameters(topologyFile, config);
+
     return builder;
   }
 
@@ -72,8 +79,30 @@ public class KafkaTopologyBuilder implements AutoCloseable {
       AccessControlProvider accessControlProvider,
       BindingsBuilderProvider bindingsBuilderProvider)
       throws Exception {
+    return build(
+        topologyFileOrDir,
+        "default",
+        config,
+        adminClient,
+        accessControlProvider,
+        bindingsBuilderProvider);
+  }
+
+  public static KafkaTopologyBuilder build(
+      String topologyFileOrDir,
+      String plansFile,
+      TopologyBuilderConfig config,
+      TopologyBuilderAdminClient adminClient,
+      AccessControlProvider accessControlProvider,
+      BindingsBuilderProvider bindingsBuilderProvider)
+      throws Exception {
 
-    Topology topology = TopologyDescriptorBuilder.build(topologyFileOrDir, config);
+    Topology topology;
+    if (plansFile.equals("default")) {
+      topology = TopologyObjectBuilder.build(topologyFileOrDir, config);
+    } else {
+      topology = TopologyObjectBuilder.build(topologyFileOrDir, plansFile, config);
+    }
 
     TopologyValidator validator = new TopologyValidator(config);
     List<String> validationResults = validator.validate(topology);
diff --git a/src/main/java/com/purbon/kafka/topology/TopologyDescriptorBuilder.java b/src/main/java/com/purbon/kafka/topology/TopologyObjectBuilder.java
similarity index 66%
rename from src/main/java/com/purbon/kafka/topology/TopologyDescriptorBuilder.java
rename to src/main/java/com/purbon/kafka/topology/TopologyObjectBuilder.java
index 6ca812b..a531a58 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyDescriptorBuilder.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyObjectBuilder.java
@@ -1,7 +1,9 @@
 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;
 import com.purbon.kafka.topology.serdes.TopologySerdes;
 import java.io.File;
 import java.io.IOException;
@@ -12,16 +14,26 @@ import java.util.List;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
-public class TopologyDescriptorBuilder {
+public class TopologyObjectBuilder {
 
-  private static final Logger LOGGER = LogManager.getLogger(TopologyDescriptorBuilder.class);
+  private static final Logger LOGGER = LogManager.getLogger(TopologyObjectBuilder.class);
 
   public static Topology build(String fileOrDir) throws IOException {
-    return build(fileOrDir, new TopologyBuilderConfig());
+    return build(fileOrDir, "", new TopologyBuilderConfig());
+  }
+
+  public static Topology build(String fileOrDir, String plansFile) throws IOException {
+    return build(fileOrDir, plansFile, new TopologyBuilderConfig());
   }
 
   public static Topology build(String fileOrDir, TopologyBuilderConfig config) throws IOException {
-    List<Topology> topologies = parseListOfTopologies(fileOrDir, config);
+    return build(fileOrDir, "", config);
+  }
+
+  public static Topology build(String fileOrDir, String plansFile, TopologyBuilderConfig config)
+      throws IOException {
+    PlanMap plans = buildPlans(plansFile);
+    List<Topology> topologies = parseListOfTopologies(fileOrDir, config, plans);
     Topology topology = topologies.get(0);
     if (topologies.size() > 1) {
       List<Topology> subTopologies = topologies.subList(1, topologies.size());
@@ -35,9 +47,14 @@ public class TopologyDescriptorBuilder {
     return topology;
   }
 
+  private static PlanMap buildPlans(String plansFile) throws IOException {
+    PlanMapSerdes plansSerdes = new PlanMapSerdes();
+    return plansFile.isEmpty() ? new PlanMap() : plansSerdes.deserialise(new File(plansFile));
+  }
+
   private static List<Topology> parseListOfTopologies(
-      String fileOrDir, TopologyBuilderConfig config) throws IOException {
-    TopologySerdes parser = new TopologySerdes(config);
+      String fileOrDir, TopologyBuilderConfig config, PlanMap plans) throws IOException {
+    TopologySerdes parser = new TopologySerdes(config, plans);
     List<Topology> topologies = new ArrayList<>();
     boolean isDir = Files.isDirectory(Paths.get(fileOrDir));
     if (isDir) {
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 22ecb8f..774e582 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
@@ -31,6 +31,9 @@ public class TopicImpl implements Topic, Cloneable {
   private List<Producer> producers;
   private List<Consumer> consumers;
 
+  @JsonInclude(Include.NON_EMPTY)
+  private String plan;
+
   private Map<String, String> config;
   @JsonIgnore private TopologyBuilderConfig appConfig;
   @JsonIgnore private Map<String, Object> context;
@@ -104,6 +107,11 @@ public class TopicImpl implements Topic, Cloneable {
     return schemas;
   }
 
+  @Override
+  public String getPlan() {
+    return plan;
+  }
+
   public void setSchemas(TopicSchemas schemas) {
     this.schemas = schemas;
   }
diff --git a/src/main/java/com/purbon/kafka/topology/model/Plan.java b/src/main/java/com/purbon/kafka/topology/model/Plan.java
new file mode 100644
index 0000000..98e60c9
--- /dev/null
+++ b/src/main/java/com/purbon/kafka/topology/model/Plan.java
@@ -0,0 +1,31 @@
+package com.purbon.kafka.topology.model;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
+import java.util.HashMap;
+import java.util.Map;
+
+public class Plan {
+
+  @JsonInclude(Include.NON_EMPTY)
+  private String alias;
+
+  private Map<String, String> config;
+
+  public Plan() {
+    this("default", new HashMap<>());
+  }
+
+  public Plan(String alias, Map<String, String> config) {
+    this.alias = alias;
+    this.config = config;
+  }
+
+  public String getAlias() {
+    return alias;
+  }
+
+  public Map<String, String> getConfig() {
+    return config;
+  }
+}
diff --git a/src/main/java/com/purbon/kafka/topology/model/PlanMap.java b/src/main/java/com/purbon/kafka/topology/model/PlanMap.java
new file mode 100644
index 0000000..c42c1f1
--- /dev/null
+++ b/src/main/java/com/purbon/kafka/topology/model/PlanMap.java
@@ -0,0 +1,29 @@
+package com.purbon.kafka.topology.model;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class PlanMap {
+
+  Map<String, Plan> plans;
+
+  public PlanMap() {
+    this.plans = new HashMap<>();
+  }
+
+  public PlanMap(Map<String, Plan> plans) {
+    this.plans = plans;
+  }
+
+  public boolean containsKey(String key) {
+    return plans.containsKey(key);
+  }
+
+  public Plan get(String key) {
+    return plans.get(key);
+  }
+
+  public int size() {
+    return plans.size();
+  }
+}
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 1833944..fe2e25f 100644
--- a/src/main/java/com/purbon/kafka/topology/model/Topic.java
+++ b/src/main/java/com/purbon/kafka/topology/model/Topic.java
@@ -17,6 +17,8 @@ public interface Topic {
 
   TopicSchemas getSchemas();
 
+  String getPlan();
+
   void setSchemas(TopicSchemas schemas);
 
   Map<String, String> getConfig();
diff --git a/src/main/java/com/purbon/kafka/topology/serdes/PlanMapSerdes.java b/src/main/java/com/purbon/kafka/topology/serdes/PlanMapSerdes.java
new file mode 100644
index 0000000..12e9716
--- /dev/null
+++ b/src/main/java/com/purbon/kafka/topology/serdes/PlanMapSerdes.java
@@ -0,0 +1,37 @@
+package com.purbon.kafka.topology.serdes;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+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.model.PlanMap;
+import java.io.File;
+import java.io.IOException;
+
+public class PlanMapSerdes {
+
+  ObjectMapper mapper;
+
+  public PlanMapSerdes() {
+    mapper = new ObjectMapper(new YAMLFactory());
+    mapper.setVisibility(PropertyAccessor.FIELD, Visibility.ANY);
+    mapper.registerModule(new SimpleModule());
+    mapper.registerModule(new Jdk8Module());
+    mapper.findAndRegisterModules();
+  }
+
+  public PlanMap deserialise(File file) throws IOException {
+    return mapper.readValue(file, PlanMap.class);
+  }
+
+  public PlanMap deserialise(String content) throws IOException {
+    return mapper.readValue(content, PlanMap.class);
+  }
+
+  public String serialise(PlanMap planMap) throws JsonProcessingException {
+    return mapper.writeValueAsString(planMap);
+  }
+}
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 1ab210b..2eed45f 100644
--- a/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
+++ b/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
@@ -6,9 +6,9 @@ import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.databind.DeserializationContext;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
-import com.google.common.collect.Maps;
 import com.purbon.kafka.topology.TopologyBuilderConfig;
 import com.purbon.kafka.topology.model.Impl.TopicImpl;
+import com.purbon.kafka.topology.model.PlanMap;
 import com.purbon.kafka.topology.model.TopicSchemas;
 import com.purbon.kafka.topology.model.User;
 import com.purbon.kafka.topology.model.users.Consumer;
@@ -16,10 +16,10 @@ import com.purbon.kafka.topology.model.users.Producer;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import jersey.repackaged.com.google.common.collect.Sets;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -28,14 +28,16 @@ public class TopicCustomDeserializer extends StdDeserializer<TopicImpl> {
   private static final Logger LOGGER = LogManager.getLogger(TopicCustomDeserializer.class);
 
   private final TopologyBuilderConfig config;
+  private PlanMap plans;
 
-  TopicCustomDeserializer(TopologyBuilderConfig config) {
-    this(null, config);
+  TopicCustomDeserializer(TopologyBuilderConfig config, PlanMap plans) {
+    this(null, config, plans);
   }
 
-  private TopicCustomDeserializer(Class<?> clazz, TopologyBuilderConfig config) {
+  private TopicCustomDeserializer(Class<?> clazz, TopologyBuilderConfig config, PlanMap plans) {
     super(clazz);
     this.config = config;
+    this.plans = plans;
   }
 
   @Override
@@ -49,16 +51,38 @@ public class TopicCustomDeserializer extends StdDeserializer<TopicImpl> {
     List<Producer> producers = getUsers(parser, rootNode, "producers", Producer.class);
 
     Optional<JsonNode> optionalDataTypeNode = Optional.ofNullable(rootNode.get("dataType"));
-    Optional<String> optionalDataType = optionalDataTypeNode.map(o -> o.asText());
+    Optional<String> optionalDataType = optionalDataTypeNode.map(JsonNode::asText);
 
     Optional<JsonNode> optionalConfigNode = Optional.ofNullable(rootNode.get("config"));
     Map<String, String> config =
         optionalConfigNode
             .map(
-                node ->
-                    Maps.asMap(Sets.newHashSet(node.fieldNames()), (key) -> node.get(key).asText()))
+                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<>());
 
+    Optional<JsonNode> optionalPlanLabel = Optional.ofNullable(rootNode.get("plan"));
+    if (optionalPlanLabel.isPresent() && plans.size() == 0) {
+      throw new IOException("A plan definition is required if the topology uses them");
+    }
+    optionalPlanLabel.ifPresent(
+        jsonNode -> {
+          String planLabel = jsonNode.asText();
+          if (plans.containsKey(planLabel)) {
+            Map<String, String> planConfigObject = plans.get(planLabel).getConfig();
+            planConfigObject.forEach(config::put);
+          } else {
+            LOGGER.warn(planLabel + " is missing in the plans definition. It will be ignored.");
+          }
+        });
+
     TopicImpl topic =
         new TopicImpl(name, producers, consumers, optionalDataType, config, this.config);
 
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 9542a14..6ee51c7 100644
--- a/src/main/java/com/purbon/kafka/topology/serdes/TopologySerdes.java
+++ b/src/main/java/com/purbon/kafka/topology/serdes/TopologySerdes.java
@@ -7,6 +7,7 @@ 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.model.Impl.TopicImpl;
+import com.purbon.kafka.topology.model.PlanMap;
 import com.purbon.kafka.topology.model.Topology;
 import java.io.File;
 import java.io.IOException;
@@ -21,15 +22,23 @@ public class TopologySerdes {
   }
 
   public TopologySerdes() {
-    this(new TopologyBuilderConfig(), FileType.YAML);
+    this(new TopologyBuilderConfig(), FileType.YAML, new PlanMap());
   }
 
   public TopologySerdes(TopologyBuilderConfig config) {
-    this(config, config.getTopologyFileType());
+    this(config, config.getTopologyFileType(), new PlanMap());
+  }
+
+  public TopologySerdes(TopologyBuilderConfig config, PlanMap plans) {
+    this(config, config.getTopologyFileType(), plans);
   }
 
   public TopologySerdes(TopologyBuilderConfig config, FileType type) {
-    mapper = ObjectMapperFactory.build(type, config);
+    mapper = ObjectMapperFactory.build(type, config, new PlanMap());
+  }
+
+  public TopologySerdes(TopologyBuilderConfig config, FileType type, PlanMap plans) {
+    mapper = ObjectMapperFactory.build(type, config, plans);
   }
 
   public Topology deserialise(File file) throws IOException {
@@ -46,7 +55,7 @@ public class TopologySerdes {
 
   private static class ObjectMapperFactory {
 
-    public static ObjectMapper build(FileType type, TopologyBuilderConfig config) {
+    public static ObjectMapper build(FileType type, TopologyBuilderConfig config, PlanMap plans) {
       ObjectMapper mapper;
       if (type.equals(FileType.JSON)) {
         mapper = new ObjectMapper();
@@ -56,7 +65,7 @@ public class TopologySerdes {
 
       SimpleModule module = new SimpleModule();
       module.addDeserializer(Topology.class, new TopologyCustomDeserializer(config));
-      module.addDeserializer(TopicImpl.class, new TopicCustomDeserializer(config));
+      module.addDeserializer(TopicImpl.class, new TopicCustomDeserializer(config, plans));
       mapper.registerModule(module);
       mapper.registerModule(new Jdk8Module());
       mapper.findAndRegisterModules();


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/CLITest.java b/src/test/java/com/purbon/kafka/topology/CLITest.java
index a86b099..f9ed20e 100644
--- a/src/test/java/com/purbon/kafka/topology/CLITest.java
+++ b/src/test/java/com/purbon/kafka/topology/CLITest.java
@@ -40,7 +40,7 @@ public class CLITest {
           "--clientConfig", "topology-builder-sasl-plain.properties"
         };
 
-    doNothing().when(cli).processTopology(eq("descriptor.yaml"), anyMap());
+    doNothing().when(cli).processTopology(eq("descriptor.yaml"), eq("default"), anyMap());
 
     Map<String, String> config = new HashMap<>();
     config.put(BROKERS_OPTION, "localhost:9092");
@@ -50,7 +50,7 @@ public class CLITest {
     config.put(ADMIN_CLIENT_CONFIG_OPTION, "topology-builder-sasl-plain.properties");
     cli.run(args);
 
-    verify(cli, times(1)).processTopology(eq("descriptor.yaml"), eq(config));
+    verify(cli, times(1)).processTopology(eq("descriptor.yaml"), eq("default"), eq(config));
   }
 
   @Test
@@ -63,7 +63,7 @@ public class CLITest {
           "--dryRun"
         };
 
-    doNothing().when(cli).processTopology(eq("descriptor.yaml"), anyMap());
+    doNothing().when(cli).processTopology(eq("descriptor.yaml"), eq("default"), anyMap());
 
     Map<String, String> config = new HashMap<>();
     config.put(BROKERS_OPTION, "localhost:9092");
@@ -73,6 +73,6 @@ public class CLITest {
     config.put(ADMIN_CLIENT_CONFIG_OPTION, "topology-builder-sasl-plain.properties");
     cli.run(args);
 
-    verify(cli, times(1)).processTopology(eq("descriptor.yaml"), eq(config));
+    verify(cli, times(1)).processTopology(eq("descriptor.yaml"), eq("default"), eq(config));
   }
 }
diff --git a/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java b/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java
index 643d45b..781ca37 100644
--- a/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java
+++ b/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java
@@ -13,7 +13,6 @@ import java.io.PrintStream;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
 import org.apache.kafka.common.resource.ResourceType;
@@ -29,15 +28,11 @@ public class ExecutionPlanTest {
   ExecutionPlan plan;
   BackendController backendController;
 
-  @Mock
-  PrintStream mockPrintStream;
+  @Mock PrintStream mockPrintStream;
 
-  @Mock
-  SimpleAclsProvider aclsProvider;
-
-  @Rule
-  public MockitoRule mockitoRule = MockitoJUnit.rule();
+  @Mock SimpleAclsProvider aclsProvider;
 
+  @Rule public MockitoRule mockitoRule = MockitoJUnit.rule();
 
   @Before
   public void before() throws IOException {
@@ -48,8 +43,10 @@ public class ExecutionPlanTest {
 
   @Test
   public void addBindingsTest() throws IOException {
-    TopologyAclBinding binding1 = new TopologyAclBinding(ResourceType.ANY, "topicA", "*", "ALL", "User:foo", "LITERAL");
-    TopologyAclBinding binding2 = new TopologyAclBinding(ResourceType.ANY, "topicB", "*", "ALL", "User:foo", "LITERAL");
+    TopologyAclBinding binding1 =
+        new TopologyAclBinding(ResourceType.ANY, "topicA", "*", "ALL", "User:foo", "LITERAL");
+    TopologyAclBinding binding2 =
+        new TopologyAclBinding(ResourceType.ANY, "topicB", "*", "ALL", "User:foo", "LITERAL");
     Set<TopologyAclBinding> bindings = new HashSet<>(Arrays.asList(binding1, binding2));
     CreateBindings addBindingsAction = new CreateBindings(aclsProvider, bindings);
 
@@ -63,8 +60,10 @@ public class ExecutionPlanTest {
 
   @Test
   public void deleteBindingsAfterCreateTest() throws IOException {
-    TopologyAclBinding binding1 = new TopologyAclBinding(ResourceType.ANY, "topicA", "*", "ALL", "User:foo", "LITERAL");
-    TopologyAclBinding binding2 = new TopologyAclBinding(ResourceType.ANY, "topicB", "*", "ALL", "User:foo", "LITERAL");
+    TopologyAclBinding binding1 =
+        new TopologyAclBinding(ResourceType.ANY, "topicA", "*", "ALL", "User:foo", "LITERAL");
+    TopologyAclBinding binding2 =
+        new TopologyAclBinding(ResourceType.ANY, "topicB", "*", "ALL", "User:foo", "LITERAL");
     Set<TopologyAclBinding> bindings = new HashSet<>(Arrays.asList(binding1, binding2));
     CreateBindings addBindingsAction = new CreateBindings(aclsProvider, bindings);
 
@@ -91,7 +90,5 @@ public class ExecutionPlanTest {
     backendController = new BackendController();
     backendController.load();
     assertEquals(1, backendController.size());
-
   }
-
 }
diff --git a/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java b/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
index a522c11..63e9012 100644
--- a/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
@@ -6,7 +6,6 @@ import static com.purbon.kafka.topology.BuilderCLI.BROKERS_OPTION;
 import static com.purbon.kafka.topology.BuilderCLI.DRY_RUN_OPTION;
 import static com.purbon.kafka.topology.BuilderCLI.QUIET_OPTION;
 import static com.purbon.kafka.topology.TopologyBuilderConfig.CONFLUENT_SCHEMA_REGISTRY_URL_CONFIG;
-import static org.junit.Assert.assertEquals;
 import static org.mockito.Matchers.anyObject;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.times;
@@ -15,9 +14,7 @@ import static org.mockito.Mockito.verify;
 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.model.Topology;
 import java.io.IOException;
-import java.net.URISyntaxException;
 import java.net.URL;
 import java.nio.file.Paths;
 import java.util.HashMap;
@@ -60,17 +57,6 @@ public class KafkaTopologyBuilderTest {
     props.put(AdminClientConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
   }
 
-  @Test
-  public void buildTopicNameTest() throws URISyntaxException, IOException {
-
-    URL dirOfDescriptors = getClass().getResource("/dir");
-    String fileOrDirPath = Paths.get(dirOfDescriptors.toURI()).toFile().toString();
-
-    Topology topology = TopologyDescriptorBuilder.build(fileOrDirPath);
-
-    assertEquals(4, topology.getProjects().size());
-  }
-
   @Test
   public void closeAdminClientTest() throws Exception {
     URL dirOfDescriptors = getClass().getResource("/descriptor.yaml");
diff --git a/src/test/java/com/purbon/kafka/topology/PlanMapSerdesTest.java b/src/test/java/com/purbon/kafka/topology/PlanMapSerdesTest.java
new file mode 100644
index 0000000..2ad2629
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/PlanMapSerdesTest.java
@@ -0,0 +1,67 @@
+package com.purbon.kafka.topology;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.purbon.kafka.topology.model.Plan;
+import com.purbon.kafka.topology.model.PlanMap;
+import com.purbon.kafka.topology.serdes.PlanMapSerdes;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Map;
+import org.assertj.core.api.Condition;
+import org.junit.Before;
+import org.junit.Test;
+
+public class PlanMapSerdesTest {
+
+  PlanMapSerdes parser;
+
+  @Before
+  public void before() {
+    parser = new PlanMapSerdes();
+  }
+
+  @Test
+  public void testSerialization() throws JsonProcessingException {
+
+    Map<String, String> config = new HashMap<>();
+    config.put("foo", "bar");
+    config.put("bar", "3");
+
+    Plan plan1 = new Plan("gold", config);
+    Plan plan2 = new Plan("silver", config);
+
+    Map<String, Plan> plans = new HashMap<>();
+    plans.put("gold", plan1);
+    plans.put("silver", plan2);
+
+    PlanMap map = new PlanMap(plans);
+
+    String serValue = new PlanMapSerdes().serialise(map);
+
+    assertThat(serValue).contains("alias: \"gold\"");
+    assertThat(serValue).contains("alias: \"silver\"");
+    assertThat(serValue).contains("config:");
+  }
+
+  @Test
+  public void testHappyDeserialization() throws URISyntaxException, IOException {
+    URL topologyDescriptor = getClass().getResource("/plans.yaml");
+    PlanMap plans = parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
+
+    assertThat(plans).has(new Condition<>(list -> list.size() == 2, "Contains two elements"));
+    assertThat(plans.get("gold").getAlias()).isEqualTo("gold");
+
+    Map<String, String> config = new HashMap<>();
+    config.put("foo", "bar");
+    config.put("bar", "3");
+
+    assertThat(plans.get("gold").getConfig()).isEqualTo(config);
+
+    assertThat(plans.get("silver").getAlias()).isEqualTo("silver");
+  }
+}
diff --git a/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java b/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java
new file mode 100644
index 0000000..0d43605
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java
@@ -0,0 +1,69 @@
+package com.purbon.kafka.topology;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertEquals;
+
+import com.purbon.kafka.topology.model.Topic;
+import com.purbon.kafka.topology.model.Topology;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.junit.Test;
+
+public class TopologyObjectBuilderTest {
+
+  @Test
+  public void buildTopicNameTest() throws URISyntaxException, IOException {
+
+    URL dirOfDescriptors = getClass().getResource("/dir");
+    String fileOrDirPath = Paths.get(dirOfDescriptors.toURI()).toFile().toString();
+
+    Topology topology = TopologyObjectBuilder.build(fileOrDirPath);
+
+    assertEquals(4, topology.getProjects().size());
+  }
+
+  @Test
+  public void testConfigUpdateWhenUsingCustomPlans() throws URISyntaxException, IOException {
+    URL dirOfDescriptors = getClass().getResource("/descriptor-with-plans.yaml");
+    String descriptorFile = Paths.get(dirOfDescriptors.toURI()).toFile().toString();
+
+    URL plansFileURL = getClass().getResource("/plans.yaml");
+    String plansFile = Paths.get(plansFileURL.toURI()).toFile().toString();
+
+    Topology topology = TopologyObjectBuilder.build(descriptorFile, plansFile);
+    assertThat(topology).isNotNull();
+
+    List<Topic> topics = topology.getProjects().get(0).getTopics();
+    Map<String, Topic> map =
+        topics.stream().collect(Collectors.toMap(Topic::getName, topic -> topic));
+
+    // should include the config values from the plan into the topic config.
+    Topic topic = map.get("fooBar");
+    Map<String, String> config = new HashMap<>();
+    config.put("foo", "bar");
+    config.put("bar", "3");
+    assertThat(topic.getConfig()).containsAllEntriesOf(config);
+
+    // should respect values from the original config if not present in the plan description
+    topic = map.get("barFoo");
+    assertThat(topic.getConfig()).containsEntry("replication.factor", "1");
+
+    // should overwide values with priority given to the plans
+    topic = map.get("barFooBar");
+    assertThat(topic.getConfig()).containsEntry("replication.factor", "1");
+    assertThat(topic.getConfig()).containsEntry("bar", "3");
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testTopologyWithPlansButWithNoPlansDef() throws URISyntaxException, IOException {
+    URL dirOfDescriptors = getClass().getResource("/descriptor-with-plans.yaml");
+    String descriptorFile = Paths.get(dirOfDescriptors.toURI()).toFile().toString();
+    TopologyObjectBuilder.build(descriptorFile);
+  }
+}
diff --git a/src/test/resources/descriptor-with-plans.yaml b/src/test/resources/descriptor-with-plans.yaml
new file mode 100644
index 0000000..1bb6dde
--- /dev/null
+++ b/src/test/resources/descriptor-with-plans.yaml
@@ -0,0 +1,80 @@
+---
+context: "contextOrg"
+source: "source"
+projects:
+  - name: "foo"
+    consumers:
+      - principal: "User:App0"
+        group: "foo"
+      - principal: "User:App1"
+    producers:
+      - principal: "User:App0"
+      - principal: "User:App2"
+        transactionId: "1234"
+      - principal: "User:App2"
+        idempotence: "true"
+    streams:
+      - principal: "User:App0"
+        topics:
+          read:
+            - "topicA"
+            - "topicB"
+          write:
+            - "topicC"
+            - "topicD"
+    connectors:
+      - principal: "User:Connect1"
+        cluster_id: "foo"
+        group: "group"
+        status_topic: "status"
+        offset_topic: "offset"
+        configs_topic: "configs"
+        topics:
+          read:
+            - "topicA"
+            - "topicB"
+      - principal: "User:Connect2"
+        topics:
+          write:
+            - "topicC"
+            - "topicD"
+    topics:
+      - name: "foo"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: "fooBar"
+        plan: "gold"
+      - name: "barFoo"
+        plan: "gold"
+        config:
+          replication.factor: "1"
+      - name: "barFooBar"
+        plan: "gold"
+        config:
+          bar: "1"
+          replication.factor: "1"
+      - name: "bar"
+        dataType: "avro"
+        schemas:
+          key.schema.file: "schemas/bar-key.avsc"
+          value.schema.file: "schemas/bar-value.avsc"
+  - name: "bar"
+    topics:
+      - dataType: "avro"
+        name: "bar"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+platform:
+  schema_registry:
+    instances:
+      - principal: "User:SchemaRegistry01"
+        topic: "foo"
+        group: "bar"
+      - principal: "User:SchemaRegistry02"
+        topic: "zet"
+  control_center:
+    instances:
+      - principal: "User:ControlCenter"
+        appId: "controlcenter"
\ No newline at end of file
diff --git a/src/test/resources/plans.yaml b/src/test/resources/plans.yaml
new file mode 100644
index 0000000..0ecd1c2
--- /dev/null
+++ b/src/test/resources/plans.yaml
@@ -0,0 +1,12 @@
+---
+plans:
+  gold:
+    alias: "gold"
+    config:
+      bar: "3"
+      foo: "bar"
+  silver:
+    alias: "silver"
+    config:
+      bar: "3"
+      foo: "bar"
\ No newline at end of file