Repository: kafka-ops/kafka-topology-builder
Pull request #150 (2020-12-01 12:14:21)

Integration test verifying that producers can produce and consumers can consume
Verifying that the populated ACL mappings are sufficient to allow the intended operations for producers and consumers.


Patch:
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/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/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/KafkaTopologyBuilderTest.java b/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
index f24bc17..20d7b37 100644
--- a/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/KafkaTopologyBuilderTest.java
@@ -8,6 +8,7 @@ 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.util.HashMap;
@@ -69,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);
diff --git a/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java b/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java
index 7cc0f3c..7964bd3 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologyObjectBuilderTest.java
@@ -3,6 +3,7 @@ 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;
@@ -58,4 +59,16 @@ public class TopologyObjectBuilderTest {
     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 f4011a3..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;
@@ -40,7 +41,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testDynamicFirstLevelAttributes() throws IOException {
+  public void testDynamicFirstLevelAttributes() {
     Topology topology =
         parser.deserialise(TestUtils.getResourceFile("/descriptor-with-others.yml"));
     Project project = topology.getProjects().get(0);
@@ -154,18 +155,18 @@ public class TopologySerdesTest {
     assertEquals(topic2.getDataType(), serdesTopic2.getDataType());
   }
 
-  @Test(expected = IOException.class)
-  public void testTopologyWithNoTeam() throws IOException {
+  @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 {
+  @Test(expected = TopologyParsingException.class)
+  public void testTopologyWithNoProject() {
     parser.deserialise(TestUtils.getResourceFile("/descriptor-with-no-project.yaml"));
   }
 
   @Test
-  public void testCoreElementsProcessing() throws IOException {
+  public void testCoreElementsProcessing() {
     Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
 
     Project project = topology.getProjects().get(0);
@@ -180,7 +181,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testStreamsApps() throws IOException {
+  public void testStreamsApps() {
     Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
 
     Project project1 = topology.getProjects().get(0);
@@ -194,12 +195,12 @@ public class TopologySerdesTest {
   }
 
   @Test(expected = IllegalArgumentException.class)
-  public void testSchemaSerdes() throws IOException {
+  public void testSchemaSerdes() {
     parser.deserialise(TestUtils.getResourceFile("/descriptor-wrong-schemas.yaml"));
   }
 
   @Test
-  public void testPlaformProcessing() throws IOException {
+  public void testPlaformProcessing() {
     Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
 
     assertEquals("contextOrg", topology.getContext());
@@ -220,7 +221,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testOnlyTopics() throws IOException {
+  public void testOnlyTopics() {
     Topology topology =
         parser.deserialise(TestUtils.getResourceFile("/descriptor-only-topics.yaml"));
 
@@ -232,7 +233,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testRBACTopics() throws IOException {
+  public void testRBACTopics() {
     Topology topology =
         parser.deserialise(TestUtils.getResourceFile("/descriptor-with-rbac-topics.yaml"));
 
@@ -255,7 +256,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testWithRBACDescriptor() throws IOException {
+  public void testWithRBACDescriptor() {
     Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor-with-rbac.yaml"));
     Project myProject = topology.getProjects().get(0);
 
@@ -288,7 +289,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testTopicNameWithCustomSeparator() throws IOException {
+  public void testTopicNameWithCustomSeparator() {
     Map<String, String> cliOps = new HashMap<>();
     cliOps.put(BROKERS_OPTION, "");
     cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, "/fooBar");
@@ -312,7 +313,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testTopicNameWithCustomPattern() throws IOException {
+  public void testTopicNameWithCustomPattern() {
     Map<String, String> cliOps = new HashMap<>();
     cliOps.put(BROKERS_OPTION, "");
     cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, "/fooBar");
@@ -336,7 +337,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testJsonDescriptorFileSerdes() throws IOException {
+  public void testJsonDescriptorFileSerdes() {
     TopologySerdes parser = new TopologySerdes(new TopologyBuilderConfig(), FileType.JSON);
     Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.json"));
 
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 c47380d..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,7 @@ 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;
@@ -69,7 +70,7 @@ public class AccessControlManagerIT {
 
   @Before
   public void before() throws IOException {
-    kafkaAdminClient = container.getAdminClient();
+    kafkaAdminClient = ContainerTestUtils.getSaslAdminClient(container);
     TopologyBuilderAdminClient adminClient = new TopologyBuilderAdminClient(kafkaAdminClient);
     adminClient.clearAcls();
     TestUtils.deleteStateFile();
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/TopicManagerIT.java b/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java
index fd06577..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,7 @@ 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;
@@ -65,7 +66,7 @@ public class TopicManagerIT {
 
   @Before
   public void before() throws IOException {
-    kafkaAdminClient = container.getAdminClient();
+    kafkaAdminClient = ContainerTestUtils.getSaslAdminClient(container);
     TopologyBuilderAdminClient adminClient = new TopologyBuilderAdminClient(kafkaAdminClient);
 
     final SchemaRegistryClient schemaRegistryClient = new MockSchemaRegistryClient();
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
index ee11e40..21c208e 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/containerutils/SaslPlaintextKafkaContainer.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/containerutils/SaslPlaintextKafkaContainer.java
@@ -1,38 +1,21 @@
 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.HashMap;
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
-import org.apache.kafka.clients.CommonClientConfigs;
-import org.apache.kafka.clients.admin.AdminClient;
-import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.images.builder.Transferable;
 import org.testcontainers.utility.DockerImageName;
 
-public final class SaslPlaintextKafkaContainer
-    extends GenericContainer<SaslPlaintextKafkaContainer> {
+public final class SaslPlaintextKafkaContainer extends 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 DEFAULT_SUPER_USERNAME = "kafka";
+  public static final String DEFAULT_SUPER_PASSWORD = "kafka";
   private static final String JAAS_CONFIG_FILE = "/tmp/broker_jaas.conf";
-  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;
-  private static final int PORT_NOT_ASSIGNED = -1;
-  /* Note difference between 0.0.0.0 and localhost: The former will be replaced by the container IP. */
-  private static final String LISTENERS =
-      "SASL_PLAINTEXT://0.0.0.0:"
-          + KAFKA_PORT
-          + ","
-          + INTERNAL_LISTENER_NAME
-          + "://127.0.0.1:"
-          + KAFKA_INTERNAL_PORT;
-  private int port = PORT_NOT_ASSIGNED;
+  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);
@@ -40,130 +23,100 @@ public final class SaslPlaintextKafkaContainer
 
   public SaslPlaintextKafkaContainer(final DockerImageName dockerImageName) {
     super(dockerImageName);
-    withExposedPorts(KAFKA_PORT, ZOOKEEPER_PORT);
-    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");
-    withEnv("KAFKA_LISTENERS", LISTENERS);
+    /* 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_INTER_BROKER_LISTENER_NAME", INTERNAL_LISTENER_NAME);
     withEnv("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "PLAIN");
     withEnv("KAFKA_SASL_ENABLED_MECHANISMS", "PLAIN");
-    withEnv(
-        "KAFKA_AUTHORIZER_CLASS_NAME",
-        dockerImageName.getVersionPart().compareTo("6") >= 0
-            ? "kafka.security.authorizer.AclAuthorizer"
-            : "kafka.security.auth.SimpleAclAuthorizer");
-    withEnv("KAFKA_SUPER_USERS", "User:kafka");
-    withEnv(
-        "KAFKA_LISTENER_NAME_SASL_PLAINTEXT_PLAIN_SASL_JAAS_CONFIG",
-        "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"kafka\" password=\"kafka\" user_kafka=\"kafka\" user_alice=\"alice-secret\" user_bob=\"bob-secret\"");
     withEnv("KAFKA_OPTS", "-Djava.security.auth.login.config=" + JAAS_CONFIG_FILE);
+    withSuperUser(superUsername, superPassword);
+    withUser("alice", "alice-secret");
+    withUser("bob", "bob-secret");
+    withAclAuthorizer();
   }
 
-  public String getBootstrapServers() {
-    if (port == PORT_NOT_ASSIGNED) {
-      throw new IllegalStateException("You should start Kafka container first");
-    }
-    return String.format("%s:%s", getHost(), port);
+  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;
   }
 
-  @Override
-  protected void doStart() {
-    withCommand(
-        "sh", "-c", "while [ ! -f " + STARTER_SCRIPT + " ]; do sleep 0.1; done; " + STARTER_SCRIPT);
-    super.doStart();
+  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 containerIsStarting(
-      final InspectContainerResponse containerInfo, final boolean reused) {
-    super.containerIsStarting(containerInfo, reused);
-    port = getMappedPort(KAFKA_PORT);
-    if (reused) {
-      return;
-    }
-    final String zookeeperConnect = startZookeeper();
+  protected void beforeStartupPreparations() {
+    withEnv("KAFKA_LISTENER_NAME_SASL_PLAINTEXT_PLAIN_SASL_JAAS_CONFIG", createJaasLoginLine());
     uploadJaasConfig();
-    createStartupScript(zookeeperConnect);
   }
 
   private void uploadJaasConfig() {
-    final String jaas =
-        "KafkaServer { org.apache.kafka.common.security.plain.PlainLoginModule required "
-            + "username=\"kafka\" password=\"kafka\" "
-            + "user_kafka=\"kafka\" user_alice=\"alice-secret\" user_bob=\"bob-secret\"; };\n";
+    final String jaas = "KafkaServer { " + createJaasLoginLine() + " };\n";
     copyFileToContainer(
         Transferable.of(jaas.getBytes(StandardCharsets.UTF_8), 0644), JAAS_CONFIG_FILE);
   }
 
-  private void createStartupScript(final String zookeeperConnect) {
-    final String listeners = getEnvMap().get("KAFKA_LISTENERS");
-    if (listeners == null) {
-      throw new RuntimeException("Need environment variable KAFKA_LISTENERS");
+  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("\"");
     }
-    final String advertisedListeners =
-        listeners
-            .replaceAll(":" + KAFKA_PORT, ":" + getMappedPort(KAFKA_PORT))
-            .replaceAll("0\\.0\\.0\\.0", getContainerIpAddress());
-    final String starterScript =
-        "#!/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(starterScript.getBytes(StandardCharsets.UTF_8), 0755), STARTER_SCRIPT);
+    sb.append(";");
+    return sb.toString();
   }
 
-  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);
+  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 "127.0.0.1:" + ZOOKEEPER_PORT;
-  }
-
-  public AdminClient getAdminClient() {
-    return AdminClient.create(getClientConfig());
-  }
-
-  private Map<String, Object> getClientConfig() {
-    final Map<String, Object> map = new HashMap<>();
-    map.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, getBootstrapServers());
-    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=\"kafka\" password=\"kafka\";");
-    return map;
+    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/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/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