Add Redpanda module (#5740)

New module uses `docker.redpanda.com/vectorized/redpanda:v22.2.1` docker image as a
base giving the `--mode dev-container` flag added in that version.

Co-authored-by: Sergei Egorov <bsideup@gmail.com>

.github/ISSUE_TEMPLATE/bug_report.yaml

@@ -42,6 +42,7 @@ body:
         - Presto
         - Pulsar
         - RabbitMQ
+        - Redpanda
         - Selenium
         - Solr
         - TiDB

.github/ISSUE_TEMPLATE/enhancement.yaml

@@ -42,6 +42,7 @@ body:
         - Presto
         - Pulsar
         - RabbitMQ
+        - Redpanda
         - Selenium
         - Solr
         - TiDB

.github/ISSUE_TEMPLATE/feature.yaml

@@ -42,6 +42,7 @@ body:
         - Presto
         - Pulsar
         - RabbitMQ
+        - Redpanda
         - Selenium
         - Solr
         - TiDB

.github/dependabot.yml

@@ -187,6 +187,11 @@ updates:
     schedule:
       interval: "monthly"
     open-pull-requests-limit: 10
+  - package-ecosystem: "gradle"
+    directory: "/modules/redpanda"
+    schedule:
+      interval: "monthly"
+    open-pull-requests-limit: 10
   - package-ecosystem: "gradle"
     directory: "/modules/selenium"
     schedule:

.github/labeler.yml

@@ -69,6 +69,8 @@
   - modules/r2dbc/*
 "modules/rabbitmq":
   - modules/rabbitmq/*
+"modules/redpanda":
+  - modules/redpanda/*
 "modules/selenium":
   - modules/selenium/*
 "modules/solr":

docs/modules/redpanda.md

@@ -0,0 +1,38 @@
+# Redpanda
+
+Testcontainers can be used to automatically instantiate and manage [Redpanda](https://redpanda.com/) containers.
+More precisely Testcontainers uses the official Docker images for [Redpanda](https://hub.docker.com/r/vectorized/redpanda/)
+
+!!! note
+    This module uses features provided in `docker.redpanda.com/vectorized/redpanda`.
+
+## Example
+
+Create a `Redpanda` to use it in your tests:
+<!--codeinclude-->
+[Creating a Redpanda](../../modules/redpanda/src/test/java/org/testcontainers/redpanda/RedpandaContainerTest.java) inside_block:constructorWithVersion
+<!--/codeinclude-->
+
+Now your tests or any other process running on your machine can get access to running Redpanda broker by using the following bootstrap server location:
+
+<!--codeinclude-->
+[Bootstrap Servers](../../modules/redpanda/src/test/java/org/testcontainers/redpanda/RedpandaContainerTest.java) inside_block:getBootstrapServers
+<!--/codeinclude-->
+
+## Adding this module to your project dependencies
+
+Add the following dependency to your `pom.xml`/`build.gradle` file:
+
+=== "Gradle"
+```groovy
+testImplementation "org.testcontainers:redpanda:{{latest_version}}"
+```
+=== "Maven"
+```xml
+<dependency>
+    <groupId>org.testcontainers</groupId>
+    <artifactId>redpanda</artifactId>
+    <version>{{latest_version}}</version>
+    <scope>test</scope>
+</dependency>
+```

mkdocs.yml

@@ -78,6 +78,7 @@ nav:
           - modules/nginx.md
           - modules/pulsar.md
           - modules/rabbitmq.md
+          - modules/redpanda.md
           - modules/solr.md
           - modules/toxiproxy.md
           - modules/vault.md

modules/redpanda/build.gradle

@@ -0,0 +1,8 @@
+description = "Testcontainers :: Redpanda"
+
+dependencies {
+    api project(':testcontainers')
+
+    testImplementation 'org.apache.kafka:kafka-clients:3.2.1'
+    testImplementation 'org.assertj:assertj-core:3.23.1'
+}

modules/redpanda/src/main/java/org/testcontainers/redpanda/RedpandaContainer.java

@@ -0,0 +1,60 @@
+package org.testcontainers.redpanda;
+
+import com.github.dockerjava.api.command.InspectContainerResponse;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.images.builder.Transferable;
+import org.testcontainers.utility.ComparableVersion;
+import org.testcontainers.utility.DockerImageName;
+
+/**
+ * Testcontainers implementation for Redpanda.
+ */
+public class RedpandaContainer extends GenericContainer<RedpandaContainer> {
+
+    private static final String REDPANDA_FULL_IMAGE_NAME = "docker.redpanda.com/vectorized/redpanda";
+
+    private static final DockerImageName REDPANDA_IMAGE = DockerImageName.parse(REDPANDA_FULL_IMAGE_NAME);
+
+    private static final int REDPANDA_PORT = 9092;
+
+    private static final String STARTER_SCRIPT = "/testcontainers_start.sh";
+
+    public RedpandaContainer(String image) {
+        this(DockerImageName.parse(image));
+    }
+
+    public RedpandaContainer(DockerImageName imageName) {
+        super(imageName);
+        imageName.assertCompatibleWith(REDPANDA_IMAGE);
+
+        boolean isLessThanBaseVersion = new ComparableVersion(imageName.getVersionPart()).isLessThan("v22.2.1");
+        if (REDPANDA_FULL_IMAGE_NAME.equals(imageName.getUnversionedPart()) && isLessThanBaseVersion) {
+            throw new IllegalArgumentException("Redpanda version must be >= v22.2.1");
+        }
+
+        withExposedPorts(REDPANDA_PORT);
+        withCreateContainerCmdModifier(cmd -> {
+            cmd.withEntrypoint("sh");
+        });
+        waitingFor(Wait.forLogMessage(".*Started Kafka API server.*", 1));
+        withCommand("-c", "while [ ! -f " + STARTER_SCRIPT + " ]; do sleep 0.1; done; " + STARTER_SCRIPT);
+    }
+
+    @Override
+    protected void containerIsStarting(InspectContainerResponse containerInfo) {
+        super.containerIsStarting(containerInfo);
+
+        String command = "#!/bin/bash\n";
+
+        command += "/usr/bin/rpk redpanda start --mode dev-container ";
+        command += "--kafka-addr PLAINTEXT://0.0.0.0:29092,OUTSIDE://0.0.0.0:9092 ";
+        command += "--advertise-kafka-addr PLAINTEXT://kafka:29092,OUTSIDE://" + getHost() + ":" + getMappedPort(9092);
+
+        copyFileToContainer(Transferable.of(command, 0777), STARTER_SCRIPT);
+    }
+
+    public String getBootstrapServers() {
+        return String.format("PLAINTEXT://%s:%s", getHost(), getMappedPort(REDPANDA_PORT));
+    }
+}

modules/redpanda/src/test/java/org/testcontainers/redpanda/RedpandaContainerTest.java

@@ -0,0 +1,130 @@
+package org.testcontainers.redpanda;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.NewTopic;
+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 org.testcontainers.utility.DockerImageName;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.tuple;
+
+public class RedpandaContainerTest {
+
+    private static final String REDPANDA_IMAGE = "docker.redpanda.com/vectorized/redpanda:v22.2.1";
+
+    private static final DockerImageName REDPANDA_DOCKER_IMAGE = DockerImageName.parse(REDPANDA_IMAGE);
+
+    @Test
+    public void testUsage() throws Exception {
+        try (RedpandaContainer container = new RedpandaContainer(REDPANDA_DOCKER_IMAGE)) {
+            container.start();
+            testKafkaFunctionality(container.getBootstrapServers());
+        }
+    }
+
+    @Test
+    public void testUsageWithStringImage() throws Exception {
+        try (
+            // constructorWithVersion {
+            RedpandaContainer container = new RedpandaContainer("docker.redpanda.com/vectorized/redpanda:v22.2.1")
+            // }
+        ) {
+            container.start();
+            testKafkaFunctionality(
+                // getBootstrapServers {
+                container.getBootstrapServers()
+                // }
+            );
+        }
+    }
+
+    @Test
+    public void testNotCompatibleVersion() {
+        assertThatThrownBy(() -> new RedpandaContainer("docker.redpanda.com/vectorized/redpanda:v21.11.19"))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("Redpanda version must be >= v22.2.1");
+    }
+
+    private void testKafkaFunctionality(String bootstrapServers) throws Exception {
+        testKafkaFunctionality(bootstrapServers, 1, 1);
+    }
+
+    private void testKafkaFunctionality(String bootstrapServers, int partitions, int rf) throws Exception {
+        try (
+            AdminClient adminClient = AdminClient.create(
+                ImmutableMap.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
+            );
+            KafkaProducer<String, String> producer = new KafkaProducer<>(
+                ImmutableMap.of(
+                    ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
+                    bootstrapServers,
+                    ProducerConfig.CLIENT_ID_CONFIG,
+                    UUID.randomUUID().toString()
+                ),
+                new StringSerializer(),
+                new StringSerializer()
+            );
+            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(
+                ImmutableMap.of(
+                    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
+                    bootstrapServers,
+                    ConsumerConfig.GROUP_ID_CONFIG,
+                    "tc-" + UUID.randomUUID(),
+                    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
+                    "earliest"
+                ),
+                new StringDeserializer(),
+                new StringDeserializer()
+            );
+        ) {
+            String topicName = "messages-" + UUID.randomUUID();
+
+            Collection<NewTopic> topics = Collections.singletonList(new NewTopic(topicName, partitions, (short) rf));
+            adminClient.createTopics(topics).all().get(30, TimeUnit.SECONDS);
+
+            consumer.subscribe(Collections.singletonList(topicName));
+
+            producer.send(new ProducerRecord<>(topicName, "testcontainers", "rulezzz")).get();
+
+            Unreliables.retryUntilTrue(
+                10,
+                TimeUnit.SECONDS,
+                () -> {
+                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(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/redpanda/src/test/resources/logback-test.xml

@@ -0,0 +1,16 @@
+<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="INFO"/>
+</configuration>