Add Kafka module

core/src/main/java/org/testcontainers/containers/Network.java

@@ -52,7 +52,7 @@ class NetworkImpl extends ExternalResource implements Network {
         private final AtomicBoolean initialized = new AtomicBoolean();
 
         @Override
-        public String getId() {
+        public synchronized String getId() {
             if (initialized.compareAndSet(false, true)) {
                 id = create();
             }

core/src/main/java/org/testcontainers/containers/SocatContainer.java

@@ -0,0 +1,40 @@
+package org.testcontainers.containers;
+
+import org.testcontainers.utility.TestcontainersConfiguration;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * A socat container is used as a TCP proxy, enabling any TCP port of another container to be exposed
+ * publicly, even if that container does not make the port public itself.
+ */
+public class SocatContainer extends GenericContainer<SocatContainer> {
+
+    private final Map<Integer, String> targets = new HashMap<>();
+
+    public SocatContainer() {
+        super(TestcontainersConfiguration.getInstance().getSocatContainerImage());
+        withCreateContainerCmdModifier(it -> it.withEntrypoint("/bin/sh"));
+    }
+
+    public SocatContainer withTarget(int exposedPort, String host) {
+        return withTarget(exposedPort, host, exposedPort);
+    }
+
+    public SocatContainer withTarget(int exposedPort, String host, int internalPort) {
+        addExposedPort(exposedPort);
+        targets.put(exposedPort, String.format("%s:%s", host, internalPort));
+        return self();
+    }
+
+    @Override
+    protected void configure() {
+        withCommand("-c",
+                targets.entrySet().stream()
+                        .map(entry -> "socat TCP-LISTEN:" + entry.getKey() + ",fork,reuseaddr TCP:" + entry.getValue())
+                        .collect(Collectors.joining(" & "))
+        );
+    }
+}

core/src/main/java/org/testcontainers/utility/TestcontainersConfiguration.java

@@ -30,6 +30,10 @@ public String getAmbassadorContainerImage() {
         return (String) properties.getOrDefault("ambassador.container.image", "richnorth/ambassador:latest");
     }
 
+    public String getSocatContainerImage() {
+        return (String) properties.getOrDefault("socat.container.image", "alpine/socat:latest");
+    }
+
     public String getVncRecordedContainerImage() {
         return (String) properties.getOrDefault("vncrecorder.container.image", "richnorth/vnc-recorder:latest");
     }

modules/kafka/pom.xml

@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.testcontainers</groupId>
+        <artifactId>testcontainers-parent</artifactId>
+        <version>0-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>kafka</artifactId>
+    <name>TestContainers :: Apache Kafka</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>testcontainers</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>1.0.0</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>3.8.0</version>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+</project>

modules/kafka/src/main/java/org.testcontainers.containers/KafkaContainer.java

@@ -0,0 +1,61 @@
+package org.testcontainers.containers;
+
+import org.testcontainers.utility.Base58;
+
+import java.util.stream.Stream;
+
+public class KafkaContainer extends GenericContainer<KafkaContainer> {
+
+    public static final int KAFKA_PORT = 9092;
+
+    public static final int ZOOKEEPER_PORT = 2181;
+
+    protected SocatContainer proxy;
+
+    public KafkaContainer() {
+        this("4.0.0");
+    }
+
+    public KafkaContainer(String confluencePlatformVersion) {
+        super("confluentinc/cp-kafka:" + confluencePlatformVersion);
+
+        withNetwork(Network.newNetwork());
+        withNetworkAliases("kafka-" + Base58.randomString(6));
+        withExposedPorts(KAFKA_PORT, ZOOKEEPER_PORT);
+
+        withEnv("KAFKA_BROKER_ID", "1");
+        withEnv("KAFKA_LISTENERS", "PLAINTEXT://0.0.0.0:9092,BROKER://127.0.0.1:9093");
+        withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT");
+        withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER");
+
+        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_ZOOKEEPER_CONNECT", "localhost:2181");
+        withClasspathResourceMapping("tc-zookeeper.properties", "/zookeeper.properties", BindMode.READ_ONLY);
+        withCommand("sh", "-c", "zookeeper-server-start /zookeeper.properties & /etc/confluent/docker/run");
+    }
+
+    public String getBootstrapServers() {
+        return String.format("PLAINTEXT://%s:%s", proxy.getContainerIpAddress(), proxy.getFirstMappedPort());
+    }
+
+    @Override
+    public void start() {
+        proxy = new SocatContainer()
+                .withNetwork(getNetwork())
+                .withTarget(9092, getNetworkAliases().get(0))
+                .withTarget(2181, getNetworkAliases().get(0));
+
+        proxy.start();
+        withEnv("KAFKA_ADVERTISED_LISTENERS", "BROKER://127.0.0.1:9093,PLAINTEXT://" + proxy.getContainerIpAddress() + ":" + proxy.getFirstMappedPort());
+
+        super.start();
+    }
+
+    @Override
+    public void stop() {
+        Stream.<Runnable>of(super::stop, proxy::stop).parallel().forEach(Runnable::run);
+    }
+}

modules/kafka/src/main/resources/tc-zookeeper.properties

@@ -0,0 +1,3 @@
+clientPort=2181
+dataDir=/var/lib/zookeeper/data
+dataLogDir=/var/lib/zookeeper/log

modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java

@@ -0,0 +1,76 @@
+package org.testcontainers.containers;
+
+import com.google.common.collect.ImmutableMap;
+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.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.Test;
+import org.rnorth.ducttape.unreliables.Unreliables;
+
+import java.util.Arrays;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.tuple;
+
+public class KafkaContainerTest {
+
+    @Test
+    public void testUsage() throws Exception {
+        try (KafkaContainer kafka = new KafkaContainer()) {
+            kafka.start();
+
+            try (
+                    KafkaProducer<String, String> producer = new KafkaProducer<>(
+                            ImmutableMap.of(
+                                    ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers(),
+                                    ProducerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString()
+                            ),
+                            new StringSerializer(),
+                            new StringSerializer()
+                    );
+
+                    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(
+                            ImmutableMap.of(
+                                    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers(),
+                                    ConsumerConfig.GROUP_ID_CONFIG, "tc-" + UUID.randomUUID(),
+                                    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"
+                            ),
+                            new StringDeserializer(),
+                            new StringDeserializer()
+                    );
+            ) {
+                String topicName = "messages";
+                consumer.subscribe(Arrays.asList(topicName));
+
+                producer.send(new ProducerRecord<>(topicName, "testcontainers", "rulezzz")).get();
+
+                Unreliables.retryUntilTrue(10, TimeUnit.SECONDS, () -> {
+                    ConsumerRecords<String, String> records = consumer.poll(100);
+
+                    if (records.isEmpty()) {
+                        return false;
+                    }
+
+                    assertThat(records)
+                            .hasSize(1)
+                            .extracting(ConsumerRecord::topic, ConsumerRecord::key, ConsumerRecord::value)
+                            .containsExactly(tuple(topicName, "testcontainers", "rulezzz"));
+
+                    return true;
+                });
+
+                consumer.unsubscribe();
+            }
+
+        }
+    }
+
+}

modules/kafka/src/test/resources/logback-test.xml

@@ -0,0 +1,18 @@
+<configuration>
+
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <!-- encoders are assigned the type
+             ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} %-5level %logger - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <root level="INFO">
+        <appender-ref ref="STDOUT"/>
+    </root>
+
+    <logger name="org.testcontainers" level="DEBUG"/>
+    <logger name="org.testcontainers.shaded" level="WARN"/>
+
+</configuration>

pom.xml

@@ -165,6 +165,7 @@
         <module>modules/postgresql</module>
         <module>modules/selenium</module>
         <module>modules/nginx</module>
+        <module>modules/kafka</module>
         <module>modules/jdbc-test</module>
     </modules>