Repository: kafka-ops/kafka-topology-builder
Pull request #126 (2020-11-11 14:31:16)

Make writing .cluster-state file idempotent
Running KTB twice with the same topology will still cause the .cluster-state file to change on every run. Because of usage of HashSet in a number of places the order of ACLs in the file will change on each run. This is undesired and especially a problem if using the FileBackend and wanting to push the file to some repo if changed.

Changed to LinkedHashSet to fix this.

akselh: This should fix #136. Sorry for not creating the issue first.
akselh: @purbon , what du you think? There are small changes, but still the PR have been pending for a while. 

Btw: I saw now that I had messed up with the parent branch. (Wasn't quite sure about your branch strategy). That is fixed now.
purbon: Hey @akselh, yeah not a big problem with this right now. Sorry I have not had much time to review properly the PR. I will take care of it this week, you know .. customers :-P 
akselh: Thank you! Fully understands that maintaining this project is not your sole task. Appreciate your efforts!

Patch:
diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml
new file mode 100644
index 0000000..a5a02cf
--- /dev/null
+++ b/.github/workflows/docker.yml
@@ -0,0 +1,49 @@
+name: Docker publish latest
+on:
+  schedule:
+    - cron: '* 0 * * *'
+  push:
+    branches:
+      - master
+jobs:
+  ktb_build:
+    name: Build the KTB package
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up the JDK
+        uses: actions/setup-java@v1
+        with:
+          java-version: 1.8
+      - name: Cache dependencies
+        uses: actions/cache@v2
+        with:
+          path: ~/.m2
+          key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}
+          restore-keys: ${{ runner.os }}-m2
+      - name: Build with Maven
+        run: mvn -B package --file pom.xml
+  docker_build:
+    name: Docker latest image build
+    runs-on: ubuntu-latest
+    needs: [ktb_build]
+    defaults:
+      run:
+        working-directory: release/docker
+    steps:
+      - name: Check out the repo
+        uses: actions/checkout@v2
+      - name: list files
+        run: ls -l
+      - name: Copy Jar and Build docker image
+        run: sh build.sh
+      - name: Login to DockerHub
+        uses: docker/login-action@v1
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+      - name: Push to Docker Hub
+        uses: docker/build-push-action@v2
+        with:
+          file: release/docker/Dockerfile
+          tags: purbon/kafka-topology-builder:latest
diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml
new file mode 100644
index 0000000..4692745
--- /dev/null
+++ b/.github/workflows/maven.yml
@@ -0,0 +1,34 @@
+# This workflow will build a Java project with Maven
+# For more information see: https://help.github.com/actions/language-and-framework-guides/building-and-testing-java-with-maven
+
+name: CI tests
+
+on:
+  push:
+    branches: [ master ]
+  pull_request:
+    branches: [ master ]
+
+jobs:
+  build:
+    strategy:
+      matrix:
+        os: [ubuntu-16.04, ubuntu-latest]
+        java: [8.0.192, 11.0.x, 13]
+
+    runs-on: ${{ matrix.os }}
+
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up the JDK
+        uses: actions/setup-java@v1
+        with:
+          java-version: ${{ matrix.java }}
+      - name: Cache dependencies
+        uses: actions/cache@v2
+        with:
+          path: ~/.m2
+          key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}
+          restore-keys: ${{ runner.os }}-m2
+      - name: Build with Maven
+        run: mvn -B package --file pom.xml
\ No newline at end of file
diff --git a/.github/workflows/stale.yml b/.github/workflows/stale.yml
new file mode 100644
index 0000000..72357df
--- /dev/null
+++ b/.github/workflows/stale.yml
@@ -0,0 +1,19 @@
+on:
+  schedule:
+    - cron: '0 * * * *'
+  push:
+    branches:
+      - master
+name: stale
+jobs:
+  stale:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/stale@v1
+        with:
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+          stale-pr-label: 'no-activity'
+          stale-pr-message: 'This pull request has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs.'
+          exempt-issue-label: 'epic'
+          days-before-stale: 30
+          days-before-close: 7
diff --git a/README.md b/README.md
index 1600913..dc46b54 100644
--- a/README.md
+++ b/README.md
@@ -162,7 +162,8 @@ projects:
       num.partitions: "3"
 ```
 
-more examples can be found at the [example/](example/) directory.
+more examples can be found at the [example/](example/) directory. 
+Also check the documentation on ACL and RBAC in [the docs](https://kafka-topology-builder.readthedocs.io/) for information on what ACLs are created.
 
 ## Interested in contributing back?
 
diff --git a/docs/config-values.rst b/docs/config-values.rst
index 57ba729..7d10364 100644
--- a/docs/config-values.rst
+++ b/docs/config-values.rst
@@ -31,6 +31,14 @@ A part from that, you need to setup the UUID for each of your clusters. This is
     topology.builder.mds.kafka.connect.cluster.id = "connect-cluster"
 
 
+Schema Management
+-----------
+If you plan to manage and deploy schemas with KTB, you must define the url to your Confluent Schema Registry as follows
+::
+    schema.registry.url = "http://localhost:8081"
+
+
+
 Topology Builder backend usage and selection
 -----------
 
diff --git a/docs/futures/what-acl-management.rst b/docs/futures/what-acl-management.rst
index 6ee13a1..542eb06 100644
--- a/docs/futures/what-acl-management.rst
+++ b/docs/futures/what-acl-management.rst
@@ -5,14 +5,14 @@ ACLs
 -----------
 
 In the topology descriptor files users can create permissions for different types of applications, Consumers, Producers, Kafka streams apps or Kafka Connectors.
-With this roles, users can easy create the permissions that map directly to their needs.
+With this roles, users can easily create the permissions that map directly to their needs.
 
 Consumers
 ^^^^^^^^^^^
 
 As a user you can configure consumers for each project.
-Consumer have a principal and optionally a consumer group name, if the consumer group is by default defined as prefix ACL with "*".
-Users can customize this property by defining a *group* attribute for each consumer.
+Consumer have a principal and optionally a consumer group name. The consumer group ACL is by default defined for all groups ("*").
+Users can customize this ACL by defining a *group* attribute for each consumer.
 
 
 .. code-block:: YAML
@@ -40,7 +40,7 @@ Consumer definition with principal "User:App0" and without an specific consumer
 
 Consumer definition with principal "User:App0" and consumer group name "foo".
 
-In the default mode the KTB will create dedicated ACL for each user and topic pair. For organisations that aim not to have dedicated pair or rules the KTB offer the option
+In the default mode the KTB will create dedicated ACL for each user and topic pair. For organisations that aim not to have dedicated pair of rules the KTB offer the option
 to optimise the number of ACLs using prefixed rules.
 
 The optimised ACLs/RBAC can be enabled using the *topology.acls.optimized* configuration property.
@@ -61,16 +61,65 @@ Producers have a principal.
         producers:
           - principal: "User:App0"
 
-In the default mode the KTB will create dedicated ACL for each user and topic pair. For organisations that aim not to have dedicated pair or rules the KTB offer the option
+In the default mode the KTB will create dedicated ACL for each user and topic pair. For organisations that aim not to have dedicated pair of rules the KTB offer the option
 to optimise the number of ACLs using prefixed rules.
 
 The optimised ACLs/RBAC can be enabled using the *topology.acls.optimized* configuration property.
 
+Streams
+^^^^^^^^^^^
+
+Users can also setup Kafka Streams applications.
+Each one of them will be composed of a principal and a list of topics that this principal needs to read and write.
+The principal is the user used by the streams app to connect to Kafka. You can also optionally specify the *applicationId*.
+
+.. code-block:: YAML
+
+  ---
+    context: "context"
+    source: "source"
+    projects:
+      - name: "foo"
+        streams:
+          - principal: "User:App0"
+            topics:
+              read:
+                - "topicA"
+              write:
+                - "topicB"
+
+KTB will create the necessary ACLs for reading and writing topics, as well as ACLs needed by the app to create/manage internal topics.
+The ACLs for the consumer group and for internal topic creation are prefixed.
+The resource name (prefix) is by default the topic name prefix in the project.
+For the example above the prefix will by default be "context.source.foo".
+
+As you see in the next example this can be overridden by specifying an *applicationId* in the topology.
+
+.. code-block:: YAML
+
+  ---
+    context: "context"
+    source: "source"
+    projects:
+      - name: "foo"
+        streams:
+          - principal: "User:App0"
+            applicationId: "streamsApplicationId"
+            topics:
+              read:
+                - "topicA"
+              write:
+                - "topicB"
+
+
+When the *applicationId* is specified this is used as the resource prefix in the ACLs for consumer groups
+and internal topics for the streams app. In the above example the prefix will be "streamsApplicationId".
+
 Connectors
 ^^^^^^^^^^^
 
 In a similar fashion as with the previous roles, users can setup specific Kafka Connect setups.
-Each one of them will be composed of a principal, this would be the user used by the connect to
+Each one of them will be composed of a principal, this would be the user used by the connector to
 connect to Kafka and a list of topics that this principal needs to read or write to, remember
 Connectors can either read (Sink) or write (Source) into Apache Kafka and they do it to topics.
 
@@ -133,13 +182,16 @@ What ACLs are created
 ^^^^^^^^^^^^^^^^^^^^^
 Kafka Topology Builder will assign the following ACLs:
 
-* each principal in the `consumers` list will get `READ` and `DESCRIBE` permissions on each topic in the containing project as well as `READ` access on every consumer group
-* each principal in the `producers` list will get `WRITE` and `DESCRIBE` permissions on each topic in the containing project
+* each principal in the `consumers` list will get `READ` and `DESCRIBE` permissions on each topic. In addition `READ` access on every consumer group (by default) or the group specified in the topology.
+
+* each principal in the `producers` list will get `WRITE` and `DESCRIBE` permissions on each topic. In addition if a *transactionId* is specified a WRITE and DESCRIBE ACL is created on the transactionId resource. And if either *transactionId* or *idempotence* is specified for the producer the IDEMPOTENT_WRITE ALLOW acl is created.
+
 * each principal in the `streams` list will get
 
   * `READ` access on every topic in its `read` sub-object
   * `WRITE` access on every topic `write` sub-object
-  * `ALL` access on every topic starting with fully-qualified project name, e.g. ``context.company.env.source.projectA`` in the example above. These are `PREFIXED` ACLs.
+  * `ALL` access on every topic starting with the fully-qualified project name (by default) or the given applicationId. These are `PREFIXED` ACLs.
+  * `READ` access on consumer groups starting with the fully-qualified project name (by default) or the given applicationId. These are `PREFIXED` ACLs.
 
 * each principal for a connector will get
 
diff --git a/docs/futures/what-schema-management.rst b/docs/futures/what-schema-management.rst
index 78ca84b..5f5df5d 100644
--- a/docs/futures/what-schema-management.rst
+++ b/docs/futures/what-schema-management.rst
@@ -34,7 +34,10 @@ 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.
+
+
 Supported Schema Registry
 -----------
 
-There are multiple options when using an Schema Registry with Kafka, currently only the Confluent Schema Registry is supported.
\ No newline at end of file
+There are multiple options when using an Schema Registry with Kafka, currently only the Confluent Schema Registry is supported.
diff --git a/example/descriptor.yaml b/example/descriptor.yaml
index 3a24795..06ae802 100644
--- a/example/descriptor.yaml
+++ b/example/descriptor.yaml
@@ -52,6 +52,24 @@ projects:
         config:
           replication.factor: "1"
           num.partitions: "1"
+  - name: "projectC"
+    streams:
+      - principal: "User:App0"
+        applicationId: "streamsApplicationId"
+        topics:
+          read:
+            - "topicE"
+          write:
+            - "topicF"
+    topics:
+      - name: "topicE"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: "topicF"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
 platform:
   schema_registry:
     instances:
diff --git a/example/topology-builder-rbac-schemas.properties b/example/topology-builder-rbac-schemas.properties
new file mode 100644
index 0000000..74469fc
--- /dev/null
+++ b/example/topology-builder-rbac-schemas.properties
@@ -0,0 +1,14 @@
+sasl.mechanism=PLAIN
+security.protocol=SASL_PLAINTEXT
+sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="professor" password="professor" ;
+
+topology.builder.access.control.class = com.purbon.kafka.topology.roles.RBACProvider
+topology.builder.mds.server = http://localhost:8090
+topology.builder.mds.user = professor
+topology.builder.mds.password = professor
+topology.builder.mds.kafka.cluster.id = flAGMST0QVCx7EynQEkQdw
+
+topology.builder.mds.schema.registry.cluster.id = schema-registry-cluster
+topology.builder.mds.kafka.connect.cluster.id = connect-cluster
+
+schema.registry.url = http://localhost:8081
diff --git a/pom.xml b/pom.xml
index c724653..e1bf35f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -328,7 +328,7 @@
                   <type>file</type>
                   <mapper>
                     <type>perm</type>
-                    <prefix>/usr/local/kafka-topology-builder</prefix>
+                    <prefix>/usr/local/kafka-topology-builder/bin</prefix>
                     <user>ktb-kafka</user>
                     <group>ktb-kafka</group>
                     <filemode>440</filemode>
@@ -423,7 +423,7 @@
     <commons.version>1.4</commons.version>
     <mockito.version>3.6.0</mockito.version>
     <junit.version>4.13.1</junit.version>
-    <testcontainers.version>1.15.0-rc2</testcontainers.version>
+    <testcontainers.version>1.15.0</testcontainers.version>
     <jedis.version>3.2.0</jedis.version>
     <confluent.version>5.5.0</confluent.version>
     <confluent-ce.version>5.5.1-ce</confluent-ce.version>
@@ -557,7 +557,7 @@
     </dependency>
     <dependency>
       <groupId>org.testcontainers</groupId>
-      <artifactId>kafka</artifactId>
+      <artifactId>testcontainers</artifactId>
       <version>${testcontainers.version}</version>
       <scope>test</scope>
     </dependency>
diff --git a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
index 25a1ae5..ff50f8a 100644
--- a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
+++ b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
@@ -25,7 +25,7 @@ import com.purbon.kafka.topology.model.users.platform.SchemaRegistryInstance;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import java.io.PrintStream;
 import java.util.ArrayList;
-import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -136,7 +136,7 @@ public class AccessControlManager {
         .forEach(
             topic -> {
               final String fullTopicName = topic.toString();
-              Set<Consumer> consumers = new HashSet<>(topic.getConsumers());
+              Set<Consumer> consumers = new LinkedHashSet<>(topic.getConsumers());
               if (includeProjectLevel) {
                 consumers.addAll(project.getConsumers());
               }
@@ -146,7 +146,7 @@ public class AccessControlManager {
                         bindingsBuilder, new ArrayList<>(consumers), fullTopicName, false);
                 actions.add(action);
               }
-              Set<Producer> producers = new HashSet<>(topic.getProducers());
+              Set<Producer> producers = new LinkedHashSet<>(topic.getProducers());
               if (includeProjectLevel) {
                 producers.addAll(project.getProducers());
               }
@@ -173,13 +173,13 @@ public class AccessControlManager {
     List<Action> updateActions = new ArrayList<>();
 
     Set<TopologyAclBinding> allFinalBindings =
-        actions.stream().flatMap(actionApplyFunction()).collect(Collectors.toSet());
+        actions.stream().flatMap(actionApplyFunction()).collect(Collectors.toCollection(LinkedHashSet::new));
 
     // Diff of bindings, so we only create what is not already created in the cluster.
     Set<TopologyAclBinding> bindingsToBeCreated =
         allFinalBindings.stream()
             .filter(binding -> !bindings.contains(binding))
-            .collect(Collectors.toSet());
+            .collect(Collectors.toCollection(LinkedHashSet::new));
 
     if (!bindingsToBeCreated.isEmpty()) {
       CreateBindings createBindings = new CreateBindings(controlProvider, bindingsToBeCreated);
diff --git a/src/main/java/com/purbon/kafka/topology/BackendController.java b/src/main/java/com/purbon/kafka/topology/BackendController.java
index fcd13d7..632ff78 100644
--- a/src/main/java/com/purbon/kafka/topology/BackendController.java
+++ b/src/main/java/com/purbon/kafka/topology/BackendController.java
@@ -4,7 +4,7 @@ import com.purbon.kafka.topology.backend.Backend;
 import com.purbon.kafka.topology.backend.FileBackend;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import java.io.IOException;
-import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 import org.apache.logging.log4j.LogManager;
@@ -30,7 +30,7 @@ public class BackendController {
 
   public BackendController(Backend backend) {
     this.backend = backend;
-    this.bindings = new HashSet<>();
+    this.bindings = new LinkedHashSet<>();
   }
 
   public void add(List<TopologyAclBinding> bindings) {
@@ -44,7 +44,7 @@ public class BackendController {
   }
 
   public Set<TopologyAclBinding> getBindings() {
-    return new HashSet<>(bindings);
+    return new LinkedHashSet<>(bindings);
   }
 
   public void flushAndClose() {
diff --git a/src/main/java/com/purbon/kafka/topology/ExecutionPlan.java b/src/main/java/com/purbon/kafka/topology/ExecutionPlan.java
index dc1aab8..271c6e0 100644
--- a/src/main/java/com/purbon/kafka/topology/ExecutionPlan.java
+++ b/src/main/java/com/purbon/kafka/topology/ExecutionPlan.java
@@ -6,7 +6,7 @@ import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import java.io.IOException;
 import java.io.PrintStream;
 import java.util.ArrayList;
-import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -26,7 +26,7 @@ public class ExecutionPlan {
       List<Action> plan, PrintStream outputStream, BackendController backendController) {
     this.plan = plan;
     this.outputStream = outputStream;
-    this.bindings = new HashSet<>();
+    this.bindings = new LinkedHashSet<>();
     this.backendController = backendController;
     if (backendController.size() > 0) {
       this.bindings.addAll(backendController.getBindings());
@@ -74,7 +74,7 @@ public class ExecutionPlan {
           bindings =
               bindings.stream()
                   .filter(binding -> !action.getBindings().contains(binding))
-                  .collect(Collectors.toSet());
+                  .collect(Collectors.toCollection(LinkedHashSet::new));
         } else {
           bindings.addAll(action.getBindings());
         }
diff --git a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
index 89901b6..0371066 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
@@ -13,7 +13,6 @@ import com.typesafe.config.ConfigFactory;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Properties;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -144,7 +143,7 @@ public class TopologyBuilderConfig {
   }
 
   private void validateGeneralConfiguration(Topology topology) throws ConfigurationException {
-    if (countOfSchemas(topology) > 0) {
+    if (hasSchemas(topology)) {
       raiseIfDefault(CONFLUENT_SCHEMA_REGISTRY_URL_CONFIG, "mock://");
     }
 
@@ -188,12 +187,10 @@ public class TopologyBuilderConfig {
     }
   }
 
-  private static long countOfSchemas(Topology topology) {
+  private static boolean hasSchemas(Topology topology) {
     return topology.getProjects().stream()
         .flatMap((Function<Project, Stream<Topic>>) project -> project.getTopics().stream())
-        .map(Topic::getSchemas)
-        .filter(Objects::nonNull)
-        .count();
+        .anyMatch(topic -> topic.getSchemas().isPresent());
   }
 
   private void raiseIfDefault(String key, String _default) throws ConfigurationException {
diff --git a/src/main/java/com/purbon/kafka/topology/TopologyObjectBuilder.java b/src/main/java/com/purbon/kafka/topology/TopologyObjectBuilder.java
index a531a58..ff20077 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyObjectBuilder.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyObjectBuilder.java
@@ -1,6 +1,5 @@
 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;
@@ -11,13 +10,9 @@ 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 TopologyObjectBuilder {
 
-  private static final Logger LOGGER = LogManager.getLogger(TopologyObjectBuilder.class);
-
   public static Topology build(String fileOrDir) throws IOException {
     return build(fileOrDir, "", new TopologyBuilderConfig());
   }
@@ -60,15 +55,7 @@ public class TopologyObjectBuilder {
     if (isDir) {
       Files.list(Paths.get(fileOrDir))
           .sorted()
-          .map(
-              path -> {
-                try {
-                  return parser.deserialise(path.toFile());
-                } catch (IOException e) {
-                  LOGGER.error(e);
-                  return new TopologyImpl();
-                }
-              })
+          .map(path -> parser.deserialise(path.toFile()))
           .forEach(subTopology -> topologies.add(subTopology));
     } else {
       Topology firstTopology = parser.deserialise(new File(fileOrDir));
diff --git a/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreams.java b/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreams.java
index 63a7b47..71a6d4e 100644
--- a/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreams.java
+++ b/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreams.java
@@ -3,7 +3,6 @@ package com.purbon.kafka.topology.actions.access.builders;
 import com.purbon.kafka.topology.BindingsBuilderProvider;
 import com.purbon.kafka.topology.actions.BaseAccessControlAction;
 import com.purbon.kafka.topology.model.users.KStream;
-import java.io.IOException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -12,24 +11,24 @@ public class BuildBindingsForKStreams extends BaseAccessControlAction {
 
   private final BindingsBuilderProvider builderProvider;
   private final KStream app;
-  private final String topicPrefix;
+  private final String prefix;
 
   public BuildBindingsForKStreams(
       BindingsBuilderProvider builderProvider, KStream app, String topicPrefix) {
     super();
     this.builderProvider = builderProvider;
     this.app = app;
-    this.topicPrefix = topicPrefix;
+    this.prefix = app.getApplicationId().orElse(topicPrefix);
   }
 
   @Override
-  protected void execute() throws IOException {
+  protected void execute() {
     List<String> readTopics = app.getTopics().get(KStream.READ_TOPICS);
     List<String> writeTopics = app.getTopics().get(KStream.WRITE_TOPICS);
 
     bindings =
         builderProvider.buildBindingsForStreamsApp(
-            app.getPrincipal(), topicPrefix, readTopics, writeTopics);
+            app.getPrincipal(), prefix, readTopics, writeTopics);
   }
 
   @Override
@@ -37,7 +36,7 @@ public class BuildBindingsForKStreams extends BaseAccessControlAction {
     Map<String, Object> map = new HashMap<>();
     map.put("Operation", getClass().getName());
     map.put("Principal", app.getPrincipal());
-    map.put("Topic", topicPrefix);
+    map.put("Topic", prefix);
     return map;
   }
 }
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 a24b022..563f4fc 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
@@ -54,8 +54,8 @@ public class SyncTopicAction extends BaseAction {
       adminClient.createTopic(topic, fullTopicName);
     }
 
-    if (topic.getSchemas() != null) {
-      final TopicSchemas schemas = topic.getSchemas();
+    if (topic.getSchemas().isPresent()) {
+      final TopicSchemas schemas = topic.getSchemas().get();
 
       schemas
           .getKeySchemaFile()
diff --git a/src/main/java/com/purbon/kafka/topology/backend/FileBackend.java b/src/main/java/com/purbon/kafka/topology/backend/FileBackend.java
index 00f54aa..90c6389 100644
--- a/src/main/java/com/purbon/kafka/topology/backend/FileBackend.java
+++ b/src/main/java/com/purbon/kafka/topology/backend/FileBackend.java
@@ -2,15 +2,11 @@ package com.purbon.kafka.topology.backend;
 
 import com.purbon.kafka.topology.BackendController.Mode;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileReader;
-import java.io.IOException;
-import java.io.RandomAccessFile;
+import java.io.*;
 import java.net.URI;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.Set;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -20,6 +16,7 @@ import org.apache.logging.log4j.Logger;
 public class FileBackend implements Backend {
 
   private static final Logger LOGGER = LogManager.getLogger(FileBackend.class);
+  public static final String STATE_FILE_NAME = ".cluster-state";
 
   private RandomAccessFile writer;
   private String expression =
@@ -39,8 +36,8 @@ public class FileBackend implements Backend {
   @Override
   public void createOrOpen(Mode mode) {
     try {
-      writer = new RandomAccessFile(filename(), "rw");
-      Path path = Paths.get(filename());
+      writer = new RandomAccessFile(STATE_FILE_NAME, "rw");
+      Path path = Paths.get(STATE_FILE_NAME);
       if (path.toFile().exists()) {
         writer.seek(0);
         if (mode.equals(Mode.TRUNCATE)) {
@@ -56,13 +53,13 @@ public class FileBackend implements Backend {
     if (writer == null) {
       throw new IOException("state file does not exist");
     }
-    File file = new File(filename());
+    File file = new File(STATE_FILE_NAME);
     return load(file.toURI());
   }
 
   public Set<TopologyAclBinding> load(URI uri) throws IOException {
     Path filePath = Paths.get(uri);
-    Set<TopologyAclBinding> bindings = new HashSet<>();
+    Set<TopologyAclBinding> bindings = new LinkedHashSet<>();
     BufferedReader in = new BufferedReader(new FileReader(filePath.toFile()));
     String type = in.readLine();
     String line = null;
@@ -132,8 +129,4 @@ public class FileBackend implements Backend {
       LOGGER.error(e);
     }
   }
-
-  private String filename() {
-    return ".cluster-state";
-  }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/exceptions/TopologyParsingException.java b/src/main/java/com/purbon/kafka/topology/exceptions/TopologyParsingException.java
new file mode 100644
index 0000000..19d999f
--- /dev/null
+++ b/src/main/java/com/purbon/kafka/topology/exceptions/TopologyParsingException.java
@@ -0,0 +1,7 @@
+package com.purbon.kafka.topology.exceptions;
+
+public class TopologyParsingException extends RuntimeException {
+  public TopologyParsingException(String msg, Throwable e) {
+    super(msg, e);
+  }
+}
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 774e582..3547d5c 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
@@ -24,7 +24,7 @@ public class TopicImpl implements Topic, Cloneable {
   private Optional<String> dataType;
 
   @JsonInclude(Include.NON_EMPTY)
-  private TopicSchemas schemas;
+  private Optional<TopicSchemas> schemas;
 
   private String name;
 
@@ -89,7 +89,7 @@ public class TopicImpl implements Topic, Cloneable {
     this.consumers = consumers;
     this.dataType = dataType;
     this.config = config;
-    this.schemas = new TopicSchemas();
+    this.schemas = Optional.empty();
     this.replicationFactor = 0;
     this.partitionCount = 0;
     this.appConfig = appConfig;
@@ -103,16 +103,16 @@ public class TopicImpl implements Topic, Cloneable {
     return name;
   }
 
-  public TopicSchemas getSchemas() {
-    return schemas;
-  }
-
   @Override
   public String getPlan() {
     return plan;
   }
 
-  public void setSchemas(TopicSchemas schemas) {
+  public Optional<TopicSchemas> getSchemas() {
+    return schemas;
+  }
+
+  public void setSchemas(Optional<TopicSchemas> schemas) {
     this.schemas = schemas;
   }
 
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 fe2e25f..2d71fe1 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();
 
-  TopicSchemas getSchemas();
+  Optional<TopicSchemas> getSchemas();
 
   String getPlan();
 
-  void setSchemas(TopicSchemas schemas);
+  void setSchemas(Optional<TopicSchemas> schemas);
 
   Map<String, String> getConfig();
 
diff --git a/src/main/java/com/purbon/kafka/topology/model/users/KStream.java b/src/main/java/com/purbon/kafka/topology/model/users/KStream.java
index 8943068..7604d8e 100644
--- a/src/main/java/com/purbon/kafka/topology/model/users/KStream.java
+++ b/src/main/java/com/purbon/kafka/topology/model/users/KStream.java
@@ -1,16 +1,36 @@
 package com.purbon.kafka.topology.model.users;
 
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
 import com.purbon.kafka.topology.model.DynamicUser;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Optional;
 
 public class KStream extends DynamicUser {
 
+  @JsonInclude(Include.NON_EMPTY)
+  private Optional<String> applicationId;
+
   public KStream() {
-    super();
+    this("", new HashMap<>());
   }
 
-  public KStream(String principal, HashMap<String, List<String>> topics) {
+  public KStream(
+      String principal, HashMap<String, List<String>> topics, Optional<String> applicationId) {
     super(principal, topics);
+    this.applicationId = applicationId;
+  }
+
+  public KStream(String principal, HashMap<String, List<String>> topics) {
+    this(principal, topics, Optional.empty());
+  }
+
+  public Optional<String> getApplicationId() {
+    return applicationId;
+  }
+
+  public void setApplicationId(Optional<String> applicationId) {
+    this.applicationId = applicationId;
   }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java b/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java
index aa657fc..f536c29 100644
--- a/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java
+++ b/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java
@@ -82,8 +82,8 @@ public class AclsBindingsBuilder implements BindingsBuilderProvider {
 
   @Override
   public List<TopologyAclBinding> buildBindingsForStreamsApp(
-      String principal, String topicPrefix, List<String> readTopics, List<String> writeTopics) {
-    return toList(streamsAppStream(principal, topicPrefix, readTopics, writeTopics));
+      String principal, String prefix, List<String> readTopics, List<String> writeTopics) {
+    return toList(streamsAppStream(principal, prefix, readTopics, writeTopics));
   }
 
   @Override
@@ -165,7 +165,7 @@ public class AclsBindingsBuilder implements BindingsBuilderProvider {
   }
 
   private Stream<AclBinding> streamsAppStream(
-      String principal, String topicPrefix, List<String> readTopics, List<String> writeTopics) {
+      String principal, String prefix, List<String> readTopics, List<String> writeTopics) {
 
     List<AclBinding> acls = new ArrayList<>();
 
@@ -178,7 +178,10 @@ public class AclsBindingsBuilder implements BindingsBuilderProvider {
             acls.add(
                 buildTopicLevelAcl(principal, topic, PatternType.LITERAL, AclOperation.WRITE)));
 
-    acls.add(buildTopicLevelAcl(principal, topicPrefix, PatternType.PREFIXED, AclOperation.ALL));
+    acls.add(buildTopicLevelAcl(principal, prefix, PatternType.PREFIXED, AclOperation.ALL));
+
+    acls.add(buildGroupLevelAcl(principal, prefix, PatternType.PREFIXED, AclOperation.READ));
+
     return acls.stream();
   }
 
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 2eed45f..d71387a 100644
--- a/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
+++ b/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
@@ -90,14 +90,16 @@ public class TopicCustomDeserializer extends StdDeserializer<TopicImpl> {
         .ifPresent(
             node -> {
               topic.setSchemas(
-                  new TopicSchemas(
-                      Optional.ofNullable(node.get("key.schema.file")),
-                      Optional.ofNullable(node.get("value.schema.file"))));
+                  Optional.of(
+                      new TopicSchemas(
+                          Optional.ofNullable(node.get("key.schema.file")),
+                          Optional.ofNullable(node.get("value.schema.file")))));
             });
-    if (rootNode.get("schemas") != null) {
-      if (!topic.getSchemas().getValueSchemaFile().isPresent()) {
-        throw new IOException("Missing required value.schema.file if key file is present");
-      }
+    if (topic.getSchemas().isPresent()
+        && !topic.getSchemas().get().getValueSchemaFile().isPresent()) {
+      throw new IOException(
+          String.format(
+              "Missing required value.schema.file on schemas for topic %s", topic.getName()));
     }
     LOGGER.debug(
         String.format("Topic %s with config %s has been created", topic.getName(), 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 6ee51c7..c9fc528 100644
--- a/src/main/java/com/purbon/kafka/topology/serdes/TopologySerdes.java
+++ b/src/main/java/com/purbon/kafka/topology/serdes/TopologySerdes.java
@@ -6,6 +6,7 @@ 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.TopologyBuilderConfig;
+import com.purbon.kafka.topology.exceptions.TopologyParsingException;
 import com.purbon.kafka.topology.model.Impl.TopicImpl;
 import com.purbon.kafka.topology.model.PlanMap;
 import com.purbon.kafka.topology.model.Topology;
@@ -41,12 +42,21 @@ public class TopologySerdes {
     mapper = ObjectMapperFactory.build(type, config, plans);
   }
 
-  public Topology deserialise(File file) throws IOException {
-    return mapper.readValue(file, Topology.class);
+  public Topology deserialise(File file) {
+    try {
+      return mapper.readValue(file, Topology.class);
+    } catch (IOException e) {
+      throw new TopologyParsingException(
+          "Failed to deserialize topology from " + file.getPath(), e);
+    }
   }
 
-  public Topology deserialise(String content) throws IOException {
-    return mapper.readValue(content, Topology.class);
+  public Topology deserialise(String content) {
+    try {
+      return mapper.readValue(content, Topology.class);
+    } catch (IOException e) {
+      throw new TopologyParsingException("Failed to deserialize topology from " + content, e);
+    }
   }
 
   public String serialise(Topology topology) throws JsonProcessingException {


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java b/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
index 98da7aa..6839f13 100644
--- a/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
@@ -10,48 +10,26 @@ import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Matchers.refEq;
-import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.*;
 
 import com.purbon.kafka.topology.actions.Action;
 import com.purbon.kafka.topology.api.adminclient.AclBuilder;
-import com.purbon.kafka.topology.model.Component;
+import com.purbon.kafka.topology.model.*;
 import com.purbon.kafka.topology.model.Impl.ProjectImpl;
 import com.purbon.kafka.topology.model.Impl.TopicImpl;
 import com.purbon.kafka.topology.model.Impl.TopologyImpl;
-import com.purbon.kafka.topology.model.Platform;
-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.model.User;
 import com.purbon.kafka.topology.model.users.Connector;
 import com.purbon.kafka.topology.model.users.Consumer;
 import com.purbon.kafka.topology.model.users.KStream;
 import com.purbon.kafka.topology.model.users.Producer;
-import com.purbon.kafka.topology.model.users.platform.ControlCenter;
-import com.purbon.kafka.topology.model.users.platform.ControlCenterInstance;
-import com.purbon.kafka.topology.model.users.platform.Kafka;
-import com.purbon.kafka.topology.model.users.platform.KafkaConnect;
-import com.purbon.kafka.topology.model.users.platform.SchemaRegistry;
-import com.purbon.kafka.topology.model.users.platform.SchemaRegistryInstance;
+import com.purbon.kafka.topology.model.users.platform.*;
 import com.purbon.kafka.topology.roles.SimpleAclsProvider;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import com.purbon.kafka.topology.roles.acls.AclsBindingsBuilder;
+import com.purbon.kafka.topology.utils.TestUtils;
 import java.io.IOException;
 import java.io.PrintStream;
-import java.nio.file.Files;
-import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
+import java.util.*;
 import java.util.stream.Collectors;
 import org.apache.kafka.common.acl.AclBinding;
 import org.apache.kafka.common.acl.AclOperation;
@@ -84,7 +62,7 @@ public class AccessControlManagerTest {
 
   @Before
   public void setup() throws IOException {
-    Files.deleteIfExists(Paths.get(".cluster-state"));
+    TestUtils.deleteStateFile();
     plan = ExecutionPlan.init(backendController, mockPrintStream);
     accessControlManager = new AccessControlManager(aclsProvider, aclsBuilder);
     doNothing().when(backendController).add(Matchers.anyList());
@@ -348,7 +326,7 @@ public class AccessControlManagerTest {
   }
 
   @Test
-  public void newControlCenterACLCreation() throws IOException {
+  public void newControlCenterACLCreation() {
 
     Project project = new ProjectImpl();
     Topology topology = new TopologyImpl();
@@ -425,8 +403,7 @@ public class AccessControlManagerTest {
   }
 
   @Test
-  public void newKafkaConnectACLsCreation() throws IOException {
-
+  public void newKafkaConnectACLsCreation() {
     Project project = new ProjectImpl();
 
     Connector connector1 = new Connector();
diff --git a/src/test/java/com/purbon/kafka/topology/AccessControlWithRBACTest.java b/src/test/java/com/purbon/kafka/topology/AccessControlWithRBACTest.java
index 7ca2bfa..b059b97 100644
--- a/src/test/java/com/purbon/kafka/topology/AccessControlWithRBACTest.java
+++ b/src/test/java/com/purbon/kafka/topology/AccessControlWithRBACTest.java
@@ -48,8 +48,7 @@ public class AccessControlWithRBACTest {
   }
 
   @Test
-  public void testPredefinedRoles() throws IOException {
-
+  public void testPredefinedRoles() {
     Map<String, List<String>> predefinedRoles = new HashMap<>();
     predefinedRoles.put("ResourceOwner", Arrays.asList("User:Foo"));
 
diff --git a/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java b/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java
index cdab754..065f180 100644
--- a/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java
@@ -184,7 +184,7 @@ public class AclsBindingsBuilderTest {
         builder.buildBindingsForStreamsApp(
             stream.getPrincipal(), "prefix", readTopics, writeTopics);
 
-    assertThat(bindings.size()).isEqualTo(3);
+    assertThat(bindings.size()).isEqualTo(4);
     assertThat(bindings)
         .contains(
             buildTopicLevelAcl(
@@ -193,6 +193,10 @@ public class AclsBindingsBuilderTest {
         .contains(
             buildTopicLevelAcl(
                 stream.getPrincipal(), "bar", PatternType.LITERAL, AclOperation.WRITE));
+    assertThat(bindings)
+        .contains(
+            buildGroupLevelAcl(
+                stream.getPrincipal(), "prefix", PatternType.PREFIXED, AclOperation.READ));
     assertThat(bindings)
         .contains(
             buildTopicLevelAcl(
diff --git a/src/test/java/com/purbon/kafka/topology/ClusterLevelRoleBuilderTest.java b/src/test/java/com/purbon/kafka/topology/ClusterLevelRoleBuilderTest.java
index c970e13..f7e37a6 100644
--- a/src/test/java/com/purbon/kafka/topology/ClusterLevelRoleBuilderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/ClusterLevelRoleBuilderTest.java
@@ -12,7 +12,6 @@ import com.purbon.kafka.topology.api.mds.ClusterIDs;
 import com.purbon.kafka.topology.api.mds.MDSApiClient;
 import com.purbon.kafka.topology.model.users.Connector;
 import com.purbon.kafka.topology.roles.rbac.ClusterLevelRoleBuilder;
-import java.io.IOException;
 import java.util.Map;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -100,7 +99,7 @@ public class ClusterLevelRoleBuilderTest {
   }
 
   @Test
-  public void testKafkaConnectRun() throws IOException {
+  public void testKafkaConnectRun() {
     when(apiClient.withClusterIDs()).thenReturn(allClusterIDs);
 
     ClusterLevelRoleBuilder runner = new ClusterLevelRoleBuilder("foo", SECURITY_ADMIN, apiClient);
@@ -112,7 +111,7 @@ public class ClusterLevelRoleBuilderTest {
   }
 
   @Test
-  public void testSchemaRegistryRun() throws IOException {
+  public void testSchemaRegistryRun() {
     when(apiClient.withClusterIDs()).thenReturn(allClusterIDs);
 
     ClusterLevelRoleBuilder runner = new ClusterLevelRoleBuilder("foo", SECURITY_ADMIN, apiClient);
diff --git a/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java b/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java
index 781ca37..b52d924 100644
--- a/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java
+++ b/src/test/java/com/purbon/kafka/topology/ExecutionPlanTest.java
@@ -8,10 +8,9 @@ import com.purbon.kafka.topology.actions.access.ClearBindings;
 import com.purbon.kafka.topology.actions.access.CreateBindings;
 import com.purbon.kafka.topology.roles.SimpleAclsProvider;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
+import com.purbon.kafka.topology.utils.TestUtils;
 import java.io.IOException;
 import java.io.PrintStream;
-import java.nio.file.Files;
-import java.nio.file.Paths;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Set;
@@ -36,7 +35,7 @@ public class ExecutionPlanTest {
 
   @Before
   public void before() throws IOException {
-    Files.deleteIfExists(Paths.get(".cluster-state"));
+    TestUtils.deleteStateFile();
     backendController = new BackendController();
     plan = ExecutionPlan.init(backendController, mockPrintStream);
   }
diff --git a/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java b/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
index 63e9012..20d7b37 100644
--- a/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
@@ -1,22 +1,16 @@
 package com.purbon.kafka.topology;
 
-import static com.purbon.kafka.topology.BuilderCLI.ADMIN_CLIENT_CONFIG_OPTION;
-import static com.purbon.kafka.topology.BuilderCLI.ALLOW_DELETE_OPTION;
-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.BuilderCLI.*;
 import static com.purbon.kafka.topology.TopologyBuilderConfig.CONFLUENT_SCHEMA_REGISTRY_URL_CONFIG;
 import static org.mockito.Matchers.anyObject;
-import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.*;
 
 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.exceptions.TopologyParsingException;
+import com.purbon.kafka.topology.utils.TestUtils;
 import java.io.IOException;
-import java.net.URL;
-import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
@@ -59,8 +53,7 @@ public class KafkaTopologyBuilderTest {
 
   @Test
   public void closeAdminClientTest() throws Exception {
-    URL dirOfDescriptors = getClass().getResource("/descriptor.yaml");
-    String fileOrDirPath = Paths.get(dirOfDescriptors.toURI()).toFile().toString();
+    String fileOrDirPath = TestUtils.getResourceFilename("/descriptor.yaml");
 
     TopologyBuilderConfig builderConfig = new TopologyBuilderConfig(cliOps, props);
 
@@ -77,7 +70,7 @@ public class KafkaTopologyBuilderTest {
     verify(topologyAdminClient, times(1)).close();
   }
 
-  @Test(expected = IOException.class)
+  @Test(expected = TopologyParsingException.class)
   public void verifyProblematicParametersTest() throws Exception {
     String file = "fileThatDoesNotExist.yaml";
     TopologyBuilderConfig builderConfig = new TopologyBuilderConfig(cliOps, props);
@@ -95,8 +88,7 @@ public class KafkaTopologyBuilderTest {
 
   @Test(expected = IOException.class)
   public void verifyProblematicParametersTest2() throws Exception {
-    URL dirOfDescriptors = getClass().getResource("/descriptor.yaml");
-    String fileOrDirPath = Paths.get(dirOfDescriptors.toURI()).toFile().toString();
+    String fileOrDirPath = TestUtils.getResourceFilename("/descriptor.yaml");
 
     TopologyBuilderConfig builderConfig = new TopologyBuilderConfig(cliOps, props);
     KafkaTopologyBuilder builder =
@@ -112,11 +104,8 @@ public class KafkaTopologyBuilderTest {
 
   @Test
   public void verifyProblematicParametersTestOK() throws Exception {
-    URL dirOfDescriptors = getClass().getResource("/descriptor.yaml");
-    String fileOrDirPath = Paths.get(dirOfDescriptors.toURI()).toFile().toString();
-
-    URL clientConfigURL = getClass().getResource("/client-config.properties");
-    String clientConfigFile = Paths.get(clientConfigURL.toURI()).toFile().toString();
+    String fileOrDirPath = TestUtils.getResourceFilename("/descriptor.yaml");
+    String clientConfigFile = TestUtils.getResourceFilename("/client-config.properties");
 
     cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, clientConfigFile);
 
@@ -134,12 +123,8 @@ public class KafkaTopologyBuilderTest {
 
   @Test
   public void builderRunTestAsFromCLI() throws Exception {
-
-    URL dirOfDescriptors = getClass().getResource("/descriptor.yaml");
-    String fileOrDirPath = Paths.get(dirOfDescriptors.toURI()).toFile().toString();
-
-    URL clientConfigURL = getClass().getResource("/client-config.properties");
-    String clientConfigFile = Paths.get(clientConfigURL.toURI()).toFile().toString();
+    String fileOrDirPath = TestUtils.getResourceFilename("/descriptor.yaml");
+    String clientConfigFile = TestUtils.getResourceFilename("/client-config.properties");
 
     Map<String, String> config = new HashMap<>();
     config.put(BROKERS_OPTION, "localhost:9092");
@@ -168,12 +153,8 @@ public class KafkaTopologyBuilderTest {
 
   @Test
   public void builderRunTestAsFromCLIWithARedisBackend() throws Exception {
-
-    URL dirOfDescriptors = getClass().getResource("/descriptor.yaml");
-    String fileOrDirPath = Paths.get(dirOfDescriptors.toURI()).toFile().toString();
-
-    URL clientConfigURL = getClass().getResource("/client-config-redis.properties");
-    String clientConfigFile = Paths.get(clientConfigURL.toURI()).toFile().toString();
+    String fileOrDirPath = TestUtils.getResourceFilename("/descriptor.yaml");
+    String clientConfigFile = TestUtils.getResourceFilename("/client-config-redis.properties");
 
     Map<String, String> config = new HashMap<>();
     config.put(BROKERS_OPTION, "localhost:9092");
@@ -205,8 +186,7 @@ public class KafkaTopologyBuilderTest {
 
   @Test
   public void buiderRunTest() throws Exception {
-    URL dirOfDescriptors = getClass().getResource("/descriptor.yaml");
-    String fileOrDirPath = Paths.get(dirOfDescriptors.toURI()).toFile().toString();
+    String fileOrDirPath = TestUtils.getResourceFilename("/descriptor.yaml");
 
     TopologyBuilderConfig builderConfig = new TopologyBuilderConfig(cliOps, props);
 
diff --git a/src/test/java/com/purbon/kafka/topology/PlanMapSerdesTest.java b/src/test/java/com/purbon/kafka/topology/PlanMapSerdesTest.java
index 2ad2629..c64523e 100644
--- a/src/test/java/com/purbon/kafka/topology/PlanMapSerdesTest.java
+++ b/src/test/java/com/purbon/kafka/topology/PlanMapSerdesTest.java
@@ -6,10 +6,8 @@ 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 com.purbon.kafka.topology.utils.TestUtils;
 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;
@@ -27,7 +25,6 @@ public class PlanMapSerdesTest {
 
   @Test
   public void testSerialization() throws JsonProcessingException {
-
     Map<String, String> config = new HashMap<>();
     config.put("foo", "bar");
     config.put("bar", "3");
@@ -49,9 +46,8 @@ public class PlanMapSerdesTest {
   }
 
   @Test
-  public void testHappyDeserialization() throws URISyntaxException, IOException {
-    URL topologyDescriptor = getClass().getResource("/plans.yaml");
-    PlanMap plans = parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
+  public void testHappyDeserialization() throws IOException {
+    PlanMap plans = parser.deserialise(TestUtils.getResourceFile("/plans.yaml"));
 
     assertThat(plans).has(new Condition<>(list -> list.size() == 2, "Contains two elements"));
     assertThat(plans.get("gold").getAlias()).isEqualTo("gold");
diff --git a/src/test/java/com/purbon/kafka/topology/RbacProviderTest.java b/src/test/java/com/purbon/kafka/topology/RbacProviderTest.java
index 584ca40..c2e37a9 100644
--- a/src/test/java/com/purbon/kafka/topology/RbacProviderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/RbacProviderTest.java
@@ -35,7 +35,6 @@ import com.purbon.kafka.topology.roles.RBACProvider;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import com.purbon.kafka.topology.roles.rbac.ClusterLevelRoleBuilder;
 import com.purbon.kafka.topology.roles.rbac.RBACBindingsBuilder;
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -213,7 +212,7 @@ public class RbacProviderTest {
   }
 
   @Test
-  public void newSchemaRegistryACLCreation() throws IOException {
+  public void newSchemaRegistryACLCreation() {
 
     Project project = new ProjectImpl();
     Topology topology = new TopologyImpl();
@@ -249,7 +248,7 @@ public class RbacProviderTest {
   }
 
   @Test
-  public void newControlCenterACLCreation() throws IOException {
+  public void newControlCenterACLCreation() {
 
     Project project = new ProjectImpl();
     Topology topology = new TopologyImpl();
@@ -277,7 +276,7 @@ public class RbacProviderTest {
   }
 
   @Test
-  public void newKafkaConnectACLsCreation() throws IOException {
+  public void newKafkaConnectACLsCreation() {
 
     Project project = new ProjectImpl();
 
diff --git a/src/test/java/com/purbon/kafka/topology/TopologyBuilderConfigTest.java b/src/test/java/com/purbon/kafka/topology/TopologyBuilderConfigTest.java
index 8b8b133..47db665 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologyBuilderConfigTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologyBuilderConfigTest.java
@@ -2,14 +2,7 @@ package com.purbon.kafka.topology;
 
 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.ACCESS_CONTROL_IMPLEMENTATION_CLASS;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.CONFLUENT_SCHEMA_REGISTRY_URL_CONFIG;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.MDS_KAFKA_CLUSTER_ID_CONFIG;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.MDS_PASSWORD_CONFIG;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.MDS_SERVER;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.MDS_USER_CONFIG;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.PROJECT_PREFIX_FORMAT_CONFIG;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.TOPIC_PREFIX_FORMAT_CONFIG;
+import static com.purbon.kafka.topology.TopologyBuilderConfig.*;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import com.purbon.kafka.topology.exceptions.ConfigurationException;
@@ -20,13 +13,8 @@ import com.purbon.kafka.topology.model.Project;
 import com.purbon.kafka.topology.model.Topic;
 import com.purbon.kafka.topology.model.TopicSchemas;
 import com.purbon.kafka.topology.model.Topology;
-import java.net.URISyntaxException;
-import java.net.URL;
-import java.nio.file.Paths;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Properties;
+import com.purbon.kafka.topology.utils.TestUtils;
+import java.util.*;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -62,7 +50,7 @@ public class TopologyBuilderConfigTest {
     Project project = new ProjectImpl();
     Topic topic = new TopicImpl();
     TopicSchemas schema = new TopicSchemas("foo", "bar");
-    topic.setSchemas(schema);
+    topic.setSchemas(Optional.of(schema));
     project.addTopic(topic);
     topology.addProject(project);
     props.put(CONFLUENT_SCHEMA_REGISTRY_URL_CONFIG, "mock://");
@@ -77,7 +65,7 @@ public class TopologyBuilderConfigTest {
     Project project = new ProjectImpl();
     Topic topic = new TopicImpl();
     TopicSchemas schema = new TopicSchemas("foo", "bar");
-    topic.setSchemas(schema);
+    topic.setSchemas(Optional.of(schema));
     project.addTopic(topic);
     topology.addProject(project);
 
@@ -101,6 +89,18 @@ public class TopologyBuilderConfigTest {
     config.validateWith(topology);
   }
 
+  @Test
+  public void testNoSchemaRegistry() throws ConfigurationException {
+    Topology topology = new TopologyImpl();
+    Project project = new ProjectImpl("project");
+    Topic topic = new TopicImpl("topic", "json");
+    project.addTopic(topic);
+    topology.addProject(project);
+
+    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    config.validateWith(topology);
+  }
+
   @Test
   public void testPrefixValidConfigFields() throws ConfigurationException {
     Topology topology = new TopologyImpl();
@@ -160,9 +160,8 @@ public class TopologyBuilderConfigTest {
   }
 
   @Test
-  public void testKafkaInternalTopicDefaultPrefix() throws URISyntaxException {
-    URL clientConfigURL = getClass().getResource("/client-config.properties");
-    String clientConfigFile = Paths.get(clientConfigURL.toURI()).toFile().toString();
+  public void testKafkaInternalTopicDefaultPrefix() {
+    String clientConfigFile = TestUtils.getResourceFilename("/client-config.properties");
 
     cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, clientConfigFile);
 
@@ -171,9 +170,9 @@ public class TopologyBuilderConfigTest {
   }
 
   @Test
-  public void testKafkaInternalTopicExtendedPrefix() throws URISyntaxException {
-    URL clientConfigURL = getClass().getResource("/config-internals-extended.properties");
-    String clientConfigFile = Paths.get(clientConfigURL.toURI()).toFile().toString();
+  public void testKafkaInternalTopicExtendedPrefix() {
+    String clientConfigFile =
+        TestUtils.getResourceFilename("/config-internals-extended.properties");
 
     cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, clientConfigFile);
 
diff --git a/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java b/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java
index 0d43605..7964bd3 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java
@@ -3,12 +3,11 @@ package com.purbon.kafka.topology;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 
+import com.purbon.kafka.topology.exceptions.TopologyParsingException;
 import com.purbon.kafka.topology.model.Topic;
 import com.purbon.kafka.topology.model.Topology;
+import com.purbon.kafka.topology.utils.TestUtils;
 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;
@@ -18,10 +17,8 @@ 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();
+  public void buildTopicNameTest() throws IOException {
+    String fileOrDirPath = TestUtils.getResourceFilename("/dir");
 
     Topology topology = TopologyObjectBuilder.build(fileOrDirPath);
 
@@ -29,12 +26,9 @@ public class TopologyObjectBuilderTest {
   }
 
   @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();
+  public void testConfigUpdateWhenUsingCustomPlans() throws IOException {
+    String descriptorFile = TestUtils.getResourceFilename("/descriptor-with-plans.yaml");
+    String plansFile = TestUtils.getResourceFilename("/plans.yaml");
 
     Topology topology = TopologyObjectBuilder.build(descriptorFile, plansFile);
     assertThat(topology).isNotNull();
@@ -61,9 +55,20 @@ public class TopologyObjectBuilderTest {
   }
 
   @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();
+  public void testTopologyWithPlansButWithNoPlansDef() throws IOException {
+    String descriptorFile = TestUtils.getResourceFilename("/descriptor-with-plans.yaml");
     TopologyObjectBuilder.build(descriptorFile);
   }
+
+  @Test(expected = TopologyParsingException.class)
+  public void testInvalidTopology() throws IOException {
+    String descriptorFile = TestUtils.getResourceFilename("/errors_dir/descriptor-with-errors.yaml");
+    TopologyObjectBuilder.build(descriptorFile);
+  }
+
+  @Test(expected = TopologyParsingException.class)
+  public void testInvalidTopologyFromDir() throws IOException {
+    String dirPath = TestUtils.getResourceFilename("/errors_dir");
+    TopologyObjectBuilder.build(dirPath);
+  }
 }
diff --git a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
index b47e3bc..87dfa4e 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
@@ -8,6 +8,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
+import com.purbon.kafka.topology.exceptions.TopologyParsingException;
 import com.purbon.kafka.topology.model.Impl.ProjectImpl;
 import com.purbon.kafka.topology.model.Impl.TopicImpl;
 import com.purbon.kafka.topology.model.Impl.TopologyImpl;
@@ -23,19 +24,9 @@ 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 com.purbon.kafka.topology.utils.TestUtils;
 import java.io.IOException;
-import java.net.URISyntaxException;
-import java.net.URL;
-import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-import java.util.Set;
+import java.util.*;
 import java.util.stream.Collectors;
 import org.junit.Before;
 import org.junit.Test;
@@ -50,18 +41,13 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testDynamicFirstLevelAttributes() throws IOException, URISyntaxException {
-
-    URL descriptorWithOptionals = getClass().getResource("/descriptor-with-others.yml");
-
-    Topology topology = parser.deserialise(Paths.get(descriptorWithOptionals.toURI()).toFile());
+  public void testDynamicFirstLevelAttributes() {
+    Topology topology =
+        parser.deserialise(TestUtils.getResourceFile("/descriptor-with-others.yml"));
     Project project = topology.getProjects().get(0);
     assertThat(project.namePrefix()).startsWith("contextOrg.source.foo.bar.zet");
 
-    URL descriptorWithoutOptionals = getClass().getResource("/descriptor.yaml");
-
-    Topology anotherTopology =
-        parser.deserialise(Paths.get(descriptorWithoutOptionals.toURI()).toFile());
+    Topology anotherTopology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
     Project anotherProject = anotherTopology.getProjects().get(0);
 
     assertEquals("contextOrg.source.foo", anotherProject.namePrefix());
@@ -69,7 +55,6 @@ public class TopologySerdesTest {
 
   @Test
   public void testTopologySerialisation() throws IOException {
-
     Topology topology = new TopologyImpl();
     topology.setContext("contextOrg");
     topology.setProjects(buildProjects());
@@ -83,7 +68,6 @@ public class TopologySerdesTest {
 
   @Test
   public void testTopicConfigSerdes() throws IOException {
-
     Topology topology = new TopologyImpl();
     topology.setContext("team");
 
@@ -143,7 +127,6 @@ public class TopologySerdesTest {
 
   @Test
   public void testTopicWithDataType() throws IOException {
-
     Project project = new ProjectImpl("foo");
 
     Topology topology = new TopologyImpl();
@@ -172,23 +155,19 @@ public class TopologySerdesTest {
     assertEquals(topic2.getDataType(), serdesTopic2.getDataType());
   }
 
-  @Test(expected = IOException.class)
-  public void testTopologyWithNoTeam() throws IOException, URISyntaxException {
-    URL topologyDescriptor = getClass().getResource("/descriptor-with-no-context.yaml");
-    parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
+  @Test(expected = TopologyParsingException.class)
+  public void testTopologyWithNoTeam() {
+    parser.deserialise(TestUtils.getResourceFile("/descriptor-with-no-context.yaml"));
   }
 
-  @Test(expected = IOException.class)
-  public void testTopologyWithNoProject() throws IOException, URISyntaxException {
-    URL topologyDescriptor = getClass().getResource("/descriptor-with-no-project.yaml");
-    parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
+  @Test(expected = TopologyParsingException.class)
+  public void testTopologyWithNoProject() {
+    parser.deserialise(TestUtils.getResourceFile("/descriptor-with-no-project.yaml"));
   }
 
   @Test
-  public void testCoreElementsProcessing() throws IOException, URISyntaxException {
-
-    URL topologyDescriptor = getClass().getResource("/descriptor.yaml");
-    Topology topology = parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
+  public void testCoreElementsProcessing() {
+    Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
 
     Project project = topology.getProjects().get(0);
     assertThat(project.getProducers()).hasSize(3);
@@ -201,18 +180,29 @@ public class TopologySerdesTest {
     assertThat(project.getProducers().get(2).getIdempotence()).isNotEmpty();
   }
 
+  @Test
+  public void testStreamsApps() {
+    Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
+
+    Project project1 = topology.getProjects().get(0);
+    assertThat(project1.getStreams()).hasSize(1);
+    assertThat(project1.getStreams()).noneMatch(s -> s.getApplicationId().isPresent());
+
+    Project project3 = topology.getProjects().get(2);
+    assertThat(project3.getStreams()).hasSize(1);
+    assertThat(project3.getStreams())
+        .allMatch(s -> s.getApplicationId().orElse("notFound").equals("applicationId-1"));
+  }
+
   @Test(expected = IllegalArgumentException.class)
-  public void testSchemaSerdes() throws URISyntaxException, IOException {
-    URL topologyDescriptor = getClass().getResource("/descriptor-wrong-schemas.yaml");
-    parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
+  public void testSchemaSerdes() {
+    parser.deserialise(TestUtils.getResourceFile("/descriptor-wrong-schemas.yaml"));
   }
 
   @Test
-  public void testPlaformProcessing() throws IOException, URISyntaxException {
-
-    URL topologyDescriptor = getClass().getResource("/descriptor.yaml");
+  public void testPlaformProcessing() {
+    Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
 
-    Topology topology = parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
     assertEquals("contextOrg", topology.getContext());
 
     List<SchemaRegistryInstance> listOfSR =
@@ -231,10 +221,9 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testOnlyTopics() throws URISyntaxException, IOException {
-
-    URL topologyDescriptor = getClass().getResource("/descriptor-only-topics.yaml");
-    Topology topology = parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
+  public void testOnlyTopics() {
+    Topology topology =
+        parser.deserialise(TestUtils.getResourceFile("/descriptor-only-topics.yaml"));
 
     assertEquals("contextOrg", topology.getContext());
     assertTrue(topology.getProjects().get(0).getConnectors().isEmpty());
@@ -244,10 +233,9 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testRBACTopics() throws URISyntaxException, IOException {
-
-    URL topologyDescriptor = getClass().getResource("/descriptor-with-rbac-topics.yaml");
-    Topology topology = parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
+  public void testRBACTopics() {
+    Topology topology =
+        parser.deserialise(TestUtils.getResourceFile("/descriptor-with-rbac-topics.yaml"));
 
     Project project = topology.getProjects().get(0);
     assertEquals("contextOrg", topology.getContext());
@@ -268,10 +256,8 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testWithRBACDescriptor() throws IOException, URISyntaxException {
-    URL descriptor = getClass().getResource("/descriptor-with-rbac.yaml");
-
-    Topology topology = parser.deserialise(Paths.get(descriptor.toURI()).toFile());
+  public void testWithRBACDescriptor() {
+    Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor-with-rbac.yaml"));
     Project myProject = topology.getProjects().get(0);
 
     assertEquals(2, myProject.getRbacRawRoles().size());
@@ -303,8 +289,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testTopicNameWithCustomSeparator() throws URISyntaxException, IOException {
-
+  public void testTopicNameWithCustomSeparator() {
     Map<String, String> cliOps = new HashMap<>();
     cliOps.put(BROKERS_OPTION, "");
     cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, "/fooBar");
@@ -315,8 +300,8 @@ public class TopologySerdesTest {
 
     TopologySerdes parser = new TopologySerdes(config);
 
-    URL topologyDescriptor = getClass().getResource("/descriptor-only-topics.yaml");
-    Topology topology = parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
+    Topology topology =
+        parser.deserialise(TestUtils.getResourceFile("/descriptor-only-topics.yaml"));
 
     assertEquals("contextOrg", topology.getContext());
 
@@ -328,8 +313,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testTopicNameWithCustomPattern() throws URISyntaxException, IOException {
-
+  public void testTopicNameWithCustomPattern() {
     Map<String, String> cliOps = new HashMap<>();
     cliOps.put(BROKERS_OPTION, "");
     cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, "/fooBar");
@@ -340,8 +324,8 @@ public class TopologySerdesTest {
 
     TopologySerdes parser = new TopologySerdes(config);
 
-    URL topologyDescriptor = getClass().getResource("/descriptor-only-topics.yaml");
-    Topology topology = parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
+    Topology topology =
+        parser.deserialise(TestUtils.getResourceFile("/descriptor-only-topics.yaml"));
 
     assertEquals("contextOrg", topology.getContext());
 
@@ -353,11 +337,9 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testJsonDescriptorFileSerdes() throws IOException, URISyntaxException {
-
+  public void testJsonDescriptorFileSerdes() {
     TopologySerdes parser = new TopologySerdes(new TopologyBuilderConfig(), FileType.JSON);
-    URL descriptorWithOptionals = getClass().getResource("/descriptor.json");
-    Topology topology = parser.deserialise(Paths.get(descriptorWithOptionals.toURI()).toFile());
+    Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.json"));
 
     assertEquals(1, topology.getProjects().size());
     assertEquals("foo", topology.getProjects().get(0).getName());
@@ -365,7 +347,6 @@ public class TopologySerdesTest {
   }
 
   private List<Project> buildProjects() {
-
     Project project = new ProjectImpl("project");
     project.setConsumers(buildConsumers());
     project.setProducers(buildProducers());
@@ -375,13 +356,13 @@ public class TopologySerdesTest {
   }
 
   private List<KStream> buildStreams() {
-    List<KStream> streams = new ArrayList<KStream>();
-    HashMap<String, List<String>> topics = new HashMap<String, List<String>>();
+    List<KStream> streams = new ArrayList<>();
+    HashMap<String, List<String>> topics = new HashMap<>();
     topics.put("read", Arrays.asList("topic1", "topic3"));
     topics.put("write", Arrays.asList("topic2", "topic4"));
     streams.add(new KStream("app3", topics));
 
-    topics = new HashMap<String, List<String>>();
+    topics = new HashMap<>();
     topics.put("read", Arrays.asList("topic2", "topic4"));
     topics.put("write", Arrays.asList("topic5"));
     streams.add(new KStream("app4", topics));
@@ -390,14 +371,14 @@ public class TopologySerdesTest {
   }
 
   private List<Producer> buildProducers() {
-    List<Producer> producers = new ArrayList<Producer>();
+    List<Producer> producers = new ArrayList<>();
     producers.add(new Producer("app1"));
     producers.add(new Producer("app3"));
     return producers;
   }
 
   private List<Consumer> buildConsumers() {
-    List<Consumer> consumers = new ArrayList<Consumer>();
+    List<Consumer> consumers = new ArrayList<>();
     consumers.add(new Consumer("app1"));
     consumers.add(new Consumer("app2"));
     return consumers;
diff --git a/src/test/java/com/purbon/kafka/topology/TopologyValidationTest.java b/src/test/java/com/purbon/kafka/topology/TopologyValidationTest.java
index fe406a7..b1d16b6 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologyValidationTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologyValidationTest.java
@@ -7,15 +7,9 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 import com.purbon.kafka.topology.model.Topology;
 import com.purbon.kafka.topology.serdes.TopologySerdes;
+import com.purbon.kafka.topology.utils.TestUtils;
 import java.io.IOException;
-import java.net.URISyntaxException;
-import java.net.URL;
-import java.nio.file.Paths;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
+import java.util.*;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -29,10 +23,10 @@ public class TopologyValidationTest {
   }
 
   @Test
-  public void testPositiveExecutionOnCamelCaseNames() throws IOException, URISyntaxException {
+  public void testPositiveExecutionOnCamelCaseNames() throws IOException {
 
-    URL descriptorWithOptionals = getClass().getResource("/descriptor-with-camelCaseNames.yml");
-    Topology topology = parser.deserialise(Paths.get(descriptorWithOptionals.toURI()).toFile());
+    Topology topology =
+        parser.deserialise(TestUtils.getResourceFile("/descriptor-with-camelCaseNames.yml"));
 
     Map<String, String> cliOps = new HashMap<>();
     cliOps.put(BROKERS_OPTION, "");
@@ -48,11 +42,10 @@ public class TopologyValidationTest {
   }
 
   @Test
-  public void testInvalidExecutionBecuaseofNumberOfPartitions()
-      throws IOException, URISyntaxException {
+  public void testInvalidExecutionBecuaseofNumberOfPartitions() throws IOException {
 
-    URL descriptorWithOptionals = getClass().getResource("/descriptor-with-camelCaseNames.yml");
-    Topology topology = parser.deserialise(Paths.get(descriptorWithOptionals.toURI()).toFile());
+    Topology topology =
+        parser.deserialise(TestUtils.getResourceFile("/descriptor-with-camelCaseNames.yml"));
 
     Map<String, String> cliOps = new HashMap<>();
     cliOps.put(BROKERS_OPTION, "");
@@ -70,10 +63,9 @@ public class TopologyValidationTest {
   }
 
   @Test
-  public void testInvalidExecutionWithFailedValidation() throws IOException, URISyntaxException {
+  public void testInvalidExecutionWithFailedValidation() throws IOException {
 
-    URL descriptorWithOptionals = getClass().getResource("/descriptor.yaml");
-    Topology topology = parser.deserialise(Paths.get(descriptorWithOptionals.toURI()).toFile());
+    Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
 
     Map<String, String> cliOps = new HashMap<>();
     cliOps.put(BROKERS_OPTION, "");
@@ -87,11 +79,15 @@ public class TopologyValidationTest {
 
     TopologyValidator validator = new TopologyValidator(config);
     List<String> results = validator.validate(topology);
-    assertThat(results).hasSize(3);
+    assertThat(results).hasSize(5);
     assertThat(results.get(0)).isEqualTo("Project name does not follow the camelCase format: foo");
     assertThat(results.get(1))
         .isEqualTo("Topic contextOrg.source.foo.foo has an invalid number of partitions: 1");
     assertThat(results.get(2))
         .isEqualTo("Topic contextOrg.source.bar.bar.avro has an invalid number of partitions: 1");
+    assertThat(results.get(3))
+        .isEqualTo("Topic contextOrg.source.baz.topicE has an invalid number of partitions: 1");
+    assertThat(results.get(4))
+        .isEqualTo("Topic contextOrg.source.baz.topicF has an invalid number of partitions: 1");
   }
 }
diff --git a/src/test/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreamsTest.java b/src/test/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreamsTest.java
new file mode 100644
index 0000000..1d83a4f
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreamsTest.java
@@ -0,0 +1,69 @@
+package com.purbon.kafka.topology.actions.access.builders;
+
+import static java.util.Collections.singletonList;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.purbon.kafka.topology.TopologyBuilderConfig;
+import com.purbon.kafka.topology.model.users.KStream;
+import com.purbon.kafka.topology.roles.acls.AclsBindingsBuilder;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Optional;
+import org.apache.kafka.common.resource.ResourceType;
+import org.junit.Before;
+import org.junit.Test;
+
+public class BuildBindingsForKStreamsTest {
+
+  private BuildBindingsForKStreams action;
+  private AclsBindingsBuilder aclsBindingsBuilder;
+
+  @Before
+  public void setUp() {
+    aclsBindingsBuilder = new AclsBindingsBuilder(new TopologyBuilderConfig());
+  }
+
+  @Test
+  public void testStreamsWithoutApplicationId() {
+    HashMap<String, List<String>> topics = new HashMap<>();
+    topics.put(KStream.READ_TOPICS, singletonList("topicA"));
+    topics.put(KStream.WRITE_TOPICS, singletonList("topicB"));
+
+    KStream app = new KStream("User:user", topics);
+    String topicPrefix = "topicPrefix";
+    action = new BuildBindingsForKStreams(aclsBindingsBuilder, app, topicPrefix);
+    action.execute();
+    assertThat(action.getBindings())
+        .anyMatch(
+            b ->
+                b.getResourceType() == ResourceType.TOPIC
+                    && b.getResourceName().equals(topicPrefix));
+    assertThat(action.getBindings())
+        .anyMatch(
+            b ->
+                b.getResourceType() == ResourceType.GROUP
+                    && b.getResourceName().equals(topicPrefix));
+  }
+
+  @Test
+  public void testStreamWithApplicationId() {
+    HashMap<String, List<String>> topics = new HashMap<>();
+    topics.put(KStream.READ_TOPICS, singletonList("topicA"));
+    topics.put(KStream.WRITE_TOPICS, singletonList("topicB"));
+
+    String applicationId = "applicationId";
+    KStream app = new KStream("User:user", topics, Optional.of(applicationId));
+    action = new BuildBindingsForKStreams(aclsBindingsBuilder, app, "topicPrefix");
+    action.execute();
+    assertThat(action.getBindings())
+        .anyMatch(
+            b ->
+                b.getResourceType() == ResourceType.TOPIC
+                    && b.getResourceName().equals(applicationId));
+    assertThat(action.getBindings())
+        .anyMatch(
+            b ->
+                b.getResourceType() == ResourceType.GROUP
+                    && b.getResourceName().equals(applicationId));
+  }
+}
diff --git a/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java b/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
index 96f5360..9eaf672 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
@@ -8,6 +8,8 @@ import com.purbon.kafka.topology.BackendController;
 import com.purbon.kafka.topology.ExecutionPlan;
 import com.purbon.kafka.topology.TopologyBuilderConfig;
 import com.purbon.kafka.topology.api.adminclient.TopologyBuilderAdminClient;
+import com.purbon.kafka.topology.integration.containerutils.ContainerTestUtils;
+import com.purbon.kafka.topology.integration.containerutils.SaslPlaintextKafkaContainer;
 import com.purbon.kafka.topology.model.Impl.ProjectImpl;
 import com.purbon.kafka.topology.model.Impl.TopicImpl;
 import com.purbon.kafka.topology.model.Impl.TopologyImpl;
@@ -26,40 +28,24 @@ import com.purbon.kafka.topology.model.users.platform.SchemaRegistryInstance;
 import com.purbon.kafka.topology.roles.SimpleAclsProvider;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import com.purbon.kafka.topology.roles.acls.AclsBindingsBuilder;
+import com.purbon.kafka.topology.utils.TestUtils;
 import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Optional;
-import java.util.Properties;
+import java.util.*;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.AdminClientConfig;
-import org.apache.kafka.common.acl.AccessControlEntryFilter;
-import org.apache.kafka.common.acl.AclBinding;
-import org.apache.kafka.common.acl.AclBindingFilter;
-import org.apache.kafka.common.acl.AclOperation;
-import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.acl.*;
 import org.apache.kafka.common.resource.PatternType;
 import org.apache.kafka.common.resource.ResourcePatternFilter;
 import org.apache.kafka.common.resource.ResourceType;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
+import org.junit.*;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnit;
 import org.mockito.junit.MockitoRule;
 
 public class AccessControlManagerIT {
 
+  private static SaslPlaintextKafkaContainer container;
   private static AdminClient kafkaAdminClient;
   private AccessControlManager accessControlManager;
   private SimpleAclsProvider aclsProvider;
@@ -71,12 +57,23 @@ public class AccessControlManagerIT {
 
   @Mock private TopologyBuilderConfig config;
 
+  @BeforeClass
+  public static void setup() {
+    container = new SaslPlaintextKafkaContainer();
+    container.start();
+  }
+
+  @AfterClass
+  public static void teardown() {
+    container.stop();
+  }
+
   @Before
   public void before() throws IOException {
-    kafkaAdminClient = AdminClient.create(config());
+    kafkaAdminClient = ContainerTestUtils.getSaslAdminClient(container);
     TopologyBuilderAdminClient adminClient = new TopologyBuilderAdminClient(kafkaAdminClient);
     adminClient.clearAcls();
-    Files.deleteIfExists(Paths.get(".cluster-state"));
+    TestUtils.deleteStateFile();
 
     this.cs = new BackendController();
     this.plan = ExecutionPlan.init(cs, System.out);
@@ -473,7 +470,7 @@ public class AccessControlManagerIT {
     acls = kafkaAdminClient.describeAcls(filter).values().get();
 
     // two acls created for the read topics
-    assertEquals(2, acls.size());
+    assertEquals(3, acls.size());
 
     entryFilter =
         new AccessControlEntryFilter(
@@ -549,19 +546,4 @@ public class AccessControlManagerIT {
       Assert.assertTrue(ops.contains(AclOperation.READ));
     }
   }
-
-  private Properties config() {
-    Properties props = new Properties();
-
-    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
-    props.put(AdminClientConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
-    props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
-    props.put("sasl.mechanism", "PLAIN");
-
-    props.put(
-        "sasl.jaas.config",
-        "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"kafka\" password=\"kafka\";");
-
-    return props;
-  }
 }
diff --git a/src/test/java/com/purbon/kafka/topology/integration/AclProducerAndConsumerIT.java b/src/test/java/com/purbon/kafka/topology/integration/AclProducerAndConsumerIT.java
new file mode 100644
index 0000000..b4bee59
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/integration/AclProducerAndConsumerIT.java
@@ -0,0 +1,138 @@
+package com.purbon.kafka.topology.integration;
+
+import com.purbon.kafka.topology.integration.containerutils.ContainerTestUtils;
+import com.purbon.kafka.topology.integration.containerutils.SaslPlaintextKafkaContainer;
+import com.purbon.kafka.topology.integration.containerutils.TestConsumer;
+import com.purbon.kafka.topology.integration.containerutils.TestProducer;
+import java.util.Set;
+import org.apache.kafka.common.errors.GroupAuthorizationException;
+import org.apache.kafka.common.errors.SaslAuthenticationException;
+import org.apache.kafka.common.errors.TopicAuthorizationException;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public final class AclProducerAndConsumerIT {
+
+  private static final String TOPIC = "producer-and-consumer-test-topic";
+  private static final String OTHER_TOPIC = "other-" + TOPIC;
+  private static final String CONSUMER_GROUP = "producer-and-consumer-test-consumer-group";
+  private static final String UNKNOWN_USERNAME = "unknown-user";
+  private static final String NO_ACCESS_USERNAME = "no-access-user";
+  private static final String PRODUCER_USERNAME = "producer";
+  private static final String CONSUMER_USERNAME = "consumer";
+  private static final String OTHER_PRODUCER_USERNAME = "other-producer";
+  private static final String OTHER_CONSUMER_USERNAME = "other-consumer";
+  private static SaslPlaintextKafkaContainer container;
+
+  @BeforeClass
+  public static void beforeClass() {
+    container =
+        new SaslPlaintextKafkaContainer()
+            .withUser(NO_ACCESS_USERNAME)
+            .withUser(PRODUCER_USERNAME)
+            .withUser(CONSUMER_USERNAME)
+            .withUser(OTHER_PRODUCER_USERNAME)
+            .withUser(OTHER_CONSUMER_USERNAME);
+    container.start();
+    ContainerTestUtils.populateAcls(
+        container,
+        "/acl-producer-and-consumer-it.yaml",
+        "/acl-producer-and-consumer-it.properties");
+  }
+
+  @AfterClass
+  public static void afterClass() {
+    container.stop();
+  }
+
+  @Test(expected = SaslAuthenticationException.class)
+  public void shouldNotProduceWhenUnknownUser() {
+    try (final TestProducer producer = TestProducer.create(container, UNKNOWN_USERNAME)) {
+      producer.produce(TOPIC, "foo");
+    }
+  }
+
+  @Test(expected = SaslAuthenticationException.class)
+  public void shouldNotConsumeWhenUnknownUser() {
+    try (final TestConsumer consumer =
+        TestConsumer.create(container, UNKNOWN_USERNAME, CONSUMER_GROUP)) {
+      consumer.consumeForAWhile(TOPIC, null);
+    }
+  }
+
+  @Test(expected = TopicAuthorizationException.class)
+  public void shouldNotProduceWithoutPermission() {
+    try (final TestProducer producer = TestProducer.create(container, NO_ACCESS_USERNAME)) {
+      producer.produce(TOPIC, "foo");
+    }
+  }
+
+  @Test(expected = TopicAuthorizationException.class)
+  public void shouldNotConsumeWithoutPermission() {
+    try (final TestConsumer consumer =
+        TestConsumer.create(container, NO_ACCESS_USERNAME, CONSUMER_GROUP)) {
+      consumer.consumeForAWhile(TOPIC, null);
+    }
+  }
+
+  @Test(expected = TopicAuthorizationException.class)
+  public void shouldNotProduceWithoutPermissionEvenIfPermittedElsewhere() {
+    try (final TestProducer producer = TestProducer.create(container, OTHER_PRODUCER_USERNAME)) {
+      producer.produce(TOPIC, "foo");
+    }
+  }
+
+  @Test(expected = TopicAuthorizationException.class)
+  public void shouldNotConsumeWithoutPermissionEvenIfPermittedElsewhere() {
+    try (final TestConsumer consumer =
+        TestConsumer.create(container, OTHER_CONSUMER_USERNAME, CONSUMER_GROUP)) {
+      consumer.consumeForAWhile(TOPIC, null);
+    }
+  }
+
+  @Test(expected = TopicAuthorizationException.class)
+  public void shouldNotProduceWhenConsumer() {
+    try (final TestProducer producer = TestProducer.create(container, CONSUMER_USERNAME)) {
+      producer.produce(TOPIC, "foo");
+    }
+  }
+
+  @Test(expected = GroupAuthorizationException.class)
+  public void shouldNotConsumeWhenProducer() {
+    try (final TestConsumer consumer =
+        TestConsumer.create(container, PRODUCER_USERNAME, CONSUMER_GROUP)) {
+      consumer.consumeForAWhile(TOPIC, null);
+    }
+  }
+
+  @Test
+  public void shouldProduceAndConsume() {
+    produceAndConsume(TOPIC, PRODUCER_USERNAME, CONSUMER_USERNAME);
+  }
+
+  @Test
+  public void shouldProduceAndConsumeElsewhere() {
+    /* Just to test that everything was spelled correctly in source and config for the above tests. */
+    produceAndConsume(OTHER_TOPIC, OTHER_PRODUCER_USERNAME, OTHER_CONSUMER_USERNAME);
+  }
+
+  private void produceAndConsume(
+      final String topicName, final String producerUsername, final String consumerUsername) {
+    try (final TestProducer producer = TestProducer.create(container, producerUsername);
+        final TestConsumer consumer =
+            TestConsumer.create(container, consumerUsername, CONSUMER_GROUP)) {
+      final Set<String> values = producer.produceSomeStrings(topicName);
+      consumer.consumeForAWhile(
+          topicName,
+          (key, value) -> {
+            values.remove(value);
+            return !values.isEmpty();
+          });
+      if (!values.isEmpty()) {
+        Assert.fail("Unable to consume all messages.");
+      }
+    }
+  }
+}
diff --git a/src/test/java/com/purbon/kafka/topology/integration/KafkaTopologyBuilderIT.java b/src/test/java/com/purbon/kafka/topology/integration/KafkaTopologyBuilderIT.java
index 1c3c140..7868931 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/KafkaTopologyBuilderIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/KafkaTopologyBuilderIT.java
@@ -1,32 +1,40 @@
 package com.purbon.kafka.topology.integration;
 
-import static com.purbon.kafka.topology.BuilderCLI.ADMIN_CLIENT_CONFIG_OPTION;
-import static com.purbon.kafka.topology.BuilderCLI.ALLOW_DELETE_OPTION;
-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.BuilderCLI.*;
 
 import com.purbon.kafka.topology.KafkaTopologyBuilder;
+import com.purbon.kafka.topology.integration.containerutils.SaslPlaintextKafkaContainer;
+import com.purbon.kafka.topology.utils.TestUtils;
 import java.io.IOException;
-import java.net.URL;
-import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.Map;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 public class KafkaTopologyBuilderIT {
 
+  private static SaslPlaintextKafkaContainer container;
+
+  @BeforeClass
+  public static void setup() {
+    container = new SaslPlaintextKafkaContainer();
+    container.start();
+  }
+
+  @AfterClass
+  public static void teardown() {
+    container.stop();
+  }
+
   @Test(expected = IOException.class)
   public void testSetupKafkaTopologyBuilderWithWrongCredentialsHC() throws Exception {
 
-    URL dirOfDescriptors = getClass().getResource("/descriptor.yaml");
-    String fileOrDirPath = Paths.get(dirOfDescriptors.toURI()).toFile().toString();
-
-    URL clientConfigURL = getClass().getResource("/wrong-client-config.properties");
-    String clientConfigFile = Paths.get(clientConfigURL.toURI()).toFile().toString();
+    String fileOrDirPath = TestUtils.getResourceFilename("/descriptor.yaml");
+    String clientConfigFile = TestUtils.getResourceFilename("/wrong-client-config.properties");
 
     Map<String, String> config = new HashMap<>();
-    config.put(BROKERS_OPTION, "localhost:9092");
+    config.put(BROKERS_OPTION, container.getBootstrapServers());
     config.put(ALLOW_DELETE_OPTION, "false");
     config.put(DRY_RUN_OPTION, "false");
     config.put(QUIET_OPTION, "true");
@@ -38,14 +46,11 @@ public class KafkaTopologyBuilderIT {
   @Test
   public void testSetupKafkaTopologyBuilderWithGoodCredentialsHC() throws Exception {
 
-    URL dirOfDescriptors = getClass().getResource("/descriptor.yaml");
-    String fileOrDirPath = Paths.get(dirOfDescriptors.toURI()).toFile().toString();
-
-    URL clientConfigURL = getClass().getResource("/client-config.properties");
-    String clientConfigFile = Paths.get(clientConfigURL.toURI()).toFile().toString();
+    String fileOrDirPath = TestUtils.getResourceFilename("/descriptor.yaml");
+    String clientConfigFile = TestUtils.getResourceFilename("/client-config.properties");
 
     Map<String, String> config = new HashMap<>();
-    config.put(BROKERS_OPTION, "localhost:9092");
+    config.put(BROKERS_OPTION, container.getBootstrapServers());
     config.put(ALLOW_DELETE_OPTION, "false");
     config.put(DRY_RUN_OPTION, "false");
     config.put(QUIET_OPTION, "true");
diff --git a/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java b/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java
index 9976664..0a72cf6 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java
@@ -7,6 +7,8 @@ import com.purbon.kafka.topology.BackendController;
 import com.purbon.kafka.topology.ExecutionPlan;
 import com.purbon.kafka.topology.TopicManager;
 import com.purbon.kafka.topology.api.adminclient.TopologyBuilderAdminClient;
+import com.purbon.kafka.topology.integration.containerutils.ContainerTestUtils;
+import com.purbon.kafka.topology.integration.containerutils.SaslPlaintextKafkaContainer;
 import com.purbon.kafka.topology.model.Impl.ProjectImpl;
 import com.purbon.kafka.topology.model.Impl.TopicImpl;
 import com.purbon.kafka.topology.model.Impl.TopologyImpl;
@@ -24,12 +26,10 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.common.config.ConfigResource;
@@ -42,12 +42,10 @@ import org.junit.Rule;
 import org.junit.Test;
 import org.mockito.junit.MockitoJUnit;
 import org.mockito.junit.MockitoRule;
-import org.testcontainers.containers.KafkaContainer;
-import org.testcontainers.utility.TestcontainersConfiguration;
 
 public class TopicManagerIT {
 
-  private static KafkaContainer container;
+  private static SaslPlaintextKafkaContainer container;
   private TopicManager topicManager;
   private AdminClient kafkaAdminClient;
 
@@ -57,9 +55,7 @@ public class TopicManagerIT {
 
   @BeforeClass
   public static void setup() {
-    container =
-        new KafkaContainer(
-            TestcontainersConfiguration.getInstance().getKafkaDockerImageName().withTag("5.5.0"));
+    container = new SaslPlaintextKafkaContainer();
     container.start();
   }
 
@@ -70,7 +66,7 @@ public class TopicManagerIT {
 
   @Before
   public void before() throws IOException {
-    kafkaAdminClient = AdminClient.create(config());
+    kafkaAdminClient = ContainerTestUtils.getSaslAdminClient(container);
     TopologyBuilderAdminClient adminClient = new TopologyBuilderAdminClient(kafkaAdminClient);
 
     final SchemaRegistryClient schemaRegistryClient = new MockSchemaRegistryClient();
@@ -341,11 +337,4 @@ public class TopicManagerIT {
 
     assertTrue("Internal topics not found", isInternal);
   }
-
-  private Properties config() {
-    Properties props = new Properties();
-    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, container.getBootstrapServers());
-    props.put(AdminClientConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
-    return props;
-  }
 }
diff --git a/src/test/java/com/purbon/kafka/topology/integration/containerutils/AlternativeKafkaContainer.java b/src/test/java/com/purbon/kafka/topology/integration/containerutils/AlternativeKafkaContainer.java
new file mode 100644
index 0000000..bfcdd1d
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/integration/containerutils/AlternativeKafkaContainer.java
@@ -0,0 +1,170 @@
+package com.purbon.kafka.topology.integration.containerutils;
+
+import com.github.dockerjava.api.command.ExecCreateCmdResponse;
+import com.github.dockerjava.api.command.InspectContainerResponse;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.TimeUnit;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.images.builder.Transferable;
+import org.testcontainers.utility.DockerImageName;
+
+/**
+ * An alternative KafkaContainer that is easier to extend than the one from the testcontainers
+ * project version 1.15.0.
+ */
+public class AlternativeKafkaContainer extends GenericContainer<AlternativeKafkaContainer> {
+
+  private static final DockerImageName DEFAULT_IMAGE =
+      DockerImageName.parse("confluentinc/cp-kafka").withTag("5.5.0");
+  private static final String STARTER_SCRIPT = "/testcontainers_start.sh";
+  public static final String INTERNAL_LISTENER_NAME = "BROKER";
+  public static final int KAFKA_PORT = 9092;
+  public static final int KAFKA_INTERNAL_PORT = 9093;
+  public static final int ZOOKEEPER_PORT = 2181;
+  /* Note difference between 0.0.0.0 and localhost: The former will be replaced by the container IP. */
+  private static final String LISTENERS =
+      "PLAINTEXT://0.0.0.0:"
+          + KAFKA_PORT
+          + ","
+          + INTERNAL_LISTENER_NAME
+          + "://127.0.0.1:"
+          + KAFKA_INTERNAL_PORT;
+  private static final int PORT_NOT_ASSIGNED = -1;
+  private int port = PORT_NOT_ASSIGNED;
+  protected String externalZookeeperConnect = null;
+
+  public AlternativeKafkaContainer() {
+    this(DEFAULT_IMAGE);
+  }
+
+  public AlternativeKafkaContainer(final DockerImageName dockerImageName) {
+    super(dockerImageName);
+    withExposedPorts(KAFKA_PORT);
+    withEnv("KAFKA_LISTENERS", LISTENERS);
+    withEnv(
+        "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
+        "PLAINTEXT:PLAINTEXT," + INTERNAL_LISTENER_NAME + ":PLAINTEXT");
+    withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", INTERNAL_LISTENER_NAME);
+    withEnv("KAFKA_BROKER_ID", "1");
+    withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1");
+    withEnv("KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS", "1");
+    withEnv("KAFKA_LOG_FLUSH_INTERVAL_MESSAGES", Long.MAX_VALUE + "");
+    withEnv("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0");
+  }
+
+  public final AlternativeKafkaContainer withEmbeddedZookeeper() {
+    externalZookeeperConnect = null;
+    return self();
+  }
+
+  public final AlternativeKafkaContainer withExternalZookeeper(final String connectString) {
+    externalZookeeperConnect = connectString;
+    return self();
+  }
+
+  public final String getBootstrapServers() {
+    if (port == PORT_NOT_ASSIGNED) {
+      throw new IllegalStateException("You should start Kafka container first");
+    }
+    return overrideBootstrapServers(String.format("%s:%s", getHost(), port));
+  }
+
+  /** Subclasses may override. */
+  protected String overrideBootstrapServers(final String bootstrapServers) {
+    return bootstrapServers;
+  }
+
+  @Override
+  protected final void doStart() {
+    withCommand(
+        "sh", "-c", "while [ ! -f " + STARTER_SCRIPT + " ]; do sleep 0.1; done; " + STARTER_SCRIPT);
+    if (externalZookeeperConnect == null) {
+      addExposedPort(ZOOKEEPER_PORT);
+    }
+    beforeStart();
+    super.doStart();
+  }
+
+  /** Subclasses may override. */
+  protected void beforeStart() {}
+
+  @Override
+  protected final void containerIsStarting(
+      final InspectContainerResponse containerInfo, final boolean reused) {
+    super.containerIsStarting(containerInfo, reused);
+    port = getMappedPort(KAFKA_PORT);
+    if (reused) {
+      return;
+    }
+    beforeStartupPreparations();
+    final String zookeeperConnect =
+        externalZookeeperConnect != null ? externalZookeeperConnect : startZookeeper();
+    createStartupScript(zookeeperConnect);
+  }
+
+  /** Subclasses may override. */
+  protected void beforeStartupPreparations() {}
+
+  private void createStartupScript(final String zookeeperConnect) {
+    final String listeners = getEnvMap().get("KAFKA_LISTENERS");
+    if (listeners == null) {
+      throw new RuntimeException("Need environment variable KAFKA_LISTENERS");
+    }
+    final String advertisedListeners =
+        overrideAdvertisedListeners(
+            listeners
+                .replaceAll(":" + KAFKA_PORT, ":" + getMappedPort(KAFKA_PORT))
+                .replaceAll("0\\.0\\.0\\.0", getContainerIpAddress()));
+    final String startupScript =
+        overrideStartupScript(
+            "#!/bin/bash\n"
+                + "export KAFKA_ZOOKEEPER_CONNECT='"
+                + zookeeperConnect
+                + "'\n"
+                + "export KAFKA_ADVERTISED_LISTENERS='"
+                + advertisedListeners
+                + "'\n"
+                + ". /etc/confluent/docker/bash-config\n"
+                + "/etc/confluent/docker/configure\n"
+                + "/etc/confluent/docker/launch\n");
+    copyFileToContainer(
+        Transferable.of(startupScript.getBytes(StandardCharsets.UTF_8), 0755), STARTER_SCRIPT);
+  }
+
+  /** Subclasses may override. */
+  protected String overrideAdvertisedListeners(final String advertisedListeners) {
+    return advertisedListeners;
+  }
+
+  /** Subclasses may override. */
+  protected String overrideStartupScript(final String startupScript) {
+    return startupScript;
+  }
+
+  private String startZookeeper() {
+    final ExecCreateCmdResponse execCreateCmdResponse =
+        dockerClient
+            .execCreateCmd(getContainerId())
+            .withCmd(
+                "sh",
+                "-c",
+                "echo '*** Starting Zookeeper'\n"
+                    + "printf 'clientPort="
+                    + ZOOKEEPER_PORT
+                    + "\n"
+                    + "dataDir=/var/lib/zookeeper/data\ndataLogDir=/var/lib/zookeeper/log' > zookeeper.properties\n"
+                    + "zookeeper-server-start zookeeper.properties\n")
+            .withAttachStderr(true)
+            .withAttachStdout(true)
+            .exec();
+    try {
+      dockerClient
+          .execStartCmd(execCreateCmdResponse.getId())
+          .start()
+          .awaitStarted(10, TimeUnit.SECONDS);
+    } catch (final InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+    return "127.0.0.1:" + ZOOKEEPER_PORT;
+  }
+}
diff --git a/src/test/java/com/purbon/kafka/topology/integration/containerutils/ContainerTestUtils.java b/src/test/java/com/purbon/kafka/topology/integration/containerutils/ContainerTestUtils.java
new file mode 100644
index 0000000..f4384e6
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/integration/containerutils/ContainerTestUtils.java
@@ -0,0 +1,104 @@
+package com.purbon.kafka.topology.integration.containerutils;
+
+import com.purbon.kafka.topology.AccessControlProvider;
+import com.purbon.kafka.topology.BindingsBuilderProvider;
+import com.purbon.kafka.topology.KafkaTopologyBuilder;
+import com.purbon.kafka.topology.TopologyBuilderConfig;
+import com.purbon.kafka.topology.api.adminclient.TopologyBuilderAdminClient;
+import com.purbon.kafka.topology.roles.SimpleAclsProvider;
+import com.purbon.kafka.topology.roles.acls.AclsBindingsBuilder;
+import com.purbon.kafka.topology.utils.TestUtils;
+import com.typesafe.config.ConfigFactory;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.AdminClient;
+
+public final class ContainerTestUtils {
+
+  private ContainerTestUtils() {}
+
+  public static AdminClient getSaslAdminClient(final AlternativeKafkaContainer container) {
+    return getSaslAdminClient(container.getBootstrapServers());
+  }
+
+  public static AdminClient getSaslAdminClient(final String boostrapServers) {
+    return AdminClient.create(
+        getSaslConfig(
+            boostrapServers,
+            SaslPlaintextKafkaContainer.DEFAULT_SUPER_USERNAME,
+            SaslPlaintextKafkaContainer.DEFAULT_SUPER_PASSWORD));
+  }
+
+  public static Map<String, Object> getSaslConfig(
+      final String bootstrapServers, final String username, final String password) {
+    final Map<String, Object> map = getBaseConfig(bootstrapServers);
+    map.put("security.protocol", "SASL_PLAINTEXT");
+    map.put("sasl.mechanism", "PLAIN");
+    map.put(
+        "sasl.jaas.config",
+        "org.apache.kafka.common.security.plain.PlainLoginModule required username="
+            + escape(username)
+            + " password="
+            + escape(password)
+            + ";");
+    return map;
+  }
+
+  private static Map<String, Object> getBaseConfig(final String bootstrapServers) {
+    final Map<String, Object> map = new HashMap<>();
+    map.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+    return map;
+  }
+
+  private static String escape(final String s) {
+    return "\"" + s.replace("\\", "\\\\").replace("\"", "\\\"") + "\"";
+  }
+
+  public static void populateAcls(
+      final SaslPlaintextKafkaContainer container,
+      final String topologyResource,
+      final String configResource) {
+    TestUtils.deleteStateFile();
+    try (final AdminClient kafkaAdminClient = getSaslAdminClient(container)) {
+      final KafkaTopologyBuilder kafkaTopologyBuilder =
+          getKafkaTopologyBuilder(kafkaAdminClient, topologyResource, configResource);
+      try {
+        kafkaTopologyBuilder.run();
+      } catch (final IOException e) {
+        throw new RuntimeException(e);
+      }
+    } finally {
+      /* Clean up static stuff buried deep below TopologyBuilderConfig.
+       * I don't like this "hack", but I didn't immediately see
+       * how to do it correctly. */
+      System.clearProperty("config.file");
+      ConfigFactory.invalidateCaches();
+    }
+  }
+
+  private static KafkaTopologyBuilder getKafkaTopologyBuilder(
+      final AdminClient kafkaAdminClient,
+      final String topologyResource,
+      final String configResource) {
+    final String fileOrDirPath = TestUtils.getResourceFilename(topologyResource);
+    final Map<String, String> cliParams = new HashMap<>();
+    final TopologyBuilderConfig builderConfig =
+        TopologyBuilderConfig.build(cliParams, TestUtils.getResourceFilename(configResource));
+    final TopologyBuilderAdminClient topologyAdminClient =
+        new TopologyBuilderAdminClient(kafkaAdminClient);
+    final AccessControlProvider accessControlProvider = new SimpleAclsProvider(topologyAdminClient);
+    final BindingsBuilderProvider bindingsBuilderProvider = new AclsBindingsBuilder(builderConfig);
+    try {
+      return KafkaTopologyBuilder.build(
+          fileOrDirPath,
+          builderConfig,
+          topologyAdminClient,
+          accessControlProvider,
+          bindingsBuilderProvider);
+    } catch (final Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+}
diff --git a/src/test/java/com/purbon/kafka/topology/integration/containerutils/SaslPlaintextKafkaContainer.java b/src/test/java/com/purbon/kafka/topology/integration/containerutils/SaslPlaintextKafkaContainer.java
new file mode 100644
index 0000000..21c208e
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/integration/containerutils/SaslPlaintextKafkaContainer.java
@@ -0,0 +1,122 @@
+package com.purbon.kafka.topology.integration.containerutils;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import org.testcontainers.images.builder.Transferable;
+import org.testcontainers.utility.DockerImageName;
+
+public final class SaslPlaintextKafkaContainer extends AlternativeKafkaContainer {
+
+  private static final DockerImageName DEFAULT_IMAGE =
+      DockerImageName.parse("confluentinc/cp-kafka").withTag("5.5.0");
+  public static final String DEFAULT_SUPER_USERNAME = "kafka";
+  public static final String DEFAULT_SUPER_PASSWORD = "kafka";
+  private static final String JAAS_CONFIG_FILE = "/tmp/broker_jaas.conf";
+  private String superUsername = DEFAULT_SUPER_USERNAME;
+  private String superPassword = DEFAULT_SUPER_PASSWORD;
+  private final Map<String, String> usernamesAndPasswords = new HashMap<>();
+
+  public SaslPlaintextKafkaContainer() {
+    this(DEFAULT_IMAGE);
+  }
+
+  public SaslPlaintextKafkaContainer(final DockerImageName dockerImageName) {
+    super(dockerImageName);
+    /* Note difference between 0.0.0.0 and localhost: The former will be replaced by the container IP. */
+    withEnv(
+        "KAFKA_LISTENERS",
+        "SASL_PLAINTEXT://0.0.0.0:"
+            + KAFKA_PORT
+            + ","
+            + INTERNAL_LISTENER_NAME
+            + "://127.0.0.1:"
+            + KAFKA_INTERNAL_PORT);
+    withEnv(
+        "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
+        "SASL_PLAINTEXT:SASL_PLAINTEXT," + INTERNAL_LISTENER_NAME + ":SASL_PLAINTEXT");
+    withEnv("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "PLAIN");
+    withEnv("KAFKA_SASL_ENABLED_MECHANISMS", "PLAIN");
+    withEnv("KAFKA_OPTS", "-Djava.security.auth.login.config=" + JAAS_CONFIG_FILE);
+    withSuperUser(superUsername, superPassword);
+    withUser("alice", "alice-secret");
+    withUser("bob", "bob-secret");
+    withAclAuthorizer();
+  }
+
+  public SaslPlaintextKafkaContainer withSuperUser(final String username, final String password) {
+    this.superUsername = assertValidUsernameAndPassword(username);
+    this.superPassword = assertValidUsernameAndPassword(password);
+    withEnv("KAFKA_SUPER_USERS", "User:" + username);
+    return this;
+  }
+
+  public SaslPlaintextKafkaContainer withUser(final String usernameAndPassword) {
+    return withUser(usernameAndPassword, usernameAndPassword);
+  }
+
+  public SaslPlaintextKafkaContainer withUser(final String username, final String password) {
+    usernamesAndPasswords.put(
+        assertValidUsernameAndPassword(username), assertValidUsernameAndPassword(password));
+    return this;
+  }
+
+  public SaslPlaintextKafkaContainer withAclAuthorizer() {
+    final DockerImageName imageName = DockerImageName.parse(getDockerImageName());
+    final boolean old =
+        imageName.getUnversionedPart().contains("confluentinc/")
+            && imageName.getVersionPart().compareTo("6") <= 0;
+    withEnv(
+        "KAFKA_AUTHORIZER_CLASS_NAME",
+        old
+            ? "kafka.security.auth.SimpleAclAuthorizer"
+            : "kafka.security.authorizer.AclAuthorizer");
+    return this;
+  }
+
+  @Override
+  protected void beforeStartupPreparations() {
+    withEnv("KAFKA_LISTENER_NAME_SASL_PLAINTEXT_PLAIN_SASL_JAAS_CONFIG", createJaasLoginLine());
+    uploadJaasConfig();
+  }
+
+  private void uploadJaasConfig() {
+    final String jaas = "KafkaServer { " + createJaasLoginLine() + " };\n";
+    copyFileToContainer(
+        Transferable.of(jaas.getBytes(StandardCharsets.UTF_8), 0644), JAAS_CONFIG_FILE);
+  }
+
+  private String createJaasLoginLine() {
+    /* Precondition: No usernames or passwords contain characters that need special handling for JAAS config. */
+    final StringBuilder sb = new StringBuilder();
+    sb.append("org.apache.kafka.common.security.plain.PlainLoginModule required username=\"");
+    sb.append(superUsername);
+    sb.append("\" password=\"");
+    sb.append(superPassword);
+    sb.append("\" user_");
+    sb.append(superUsername);
+    sb.append("=\"");
+    sb.append(superPassword);
+    sb.append("\"");
+    for (final Map.Entry<String, String> entry : usernamesAndPasswords.entrySet()) {
+      sb.append(" user_");
+      sb.append(entry.getKey());
+      sb.append("=\"");
+      sb.append(entry.getValue());
+      sb.append("\"");
+    }
+    sb.append(";");
+    return sb.toString();
+  }
+
+  private static String assertValidUsernameAndPassword(final String s) {
+    /* Enforcing, in order to not have to deal with escaping for the JAAS config. */
+    for (final char c : s.toCharArray()) {
+      if (!(c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '-')) {
+        throw new RuntimeException(
+            "Only letters, digits and hyphens allowed in usernames and passwords.");
+      }
+    }
+    return s;
+  }
+}
diff --git a/src/test/java/com/purbon/kafka/topology/integration/containerutils/TestConsumer.java b/src/test/java/com/purbon/kafka/topology/integration/containerutils/TestConsumer.java
new file mode 100644
index 0000000..2f77c8e
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/integration/containerutils/TestConsumer.java
@@ -0,0 +1,72 @@
+package com.purbon.kafka.topology.integration.containerutils;
+
+import java.io.Closeable;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+
+public final class TestConsumer implements Closeable {
+
+  /* This timeout includes the time spent fetching meta data, which may actually be several hundred ms.
+   * If the timeout is too short, the poll function will return after fetching meta data, but before
+   * actually trying to access the topic, thus not throwing the exception we are looking for. */
+  private static final long MAX_MS_TO_CONSUME = 20 * 1000L;
+  private final KafkaConsumer<String, String> consumer;
+
+  public interface RecordHandler {
+
+    /**
+     * @return <code>true</code> to continue processing, <code>false</code> if no more records are
+     *     requested.
+     */
+    boolean handle(String key, String value);
+  }
+
+  private TestConsumer(final KafkaConsumer<String, String> consumer) {
+    this.consumer = consumer;
+  }
+
+  public static TestConsumer create(
+      final AlternativeKafkaContainer container,
+      final String usernameAndPassword,
+      final String consumerGroup) {
+    return create(container.getBootstrapServers(), usernameAndPassword, consumerGroup);
+  }
+
+  private static TestConsumer create(
+      final String bootstrapServers, final String usernameAndPassword, final String consumerGroup) {
+    final Map<String, Object> config =
+        ContainerTestUtils.getSaslConfig(
+            bootstrapServers, usernameAndPassword, usernameAndPassword);
+    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+    config.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
+    config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+    config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+    return new TestConsumer(new KafkaConsumer<>(config));
+  }
+
+  public void consumeForAWhile(final String topicName, final RecordHandler handler) {
+    consumer.subscribe(Collections.singleton(topicName));
+    final long endTime = System.currentTimeMillis() + MAX_MS_TO_CONSUME;
+    while (System.currentTimeMillis() < endTime) {
+      final ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
+      for (final ConsumerRecord<String, String> record : records) {
+        if (handler != null && !handler.handle(record.key(), record.value())) {
+          break;
+        }
+        consumer.commitAsync();
+      }
+    }
+  }
+
+  @Override
+  public void close() {
+    consumer.close();
+  }
+}
diff --git a/src/test/java/com/purbon/kafka/topology/integration/containerutils/TestProducer.java b/src/test/java/com/purbon/kafka/topology/integration/containerutils/TestProducer.java
new file mode 100644
index 0000000..a174025
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/integration/containerutils/TestProducer.java
@@ -0,0 +1,83 @@
+package com.purbon.kafka.topology.integration.containerutils;
+
+import java.io.Closeable;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+public final class TestProducer implements Closeable {
+
+  private static final int NUM_VALUES_TO_PRODUCE = 3;
+  private final KafkaProducer<String, String> producer;
+
+  private TestProducer(final KafkaProducer<String, String> producer) {
+    this.producer = producer;
+  }
+
+  public static TestProducer create(
+      final AlternativeKafkaContainer container, final String usernameAndPassword) {
+    return create(container.getBootstrapServers(), usernameAndPassword);
+  }
+
+  private static TestProducer create(
+      final String bootstrapServers, final String usernameAndPassword) {
+    final Map<String, Object> config =
+        ContainerTestUtils.getSaslConfig(
+            bootstrapServers, usernameAndPassword, usernameAndPassword);
+    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+    config.put(ProducerConfig.ACKS_CONFIG, "all");
+    config.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, "3000");
+    config.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, "3000");
+    config.put(ProducerConfig.LINGER_MS_CONFIG, "0");
+    config.put(ProducerConfig.RETRIES_CONFIG, "0");
+    return new TestProducer(new KafkaProducer<>(config));
+  }
+
+  public void produce(final String topicName, final String recordValue) {
+    produce(topicName, null, recordValue);
+  }
+
+  public void produce(final String topicName, final String key, final String recordValue) {
+    final ProducerRecord<String, String> record = new ProducerRecord<>(topicName, key, recordValue);
+    try {
+      producer
+          .send(
+              record,
+              (metadata, exception) -> {
+                if (exception != null) {
+                  throw (exception instanceof RuntimeException)
+                      ? (RuntimeException) exception
+                      : new RuntimeException(exception);
+                }
+              })
+          .get(); // Make call synchronous, to be able to get exceptions in time.
+    } catch (final InterruptedException | ExecutionException e) {
+      final Throwable cause = e.getCause();
+      throw (cause instanceof RuntimeException)
+          ? (RuntimeException) cause
+          : new RuntimeException(e);
+    }
+    producer.flush();
+  }
+
+  public Set<String> produceSomeStrings(final String topicName) {
+    final Set<String> values = new HashSet<>();
+    for (int q = 0; q < NUM_VALUES_TO_PRODUCE; q++) {
+      final String value = (q + 1) + "-" + System.currentTimeMillis();
+      values.add(value);
+      produce(topicName, value);
+    }
+    return values;
+  }
+
+  @Override
+  public void close() {
+    producer.close();
+  }
+}
diff --git a/src/test/java/com/purbon/kafka/topology/schemas/SchemaRegistryManagerTest.java b/src/test/java/com/purbon/kafka/topology/schemas/SchemaRegistryManagerTest.java
index 41ab6f0..cffa9c5 100644
--- a/src/test/java/com/purbon/kafka/topology/schemas/SchemaRegistryManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/schemas/SchemaRegistryManagerTest.java
@@ -50,12 +50,12 @@ public class SchemaRegistryManagerTest {
   }
 
   @Test(expected = SchemaRegistryManagerException.class)
-  public void shouldThrowAnExceptionWithFailedFilePath() throws Exception {
+  public void shouldThrowAnExceptionWithFailedFilePath() {
     manager.register(subjectName, "schemas/wrong-file-value.avsc");
   }
 
   @Test
-  public void shouldThrowAnExceptionWithValidRelativeFilePath() throws Exception {
+  public void shouldThrowAnExceptionWithValidRelativeFilePath() {
     manager.register(subjectName, "schemas/bar-value.avsc");
   }
 
diff --git a/src/test/java/com/purbon/kafka/topology/utils/TestUtils.java b/src/test/java/com/purbon/kafka/topology/utils/TestUtils.java
new file mode 100644
index 0000000..a5cef66
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/utils/TestUtils.java
@@ -0,0 +1,35 @@
+package com.purbon.kafka.topology.utils;
+
+import com.purbon.kafka.topology.backend.FileBackend;
+import java.io.File;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+
+public final class TestUtils {
+
+  private TestUtils() {}
+
+  public static void deleteStateFile() {
+    try {
+      Files.deleteIfExists(Paths.get(FileBackend.STATE_FILE_NAME));
+    } catch (final IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  public static String getResourceFilename(final String resource) {
+    return getResourceFile(resource).toString();
+  }
+
+  public static File getResourceFile(final String resource) {
+    final URL resourceUrl = TestUtils.class.getResource(resource);
+    try {
+      return Paths.get(resourceUrl.toURI()).toFile();
+    } catch (final URISyntaxException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}
diff --git a/src/test/resources/acl-producer-and-consumer-it.properties b/src/test/resources/acl-producer-and-consumer-it.properties
new file mode 100644
index 0000000..680b185
--- /dev/null
+++ b/src/test/resources/acl-producer-and-consumer-it.properties
@@ -0,0 +1,4 @@
+kafka.internal.topic.prefixes.0=_
+topology.topic.prefix.format={{topic}}
+topology.project.prefix.format=
+bootstrap.servers=Ignored. Just to please the config parser.
diff --git a/src/test/resources/acl-producer-and-consumer-it.yaml b/src/test/resources/acl-producer-and-consumer-it.yaml
new file mode 100644
index 0000000..52f77a9
--- /dev/null
+++ b/src/test/resources/acl-producer-and-consumer-it.yaml
@@ -0,0 +1,23 @@
+---
+context: "context"
+projects:
+  - name: "project"
+    producers:
+      - principal: "User:producer"
+    consumers:
+      - principal: "User:consumer"
+    topics:
+      - name: "producer-and-consumer-test-topic"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+  - name: "other-project"
+    producers:
+      - principal: "User:other-producer"
+    consumers:
+      - principal: "User:other-consumer"
+    topics:
+      - name: "other-producer-and-consumer-test-topic"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
diff --git a/src/test/resources/descriptor.yaml b/src/test/resources/descriptor.yaml
index 5692542..ad085c7 100644
--- a/src/test/resources/descriptor.yaml
+++ b/src/test/resources/descriptor.yaml
@@ -57,6 +57,24 @@ projects:
         config:
           replication.factor: "1"
           num.partitions: "1"
+  - name: "baz"
+    streams:
+      - principal: "User:App1"
+        applicationId: "applicationId-1"
+        topics:
+          read:
+            - "topicE"
+          write:
+            - "topicF"
+    topics:
+      - name: "topicE"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: "topicF"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
 platform:
   schema_registry:
     instances:
diff --git a/src/test/resources/errors_dir/descriptor-with-errors.yaml b/src/test/resources/errors_dir/descriptor-with-errors.yaml
new file mode 100644
index 0000000..aeb2c72
--- /dev/null
+++ b/src/test/resources/errors_dir/descriptor-with-errors.yaml
@@ -0,0 +1,22 @@
+---
+context: "contextOrg"
+source: "source"
+projects:
+  - name: "foo"
+    streams:
+      - principal: "User:streams"
+        unknownProperty: "catchMe"
+        topics:
+          read:
+            - foo
+          write:
+            - bar
+    topics:
+      - name: "foo"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: "bar"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
diff --git a/src/test/resources/errors_dir/descriptor.yaml b/src/test/resources/errors_dir/descriptor.yaml
new file mode 100644
index 0000000..0cb7005
--- /dev/null
+++ b/src/test/resources/errors_dir/descriptor.yaml
@@ -0,0 +1,11 @@
+---
+context: "contextOrg"
+source: "source"
+projects:
+  - name: "bar"
+    topics:
+      - dataType: "avro"
+        name: "bar"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
\ No newline at end of file