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

Issue:
Multiple event types in same topic in topology configuration
I would be great if we could support multiple event types in the same topic. For value TopicRecordNameStrategy: The subject name is <topic>-<type>-value, where <topic> is the Kafka topic name, and <type> is the fully-qualified name of the Avro record type of the message.

Currently the descriptor only supports single key, value schema pairs. 

This issue is to support multiple schema's required for one topic.

Pull request description:
Support multiple schema per topic
* **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.

takes over from #158 

closes #155 

Patch:
diff --git a/docs/futures/what-schema-management.rst b/docs/futures/what-schema-management.rst
index 5f5df5d..dcb6be7 100644
--- a/docs/futures/what-schema-management.rst
+++ b/docs/futures/what-schema-management.rst
@@ -17,14 +17,24 @@ An example topology for managing schemas would look like this (*only the topics
       - name: "projectA"
         topics:
           - name: "foo"
+            schemas:
+              value.schema.file: "schemas/foo-value.avsc"
             config:
               replication.factor: "1"
               num.partitions: "1"
           - name: "bar"
             dataType: "avro"
+            subject.name.strategy: "TopicRecordNameStrategy"
             schemas:
-              key.schema.file: "schemas/bar-key.avsc"
-              value.schema.file: "schemas/bar-value.avsc"
+              - key.schema.file: "schemas/foo-key.avsc"
+                value.schema.file: "schemas/foo-value.avsc"
+                value.record.type: "foo"
+              - key.schema.file: "schemas/bar-key.avsc"
+                value.schema.file: "schemas/bar-value.avsc"
+                value.record.type: "bar"
+              - key.schema.file: "schemas/zet-key.avsc"
+                value.schema.file: "schemas/zet-value.avsc"
+                value.record.type: "zet"
             config:
               replication.factor: "1"
               num.partitions: "1"
@@ -34,8 +44,17 @@ Currently the schemas are managed as files and need to be accessible for the KTB
 
 *NOTE*: The path for the files is relative to the location of the of the topology.
 
-*NOTE*: Keep in mind that for a usecase like in the example above you have to define the value for `schema.registry.url` in your properties file.
+*NOTE*: Keep in mind that for a use case like in the example above you have to define the value for `schema.registry.url` in your properties file.
+
+Multiple schemas support per topic
+-----------
+
+It is possible to setup multiple schemas per topic, this is a common use case when using a subject name strategy different than *TopicNameStrategy*.
+
+If using multiple schemas, as it can be seen in the earlier example, the user must set the subject naming strategy for the topic to be one of the
+other options.
 
+The user should as well set the *record.type* for each of the keys as this is a value required when configuring the final subject.
 
 Supported Schema Registry
 -----------
diff --git a/example/descriptor-with-schemas.yaml b/example/descriptor-with-schemas.yaml
index 785188d..14fb385 100644
--- a/example/descriptor-with-schemas.yaml
+++ b/example/descriptor-with-schemas.yaml
@@ -37,14 +37,20 @@ projects:
             - "topicD"
     topics:
       - name: "foo"
+        schemas:
+          value.schema.file: "schemas/foo-value.avsc"
         config:
           replication.factor: "1"
           num.partitions: "1"
       - name: "bar"
         dataType: "avro"
         schemas:
-          key.schema.file: "schemas/bar-key.avsc"
-          value.schema.file: "schemas/bar-value.avsc"
+          - key.schema.file: "schemas/bar-key.avsc"
+            value.schema.file: "schemas/bar-value.avsc"
+          - key.schema.file: "schemas/bar-key.avsc"
+            value.schema.file: "schemas/bar-value.avsc"
+          - key.schema.file: "schemas/bar-key.avsc"
+            value.schema.file: "schemas/bar-value.avsc"
         config:
           replication.factor: "1"
           num.partitions: "1"
diff --git a/example/schemas/foo-value.avsc b/example/schemas/foo-value.avsc
new file mode 100644
index 0000000..712e96f
--- /dev/null
+++ b/example/schemas/foo-value.avsc
@@ -0,0 +1,1 @@
+{"type": "string"}
\ No newline at end of file
diff --git a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
index 393f971..9661927 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
@@ -202,7 +202,7 @@ public class TopologyBuilderConfig {
   private static boolean hasSchemas(Topology topology) {
     return topology.getProjects().stream()
         .flatMap((Function<Project, Stream<Topic>>) project -> project.getTopics().stream())
-        .anyMatch(topic -> topic.getSchemas().isPresent());
+        .anyMatch(topic -> !topic.getSchemas().isEmpty());
   }
 
   private void raiseIfDefault(String key, String _default) throws ConfigurationException {
diff --git a/src/main/java/com/purbon/kafka/topology/actions/topics/SyncTopicAction.java b/src/main/java/com/purbon/kafka/topology/actions/topics/SyncTopicAction.java
index 746c74b..30f5f2c 100644
--- a/src/main/java/com/purbon/kafka/topology/actions/topics/SyncTopicAction.java
+++ b/src/main/java/com/purbon/kafka/topology/actions/topics/SyncTopicAction.java
@@ -4,6 +4,7 @@ import com.purbon.kafka.topology.actions.BaseAction;
 import com.purbon.kafka.topology.api.adminclient.TopologyBuilderAdminClient;
 import com.purbon.kafka.topology.model.Topic;
 import com.purbon.kafka.topology.model.TopicSchemas;
+import com.purbon.kafka.topology.model.schema.Subject;
 import com.purbon.kafka.topology.schemas.SchemaRegistryManager;
 import java.io.IOException;
 import java.util.HashMap;
@@ -58,20 +59,17 @@ public class SyncTopicAction extends BaseAction {
       adminClient.createTopic(topic, fullTopicName);
     }
 
-    if (topic.getSchemas().isPresent()) {
-      final TopicSchemas schemas = topic.getSchemas().get();
-
-      schemas
-          .getKeySchemaFile()
-          .ifPresent(
-              keySchemaFile ->
-                  schemaRegistryManager.register(fullTopicName + "-key", keySchemaFile));
-
-      schemas
-          .getValueSchemaFile()
-          .ifPresent(
-              valueSchemaFile ->
-                  schemaRegistryManager.register(fullTopicName + "-value", valueSchemaFile));
+    for (TopicSchemas schema : topic.getSchemas()) {
+      Subject keySubject = schema.getKeySubject();
+      Subject valueSubject = schema.getValueSubject();
+      if (keySubject.hasSchemaFile()) {
+        String keySchemaFile = keySubject.getSchemaFile();
+        schemaRegistryManager.register(keySubject.buildSubjectName(topic), keySchemaFile);
+      }
+      if (valueSubject.hasSchemaFile()) {
+        String valueSchemaFile = valueSubject.getSchemaFile();
+        schemaRegistryManager.register(valueSubject.buildSubjectName(topic), valueSchemaFile);
+      }
     }
   }
 
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 3547d5c..1e3ccea 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
@@ -5,6 +5,7 @@ import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonInclude.Include;
 import com.purbon.kafka.topology.TopicManager;
 import com.purbon.kafka.topology.TopologyBuilderConfig;
+import com.purbon.kafka.topology.model.SubjectNameStrategy;
 import com.purbon.kafka.topology.model.Topic;
 import com.purbon.kafka.topology.model.TopicSchemas;
 import com.purbon.kafka.topology.model.users.Consumer;
@@ -24,7 +25,7 @@ public class TopicImpl implements Topic, Cloneable {
   private Optional<String> dataType;
 
   @JsonInclude(Include.NON_EMPTY)
-  private Optional<TopicSchemas> schemas;
+  private List<TopicSchemas> schemas;
 
   private String name;
 
@@ -45,6 +46,9 @@ public class TopicImpl implements Topic, Cloneable {
   private static String DEFAULT_PARTITION_COUNT = "3";
   private static String DEFAULT_REPLICATION_FACTOR = "2";
 
+  @JsonInclude(Include.NON_EMPTY)
+  private Optional<SubjectNameStrategy> subjectNameStrategy;
+
   public TopicImpl() {
     this(DEFAULT_TOPIC_NAME, Optional.empty(), new HashMap<>(), new TopologyBuilderConfig());
   }
@@ -89,7 +93,7 @@ public class TopicImpl implements Topic, Cloneable {
     this.consumers = consumers;
     this.dataType = dataType;
     this.config = config;
-    this.schemas = Optional.empty();
+    this.schemas = new ArrayList<>();
     this.replicationFactor = 0;
     this.partitionCount = 0;
     this.appConfig = appConfig;
@@ -97,6 +101,7 @@ public class TopicImpl implements Topic, Cloneable {
     partitionCount = Integer.parseInt(value);
     value = config.getOrDefault(TopicManager.REPLICATION_FACTOR, DEFAULT_REPLICATION_FACTOR);
     replicationFactor = Short.parseShort(value);
+    subjectNameStrategy = Optional.empty();
   }
 
   public String getName() {
@@ -108,14 +113,22 @@ public class TopicImpl implements Topic, Cloneable {
     return plan;
   }
 
-  public Optional<TopicSchemas> getSchemas() {
+  public List<TopicSchemas> getSchemas() {
     return schemas;
   }
 
-  public void setSchemas(Optional<TopicSchemas> schemas) {
+  public void setSchemas(List<TopicSchemas> schemas) {
     this.schemas = schemas;
   }
 
+  public void setSubjectNameStrategy(Optional<SubjectNameStrategy> subjectNameStrategy) {
+    this.subjectNameStrategy = subjectNameStrategy;
+  }
+
+  public SubjectNameStrategy getSubjectNameStrategy() {
+    return subjectNameStrategy.orElse(SubjectNameStrategy.TOPIC_NAME_STRATEGY);
+  }
+
   private String toString(String projectPrefix) {
     switch (appConfig.getTopicPrefixFormat()) {
       case "default":
diff --git a/src/main/java/com/purbon/kafka/topology/model/SubjectNameStrategy.java b/src/main/java/com/purbon/kafka/topology/model/SubjectNameStrategy.java
new file mode 100644
index 0000000..f01541f
--- /dev/null
+++ b/src/main/java/com/purbon/kafka/topology/model/SubjectNameStrategy.java
@@ -0,0 +1,27 @@
+package com.purbon.kafka.topology.model;
+
+public enum SubjectNameStrategy {
+  TOPIC_NAME_STRATEGY("TopicNameStrategy"),
+  RECORD_NAME_STRATEGY("RecordNameStrategy"),
+  TOPIC_RECORD_NAME_STRATEGY("TopicRecordNameStrategy");
+
+  private final String label;
+
+  SubjectNameStrategy(String label) {
+    this.label = label;
+  }
+
+  @Override
+  public String toString() {
+    return label;
+  }
+
+  public static SubjectNameStrategy valueOfLabel(String label) {
+    for (SubjectNameStrategy s : values()) {
+      if (s.label.equals(label)) {
+        return s;
+      }
+    }
+    return null;
+  }
+}
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 2d71fe1..65b933c 100644
--- a/src/main/java/com/purbon/kafka/topology/model/Topic.java
+++ b/src/main/java/com/purbon/kafka/topology/model/Topic.java
@@ -15,11 +15,11 @@ public interface Topic {
 
   String getName();
 
-  Optional<TopicSchemas> getSchemas();
+  List<TopicSchemas> getSchemas();
 
   String getPlan();
 
-  void setSchemas(Optional<TopicSchemas> schemas);
+  void setSchemas(List<TopicSchemas> schemas);
 
   Map<String, String> getConfig();
 
@@ -44,4 +44,6 @@ public interface Topic {
   void setConsumers(List<Consumer> consumers);
 
   void setProducers(List<Producer> producers);
+
+  SubjectNameStrategy getSubjectNameStrategy();
 }
diff --git a/src/main/java/com/purbon/kafka/topology/model/TopicSchemas.java b/src/main/java/com/purbon/kafka/topology/model/TopicSchemas.java
index 85cdfb7..da9bebf 100644
--- a/src/main/java/com/purbon/kafka/topology/model/TopicSchemas.java
+++ b/src/main/java/com/purbon/kafka/topology/model/TopicSchemas.java
@@ -3,35 +3,35 @@ package com.purbon.kafka.topology.model;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.PropertyNamingStrategy;
 import com.fasterxml.jackson.databind.annotation.JsonNaming;
+import com.purbon.kafka.topology.model.schema.Subject;
+import com.purbon.kafka.topology.model.schema.Subject.SubjectKind;
 import java.util.Optional;
 
 @JsonNaming(PropertyNamingStrategy.LowerDotCaseStrategy.class)
 public class TopicSchemas {
 
-  private Optional<String> keySchemaFile;
-  private Optional<String> valueSchemaFile;
+  private Subject keySubject;
+  private Subject valueSubject;
 
-  public TopicSchemas() {
-    this(Optional.empty(), Optional.empty());
-  }
-
-  public TopicSchemas(Optional<JsonNode> keyJsonNode, Optional<JsonNode> valueJsonNode) {
-    this.keySchemaFile = Optional.empty();
-    this.valueSchemaFile = Optional.empty();
-    keyJsonNode.ifPresent(node -> keySchemaFile = Optional.of(node.asText()));
-    valueJsonNode.ifPresent(node -> valueSchemaFile = Optional.of(node.asText()));
+  public TopicSchemas(
+      Optional<JsonNode> keyJsonNode,
+      Optional<JsonNode> keyRecordJsonNode,
+      Optional<JsonNode> valueJsonNode,
+      Optional<JsonNode> valueRecordJsonNode) {
+    this.keySubject = new Subject(keyJsonNode, keyRecordJsonNode, SubjectKind.KEY);
+    this.valueSubject = new Subject(valueJsonNode, valueRecordJsonNode, SubjectKind.VALUE);
   }
 
   public TopicSchemas(String keySchemaFile, String valueSchemaFile) {
-    this.keySchemaFile = Optional.of(keySchemaFile);
-    this.valueSchemaFile = Optional.of(valueSchemaFile);
+    this.keySubject = new Subject(keySchemaFile, null, SubjectKind.KEY);
+    this.valueSubject = new Subject(valueSchemaFile, null, SubjectKind.VALUE);
   }
 
-  public Optional<String> getKeySchemaFile() {
-    return keySchemaFile;
+  public Subject getKeySubject() {
+    return keySubject;
   }
 
-  public Optional<String> getValueSchemaFile() {
-    return valueSchemaFile;
+  public Subject getValueSubject() {
+    return valueSubject;
   }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/model/schema/Subject.java b/src/main/java/com/purbon/kafka/topology/model/schema/Subject.java
new file mode 100644
index 0000000..83bceac
--- /dev/null
+++ b/src/main/java/com/purbon/kafka/topology/model/schema/Subject.java
@@ -0,0 +1,64 @@
+package com.purbon.kafka.topology.model.schema;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.purbon.kafka.topology.model.Topic;
+import java.io.IOException;
+import java.util.Optional;
+
+public class Subject {
+
+  private Optional<String> schemaFile;
+  private Optional<String> recordType;
+  private SubjectKind kind;
+
+  public enum SubjectKind {
+    KEY("key"),
+    VALUE("value");
+
+    private final String label;
+
+    SubjectKind(String label) {
+      this.label = label;
+    }
+  }
+
+  public Subject(
+      Optional<JsonNode> schemaFileJsonNode,
+      Optional<JsonNode> recordTypeJsonNode,
+      SubjectKind kind) {
+    this.schemaFile = schemaFileJsonNode.map(JsonNode::asText);
+    this.recordType = recordTypeJsonNode.map(JsonNode::asText);
+    this.kind = kind;
+  }
+
+  public Subject(String schemaFile, String recordType, SubjectKind kind) {
+    this.schemaFile = Optional.ofNullable(schemaFile);
+    this.recordType = Optional.ofNullable(recordType);
+    this.kind = kind;
+  }
+
+  public String getSchemaFile() throws IOException {
+    return schemaFile.orElseThrow(() -> new IOException("SchemaFile not present"));
+  }
+
+  public boolean hasSchemaFile() {
+    return schemaFile.isPresent();
+  }
+
+  private String recordTypeAsString() throws IOException {
+    return recordType.orElseThrow(() -> new IOException("Missing record type for " + schemaFile));
+  }
+
+  public String buildSubjectName(Topic topic) throws IOException {
+    switch (topic.getSubjectNameStrategy()) {
+      case TOPIC_NAME_STRATEGY:
+        return topic.toString() + "-" + kind.label;
+      case RECORD_NAME_STRATEGY:
+        return recordTypeAsString();
+      case TOPIC_RECORD_NAME_STRATEGY:
+        return topic.toString() + "-" + recordTypeAsString();
+      default:
+        return "";
+    }
+  }
+}
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 d71387a..df3b960 100644
--- a/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
+++ b/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
@@ -1,25 +1,30 @@
 package com.purbon.kafka.topology.serdes;
 
+import static com.purbon.kafka.topology.model.SubjectNameStrategy.TOPIC_NAME_STRATEGY;
 import static com.purbon.kafka.topology.serdes.JsonSerdesUtils.validateRequiresKeys;
+import static java.util.Collections.singletonList;
 
 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.fasterxml.jackson.databind.node.ArrayNode;
 import com.purbon.kafka.topology.TopologyBuilderConfig;
+import com.purbon.kafka.topology.exceptions.ValidationException;
 import com.purbon.kafka.topology.model.Impl.TopicImpl;
 import com.purbon.kafka.topology.model.PlanMap;
+import com.purbon.kafka.topology.model.SubjectNameStrategy;
+import com.purbon.kafka.topology.model.Topic;
 import com.purbon.kafka.topology.model.TopicSchemas;
 import com.purbon.kafka.topology.model.User;
 import com.purbon.kafka.topology.model.users.Consumer;
 import com.purbon.kafka.topology.model.users.Producer;
+import com.purbon.kafka.topology.utils.Either;
 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 java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -27,6 +32,9 @@ public class TopicCustomDeserializer extends StdDeserializer<TopicImpl> {
 
   private static final Logger LOGGER = LogManager.getLogger(TopicCustomDeserializer.class);
 
+  private List<String> validSchemaKeys =
+      Arrays.asList("key.schema.file", "key.record.type", "value.schema.file", "value.record.type");
+
   private final TopologyBuilderConfig config;
   private PlanMap plans;
 
@@ -86,26 +94,100 @@ public class TopicCustomDeserializer extends StdDeserializer<TopicImpl> {
     TopicImpl topic =
         new TopicImpl(name, producers, consumers, optionalDataType, config, this.config);
 
-    Optional.ofNullable(rootNode.get("schemas"))
-        .ifPresent(
-            node -> {
-              topic.setSchemas(
-                  Optional.of(
-                      new TopicSchemas(
-                          Optional.ofNullable(node.get("key.schema.file")),
-                          Optional.ofNullable(node.get("value.schema.file")))));
-            });
-    if (topic.getSchemas().isPresent()
-        && !topic.getSchemas().get().getValueSchemaFile().isPresent()) {
+    Optional<SubjectNameStrategy> subjectNameStrategy =
+        Optional.ofNullable(rootNode.get("subject.name.strategy"))
+            .map(JsonNode::asText)
+            .map(SubjectNameStrategy::valueOfLabel);
+    topic.setSubjectNameStrategy(subjectNameStrategy);
+
+    List<TopicSchemas> schemas = new ArrayList<>();
+
+    if (rootNode.get("schemas") != null) {
+      JsonNode schemasNode = rootNode.get("schemas");
+      Iterator<JsonNode> it =
+          schemasNode instanceof ArrayNode
+              ? schemasNode.elements()
+              : singletonList(schemasNode).iterator();
+      Iterable<JsonNode> iterable = () -> it;
+
+      List<Either<ValidationException, TopicSchemas>> listOfResultsOrErrors =
+          StreamSupport.stream(iterable.spliterator(), true)
+              .map(validateAndBuildSchemas(topic))
+              .collect(Collectors.toList());
+
+      List<ValidationException> errors =
+          listOfResultsOrErrors.stream()
+              .filter(Either::isLeft)
+              .map(Either::getLeft)
+              .map(Optional::get)
+              .collect(Collectors.toList());
+      if (errors.size() > 0) {
+        throw new IOException(errors.get(0));
+      }
+
+      schemas =
+          listOfResultsOrErrors.stream()
+              .filter(Either::isRight)
+              .map(Either::getRight)
+              .map(Optional::get)
+              .collect(Collectors.toList());
+    }
+
+    if (schemas.size() > 1 && topic.getSubjectNameStrategy().equals(TOPIC_NAME_STRATEGY)) {
       throw new IOException(
           String.format(
-              "Missing required value.schema.file on schemas for topic %s", topic.getName()));
+              "%s is not a valid strategy when registering multiple schemas", TOPIC_NAME_STRATEGY));
     }
+
+    topic.setSchemas(schemas);
+
     LOGGER.debug(
         String.format("Topic %s with config %s has been created", topic.getName(), config));
     return topic;
   }
 
+  private Function<JsonNode, Either<ValidationException, TopicSchemas>> validateAndBuildSchemas(
+      Topic topic) {
+    return node -> {
+      List<String> elements = new ArrayList<>();
+      node.fieldNames().forEachRemaining(elements::add);
+      try {
+        validateSchemaKeys(elements, topic);
+        TopicSchemas schema =
+            new TopicSchemas(
+                Optional.ofNullable(node.get("key.schema.file")),
+                Optional.ofNullable(node.get("key.record.type")),
+                Optional.ofNullable(node.get("value.schema.file")),
+                Optional.ofNullable(node.get("value.record.type")));
+        return Either.Right(schema);
+      } catch (ValidationException ex) {
+        return Either.Left(ex);
+      }
+    };
+  }
+
+  private void validateSchemaKeys(List<String> elements, Topic topic) throws ValidationException {
+    for (String element : elements) {
+      if (!validSchemaKeys.contains(element)) {
+        throw new ValidationException(
+            String.format("Key %s is not a valid Topic Schema property", element));
+      }
+    }
+    if (!topic.getSubjectNameStrategy().equals(TOPIC_NAME_STRATEGY)) {
+      if (!elements.contains("key.record.type") && !elements.contains("value.record.type")) {
+        throw new ValidationException(
+            String.format(
+                "For a subject name strategy %s record.type is required!",
+                topic.getSubjectNameStrategy()));
+      }
+    }
+    if (!elements.contains("value.schema.file")) {
+      throw new ValidationException(
+          String.format(
+              "Missing required value.schema.file on schemas for topic %s", topic.getName()));
+    }
+  }
+
   private <T extends User> List<T> getUsers(
       JsonParser parser, JsonNode rootNode, String fieldName, Class<T> tClass)
       throws com.fasterxml.jackson.core.JsonProcessingException {


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/TopologyBuilderConfigTest.java b/src/test/java/com/purbon/kafka/topology/TopologyBuilderConfigTest.java
index 47db665..994f272 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologyBuilderConfigTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologyBuilderConfigTest.java
@@ -50,7 +50,7 @@ public class TopologyBuilderConfigTest {
     Project project = new ProjectImpl();
     Topic topic = new TopicImpl();
     TopicSchemas schema = new TopicSchemas("foo", "bar");
-    topic.setSchemas(Optional.of(schema));
+    topic.setSchemas(Collections.singletonList(schema));
     project.addTopic(topic);
     topology.addProject(project);
     props.put(CONFLUENT_SCHEMA_REGISTRY_URL_CONFIG, "mock://");
@@ -65,7 +65,7 @@ public class TopologyBuilderConfigTest {
     Project project = new ProjectImpl();
     Topic topic = new TopicImpl();
     TopicSchemas schema = new TopicSchemas("foo", "bar");
-    topic.setSchemas(Optional.of(schema));
+    topic.setSchemas(Collections.singletonList(schema));
     project.addTopic(topic);
     topology.addProject(project);
 
@@ -152,7 +152,6 @@ public class TopologyBuilderConfigTest {
     project.addTopic(topic);
     topology.addProject(project);
 
-    props.put(PROJECT_PREFIX_FORMAT_CONFIG, "{{foo}}{{topic}}");
     props.put(PROJECT_PREFIX_FORMAT_CONFIG, "{{banana}}");
 
     TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
@@ -166,7 +165,7 @@ public class TopologyBuilderConfigTest {
     cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, clientConfigFile);
 
     TopologyBuilderConfig config = TopologyBuilderConfig.build(cliOps);
-    assertThat(config.getKafkaInternalTopicPrefixes()).isEqualTo(Arrays.asList("_"));
+    assertThat(config.getKafkaInternalTopicPrefixes()).isEqualTo(Collections.singletonList("_"));
   }
 
   @Test
diff --git a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
index 4e8daf2..458de90 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
@@ -4,6 +4,8 @@ import static com.purbon.kafka.topology.BuilderCLI.ADMIN_CLIENT_CONFIG_OPTION;
 import static com.purbon.kafka.topology.BuilderCLI.BROKERS_OPTION;
 import static com.purbon.kafka.topology.TopologyBuilderConfig.TOPIC_PREFIX_FORMAT_CONFIG;
 import static com.purbon.kafka.topology.TopologyBuilderConfig.TOPIC_PREFIX_SEPARATOR_CONFIG;
+import static com.purbon.kafka.topology.model.SubjectNameStrategy.TOPIC_NAME_STRATEGY;
+import static com.purbon.kafka.topology.model.SubjectNameStrategy.TOPIC_RECORD_NAME_STRATEGY;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -201,10 +203,30 @@ public class TopologySerdesTest {
   }
 
   @Test(expected = IllegalArgumentException.class)
-  public void testSchemaSerdes() {
+  public void testInvalidSchemaSerdes() {
     parser.deserialise(TestUtils.getResourceFile("/descriptor-wrong-schemas.yaml"));
   }
 
+  @Test
+  public void testSchemaSerdes() {
+    Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
+    Project project = topology.getProjects().get(0);
+    List<Topic> topics = project.getTopics();
+    Optional<Topic> topicBar = topics.stream().filter(t -> t.getName().equals("bar")).findFirst();
+    Optional<Topic> topicCat = topics.stream().filter(t -> t.getName().equals("cat")).findFirst();
+
+    assertThat(topicBar).isPresent();
+    assertThat(topicBar.get().getSchemas()).hasSize(1);
+    assertThat(topicBar.get().getSchemas().get(0).getValueSubject().hasSchemaFile()).isTrue();
+    assertThat(topicBar.get().getSubjectNameStrategy()).isEqualTo(TOPIC_NAME_STRATEGY);
+
+    assertThat(topicCat).isPresent();
+    assertThat(topicCat.get().getSchemas()).hasSize(2);
+    assertThat(topicCat.get().getSchemas().get(0).getValueSubject().hasSchemaFile()).isTrue();
+    assertThat(topicCat.get().getSchemas().get(1).getValueSubject().hasSchemaFile()).isTrue();
+    assertThat(topicCat.get().getSubjectNameStrategy()).isEqualTo(TOPIC_RECORD_NAME_STRATEGY);
+  }
+
   @Test
   public void testPlaformProcessing() {
     Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
diff --git a/src/test/resources/descriptor.yaml b/src/test/resources/descriptor.yaml
index ad085c7..62a4475 100644
--- a/src/test/resources/descriptor.yaml
+++ b/src/test/resources/descriptor.yaml
@@ -46,12 +46,22 @@ projects:
       - name: "bar"
         dataType: "avro"
         schemas:
-          key.schema.file: "schemas/bar-key.avsc"
           value.schema.file: "schemas/bar-value.avsc"
+      - name: "cat"
+        dataType: "avro"
+        subject.name.strategy: "TopicRecordNameStrategy"
+        schemas:
+          - key.schema.file: "schemas/bar-key.avsc"
+            value.schema.file: "schemas/bar-value.avsc"
+            key.record.type: "foo"
+            value.record.type: "bar"
+          - value.schema.file: "schemas/bar-value.avsc"
+            value.record.type: "value"
   - name: "bar"
     topics:
       - dataType: "avro"
         schemas:
+          key.schema.file: "schemas/bar-key.avsc"
           value.schema.file: "schemas/bar-value.avsc"
         name: "bar"
         config: