Testcontainers: make Schema Registry configurable (#1489)

akka · Apr 4, 2022 · 9419b5d · 9419b5d
1 parent 3d5cafb
commit 9419b5d
Show file tree

Hide file tree

Showing 5 changed files with 30 additions and 14 deletions.
diff --git a/docs/src/main/paradox/testing-testcontainers.md b/docs/src/main/paradox/testing-testcontainers.md
@@ -9,7 +9,7 @@ Testcontainers also allow you to create a complete Kafka cluster (using Docker c
 
 ## Settings
 
-You can override testcontainers settings to create multi-broker Kafka clusters, or to finetune Kafka Broker and ZooKeeper configuration, by updating @apidoc[KafkaTestkitTestcontainersSettings] in code or configuration.
+You can override testcontainers settings to create multi-broker Kafka clusters, or to finetune Kafka Broker, ZooKeeper and Schema Registry configuration, by updating @apidoc[KafkaTestkitTestcontainersSettings] in code or configuration.
 The @apidoc[KafkaTestkitTestcontainersSettings] type can be used to perform actions such as:
 
 * Set the docker image and tag of Kafka, ZooKeeper, and Schema Registry version to use (a recent Confluent Platform version is used by default)

diff --git a/testkit/src/main/java/akka/kafka/testkit/internal/KafkaContainerCluster.java b/testkit/src/main/java/akka/kafka/testkit/internal/KafkaContainerCluster.java
@@ -126,6 +126,16 @@ public KafkaContainerCluster(
                             "KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", internalTopicsRf + "")
                         .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", internalTopicsRf + ""))
             .collect(Collectors.toList());
+
+    if (useSchemaRegistry) {
+      this.schemaRegistry =
+          Optional.of(
+              new SchemaRegistryContainer(this.schemaRegistryImage)
+                  .withNetworkAliases("schema-registry")
+                  .withCluster(this));
+    } else {
+      this.schemaRegistry = Optional.empty();
+    }
   }
 
   public Network getNetwork() {
@@ -186,16 +196,6 @@ public void start() {
 
       waitForClusterFormation();
 
-      if (useSchemaRegistry) {
-        this.schemaRegistry =
-            Optional.of(
-                new SchemaRegistryContainer(this.schemaRegistryImage)
-                    .withNetworkAliases("schema-registry")
-                    .withCluster(this));
-      } else {
-        this.schemaRegistry = Optional.empty();
-      }
-
       // start schema registry if the container is initialized
       Startables.deepStart(optionalStream(this.schemaRegistry))
           .get(clusterStartTimeout.getSeconds(), SECONDS);

diff --git a/testkit/src/main/mima-filters/3.0.1.backwards.excludes/PR1489.excludes b/testkit/src/main/mima-filters/3.0.1.backwards.excludes/PR1489.excludes
@@ -0,0 +1 @@
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.kafka.testkit.KafkaTestkitTestcontainersSettings.this")
diff --git a/testkit/src/main/scala/akka/kafka/testkit/KafkaTestkitTestcontainersSettings.scala b/testkit/src/main/scala/akka/kafka/testkit/KafkaTestkitTestcontainersSettings.scala
@@ -38,7 +38,8 @@ final class KafkaTestkitTestcontainersSettings private (
     val configureZooKeeperConsumer: java.util.function.Consumer[GenericContainer[_]] =
       new Consumer[GenericContainer[_]]() {
         override def accept(arg: GenericContainer[_]): Unit = ()
-      }
+      },
+    val configureSchemaRegistry: GenericContainer[_] => Unit = _ => ()
 ) {
 
   /**
@@ -180,6 +181,14 @@ final class KafkaTestkitTestcontainersSettings private (
   ): KafkaTestkitTestcontainersSettings =
     copy(configureZooKeeperConsumer = configureZooKeeperConsumer)
 
+  /**
+   * Java Api
+   * Replaces the default schema registry testcontainers configuration logic
+   */
+  def withConfigureSchemaRegistry(
+      configureSchemaRegistry: GenericContainer[_] => Unit
+  ): KafkaTestkitTestcontainersSettings = copy(configureSchemaRegistry = configureSchemaRegistry)
+
   /**
    * Use Schema Registry container.
    */
@@ -237,7 +246,8 @@ final class KafkaTestkitTestcontainersSettings private (
       configureKafkaConsumer: java.util.function.Consumer[java.util.Collection[AlpakkaKafkaContainer]] =
         configureKafkaConsumer,
       configureZooKeeper: GenericContainer[_] => Unit = configureZooKeeper,
-      configureZooKeeperConsumer: java.util.function.Consumer[GenericContainer[_]] = configureZooKeeperConsumer
+      configureZooKeeperConsumer: java.util.function.Consumer[GenericContainer[_]] = configureZooKeeperConsumer,
+      configureSchemaRegistry: GenericContainer[_] => Unit = configureSchemaRegistry
   ): KafkaTestkitTestcontainersSettings =
     new KafkaTestkitTestcontainersSettings(zooKeeperImage,
                                            zooKeeperImageTag,
@@ -254,7 +264,8 @@ final class KafkaTestkitTestcontainersSettings private (
                                            configureKafka,
                                            configureKafkaConsumer,
                                            configureZooKeeper,
-                                           configureZooKeeperConsumer)
+                                           configureZooKeeperConsumer,
+                                           configureSchemaRegistry)
 
   override def toString: String =
     "KafkaTestkitTestcontainersSettings(" +

diff --git a/testkit/src/main/scala/akka/kafka/testkit/internal/TestcontainersKafka.scala b/testkit/src/main/scala/akka/kafka/testkit/internal/TestcontainersKafka.scala
@@ -73,6 +73,10 @@ object TestcontainersKafka {
         configureKafkaConsumer.accept(brokerContainers.asJavaCollection)
         configureZooKeeper(zookeeperContainer)
         configureZooKeeperConsumer.accept(zookeeperContainer)
+        schemaRegistryContainer match {
+          case Some(container) => configureSchemaRegistry(container)
+          case _ =>
+        }
         log.info("Starting Kafka cluster with settings: {}", settings)
         cluster.start()
         kafkaBootstrapServersInternal = cluster.getBootstrapServers
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1 @@
		ProblemFilters.exclude[DirectMissingMethodProblem]("akka.kafka.testkit.KafkaTestkitTestcontainersSettings.this")