This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new 6b970f1e test with Kafka 4 server as well as Kafka 3 (#646)
6b970f1e is described below
commit 6b970f1e2ed5766287fa8eff73ac8f65a3d5f2af
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Sep 1 10:09:35 2026 +0100
test with Kafka 4 server as well as Kafka 3 (#646)
* test with Kafka 4 server as well as Kafka 3
Motivation:
The testkit could start Confluent Platform 8.x (Kafka 4) images only far
enough to skip the ZooKeeper container; the broker itself was still
configured in ZooKeeper mode, which Kafka 4 no longer supports, so the
container never started. CI also only tested against a Kafka 3 server.
Modification:
- PekkoConnectorsKafkaContainer detects 8.x image tags and configures the
broker in KRaft mode (CLUSTER_ID, node id, process roles, controller
listener and quorum voters) and runs /etc/confluent/docker/ensure so the
KRaft storage is formatted; ZooKeeper mode is kept for older images.
- KafkaContainerCluster decides ZooKeeper vs KRaft from the actual Kafka
image tag (instead of a system property that did not reach forked test
JVMs when only the env var was set), shares one cluster id and controller
quorum across brokers, and uses --bootstrap-server for the readiness
check producer since Kafka 4 removed --broker-list.
- withBrokerNum now always applies the broker-<n> network alias (previously
broker number 1 never got its alias), which KRaft quorum voters rely on.
- DEFAULT_CONFLUENT_PLATFORM_VERSION falls back to the
CONFLUENT_PLATFORM_VERSION env var so CI can select the server version.
- CI runs the test job and the integration tests against both Confluent
7.9.2 (Kafka 3) and 8.3.1 (Kafka 4).
Result:
Tests run against a Kafka 4 server in CI while Kafka 3 server coverage is
retained. Testkit users can test against Confluent Platform 8.x images via
CONFLUENT_PLATFORM_VERSION or the testcontainers settings.
Tests:
- CONFLUENT_PLATFORM_VERSION=8.3.1 sbt "tests/testOnly
org.apache.pekko.kafka.scaladsl.TimestampSpec" (KRaft broker + schema registry,
passed)
- CONFLUENT_PLATFORM_VERSION=8.3.1 sbt "tests/testOnly
org.apache.pekko.kafka.scaladsl.ReconnectSpec" (broker stop/start under KRaft,
passed)
- sbt "tests/testOnly org.apache.pekko.kafka.scaladsl.TimestampSpec"
(default 7.9.2 ZooKeeper mode, passed)
- sbt testkit/mimaReportBinaryIssues (passed)
- multi-broker int-tests left to the new CI matrix
References:
Fixes #301
* fix duplicate broker network alias breaking multi-broker KRaft quorum
Motivation:
The 3-broker KRaft integration tests failed in CI: the constructor-default
broker-1 network alias was never removed when the broker number changed,
because testcontainers' getNetworkAliases() returns a copy of the alias
list, so remove() mutated a throwaway list. Every broker container then
also carried the broker-1 alias, and Docker DNS resolved broker-1
round-robin across all brokers (visible in the CI container logs), which
broke the controller quorum and the failover tests.
Modification:
withBrokerNum builds the new alias list and applies it via
setNetworkAliases, which writes through to the container definition.
Result:
Each broker container carries exactly its own broker-<n> alias.
Tests:
- standalone check instantiating brokers 0..2: each container ends up
with exactly its own broker-<n> alias
- multi-broker KRaft covered by the CI integration-test matrix (8.3.1)
References:
Refs #301
---
.github/workflows/check-build-test.yml | 19 +++-
.../testkit/internal/KafkaContainerCluster.java | 36 ++++++-
.../internal/PekkoConnectorsKafkaContainer.java | 109 +++++++++++++++++----
testkit/src/main/resources/reference.conf | 2 +
.../pekko/kafka/scaladsl/RetentionPeriodSpec.scala | 2 +-
5 files changed, 143 insertions(+), 25 deletions(-)
diff --git a/.github/workflows/check-build-test.yml
b/.github/workflows/check-build-test.yml
index f6848013..7c1adeb6 100644
--- a/.github/workflows/check-build-test.yml
+++ b/.github/workflows/check-build-test.yml
@@ -113,9 +113,11 @@ jobs:
fail-fast: false
matrix:
include:
- - { java-version: 17, scala-version: scala213, sbt-opts: '' }
- - { java-version: 17, scala-version: scala3, sbt-opts: '' }
- - { java-version: 17, scala-version: next, sbt-opts: '' }
+ # confluent-version 7.x runs a Kafka 3 server, 8.x runs a Kafka 4
server
+ - { java-version: 17, scala-version: scala213, sbt-opts: '',
confluent-version: '7.9.2' }
+ - { java-version: 17, scala-version: scala213, sbt-opts: '',
confluent-version: '8.3.1' }
+ - { java-version: 17, scala-version: scala3, sbt-opts: '',
confluent-version: '7.9.2' }
+ - { java-version: 17, scala-version: next, sbt-opts: '',
confluent-version: '7.9.2' }
steps:
- name: Checkout
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 #
v7.0.1
@@ -139,6 +141,8 @@ jobs:
uses: sbt/setup-sbt@c7d2d6258b4bd0d3ec5129e6b3453199d3c79729 # v1.5.8
- name: Run tests with Scala ${{ matrix.scala-version }} and Java ${{
matrix.java-version }}
+ env:
+ CONFLUENT_PLATFORM_VERSION: ${{ matrix.confluent-version }}
run: |-
scala_version="$(bash .github/scripts/resolve-scala-version.sh "${{
matrix.scala-version }}")"
sbt "++${scala_version}! test" ${{ matrix.sbt-opts }}
@@ -148,9 +152,14 @@ jobs:
run: find . -name "*.log" -exec ./scripts/cat-log.sh {} \;
integration-test:
- name: Integration tests
+ name: Integration tests (Confluent ${{ matrix.confluent-version }})
needs: [check-code-style, check-code-compilation, check-docs]
runs-on: ubuntu-22.04
+ strategy:
+ fail-fast: false
+ matrix:
+ # confluent-version 7.x runs a Kafka 3 server, 8.x runs a Kafka 4
server
+ confluent-version: ['7.9.2', '8.3.1']
steps:
- name: Checkout
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 #
v7.0.1
@@ -174,6 +183,8 @@ jobs:
uses: sbt/setup-sbt@c7d2d6258b4bd0d3ec5129e6b3453199d3c79729 # v1.5.8
- name: Run multi-broker and long running integration tests
+ env:
+ CONFLUENT_PLATFORM_VERSION: ${{ matrix.confluent-version }}
run: sbt "int-tests/test"
- name: Print logs on failure
diff --git
a/testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/KafkaContainerCluster.java
b/testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/KafkaContainerCluster.java
index 0bd1cf46..b96bf6ad 100644
---
a/testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/KafkaContainerCluster.java
+++
b/testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/KafkaContainerCluster.java
@@ -55,6 +55,8 @@ public class KafkaContainerCluster implements Startable {
private static final String READINESS_CHECK_SCRIPT =
"/testcontainers_readiness_check.sh";
private static final String READINESS_CHECK_TOPIC =
"ready-kafka-container-cluster";
private static final Version BOOTSTRAP_PARAM_MIN_VERSION = new
Version("5.2.0");
+ // Confluent Platform 8.x images run Kafka 4, which supports KRaft mode only
(no ZooKeeper)
+ private static final Version KRAFT_MIN_VERSION = new Version("8");
private final Logger log = LoggerFactory.getLogger(getClass());
private final Version kafkaImageTag;
@@ -111,7 +113,8 @@ public class KafkaContainerCluster implements Startable {
this.network = Network.newNetwork();
this.schemaRegistryImage = schemaRegistryImage;
- if
(!PekkoConnectorsKafkaContainer.DEFAULT_CONFLUENT_PLATFORM_VERSION.startsWith("8."))
{
+ boolean useKraft = this.kafkaImageTag.compareTo(KRAFT_MIN_VERSION) >= 0;
+ if (!useKraft) {
this.zookeeper =
Optional.of(
new GenericContainer(zooKeeperImage)
@@ -122,6 +125,18 @@ public class KafkaContainerCluster implements Startable {
String.valueOf(PekkoConnectorsKafkaContainer.ZOOKEEPER_PORT)));
}
+ // all KRaft nodes must agree on the controller quorum and share one
cluster id
+ String controllerQuorumVoters =
+ IntStream.range(0, this.brokersNum)
+ .mapToObj(
+ brokerNum ->
+ "%s@broker-%s:%s"
+ .formatted(
+ brokerNum,
+ brokerNum,
+
PekkoConnectorsKafkaContainer.KAFKA_CONTROLLER_PORT))
+ .collect(Collectors.joining(","));
+
this.brokers =
IntStream.range(0, this.brokersNum)
.mapToObj(
@@ -131,7 +146,6 @@ public class KafkaContainerCluster implements Startable {
.withNetwork(this.network)
.withBrokerNum(brokerNum)
.withRemoteJmxService()
- .withEnv("KAFKA_BROKER_ID", brokerNum + "")
.withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR",
internalTopicsRf + "")
.withEnv("KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS",
internalTopicsRf + "")
.withEnv(
@@ -144,6 +158,11 @@ public class KafkaContainerCluster implements Startable {
.dependsOn(this.zookeeper.get())
.withExternalZookeeper(
"zookeeper:" +
PekkoConnectorsKafkaContainer.ZOOKEEPER_PORT);
+ } else if (useKraft) {
+ container =
+ container
+ .withEnv("CLUSTER_ID",
PekkoConnectorsKafkaContainer.DEFAULT_CLUSTER_ID)
+ .withEnv("KAFKA_CONTROLLER_QUORUM_VOTERS",
controllerQuorumVoters);
}
return container;
})
@@ -320,7 +339,9 @@ public class KafkaContainerCluster implements Startable {
+ " \n";
command += "MESSAGE=\"`date -u`\" \n";
command +=
- "echo \"$MESSAGE\" | kafka-console-producer --broker-list
localhost:9092 --topic "
+ "echo \"$MESSAGE\" | kafka-console-producer "
+ + consoleProducerConnectParam()
+ + " --topic "
+ READINESS_CHECK_TOPIC
+ " --producer-property acks=all \n";
command +=
@@ -340,6 +361,15 @@ public class KafkaContainerCluster implements Startable {
}
}
+ private String consoleProducerConnectParam() {
+ // Kafka 4 removed the deprecated --broker-list option
+ if (this.kafkaImageTag.compareTo(KRAFT_MIN_VERSION) >= 0) {
+ return "--bootstrap-server localhost:9092";
+ } else {
+ return "--broker-list localhost:9092";
+ }
+ }
+
private Boolean runReadinessCheck(GenericContainer c) {
try {
Container.ExecResult result = c.execInContainer("sh", "-c",
READINESS_CHECK_SCRIPT);
diff --git
a/testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/PekkoConnectorsKafkaContainer.java
b/testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/PekkoConnectorsKafkaContainer.java
index c5be7232..ffb9ba08 100644
---
a/testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/PekkoConnectorsKafkaContainer.java
+++
b/testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/PekkoConnectorsKafkaContainer.java
@@ -42,7 +42,9 @@ public class PekkoConnectorsKafkaContainer extends
GenericContainer<PekkoConnect
// Align these confluent platform constants with
testkit/src/main/resources/reference.conf
public static final String DEFAULT_CONFLUENT_PLATFORM_VERSION =
- System.getProperty("CONFLUENT_PLATFORM_VERSION", "7.9.2");
+ System.getProperty(
+ "CONFLUENT_PLATFORM_VERSION",
+ System.getenv().getOrDefault("CONFLUENT_PLATFORM_VERSION", "7.9.2"));
public static final DockerImageName DEFAULT_ZOOKEEPER_IMAGE_NAME =
DockerImageName.parse("confluentinc/cp-zookeeper")
@@ -57,6 +59,11 @@ public class PekkoConnectorsKafkaContainer extends
GenericContainer<PekkoConnect
public static final int ZOOKEEPER_PORT = 2181;
+ public static final int KAFKA_CONTROLLER_PORT = 9094;
+
+ // matches the default cluster id used by testcontainers' KafkaContainer
+ public static final String DEFAULT_CLUSTER_ID = "4L6g3nShT-eMCtK--X86sw";
+
private static final int PORT_NOT_ASSIGNED = -1;
protected String externalZookeeperConnect = null;
@@ -70,6 +77,8 @@ public class PekkoConnectorsKafkaContainer extends
GenericContainer<PekkoConnect
private boolean enableRemoteJmxService = false;
+ private final boolean useKraft;
+
public PekkoConnectorsKafkaContainer() {
this(DEFAULT_KAFKA_IMAGE_NAME);
}
@@ -77,6 +86,9 @@ public class PekkoConnectorsKafkaContainer extends
GenericContainer<PekkoConnect
public PekkoConnectorsKafkaContainer(final DockerImageName dockerImageName) {
super(dockerImageName);
+ // Kafka 4 (Confluent Platform 8.x) has no ZooKeeper support, brokers must
run in KRaft mode
+ this.useKraft = requiresKraftMode(dockerImageName.getVersionPart());
+
super.withNetwork(Network.SHARED);
withExposedPorts(KAFKA_PORT);
@@ -100,6 +112,23 @@ public class PekkoConnectorsKafkaContainer extends
GenericContainer<PekkoConnect
withEnv("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0");
}
+ private static boolean requiresKraftMode(String imageVersionPart) {
+ if (imageVersionPart == null) {
+ return false;
+ }
+ int dotIndex = imageVersionPart.indexOf('.');
+ String major = dotIndex > 0 ? imageVersionPart.substring(0, dotIndex) :
imageVersionPart;
+ try {
+ return Integer.parseInt(major) >= 8;
+ } catch (NumberFormatException e) {
+ return false;
+ }
+ }
+
+ public boolean usesKraftMode() {
+ return useKraft;
+ }
+
@Override
public PekkoConnectorsKafkaContainer withNetwork(Network network) {
useImplicitNetwork = false;
@@ -107,12 +136,16 @@ public class PekkoConnectorsKafkaContainer extends
GenericContainer<PekkoConnect
}
public PekkoConnectorsKafkaContainer withBrokerNum(int brokerNum) {
- if (brokerNum != this.brokerNum) {
- this.brokerNum = brokerNum;
- return super.withNetworkAliases("broker-" + this.brokerNum)
- .withEnv("KAFKA_BROKER_ID", "" + this.brokerNum);
+ // getNetworkAliases() returns a copy, so build the new alias list and set
it explicitly
+ // to make sure the alias of the previous broker number is dropped
+ List<String> aliases = getNetworkAliases();
+ aliases.remove("broker-" + this.brokerNum);
+ this.brokerNum = brokerNum;
+ if (!aliases.contains("broker-" + brokerNum)) {
+ aliases.add("broker-" + brokerNum);
}
- return this;
+ setNetworkAliases(aliases);
+ return withEnv("KAFKA_BROKER_ID", "" + this.brokerNum);
}
@Override
@@ -157,6 +190,11 @@ public class PekkoConnectorsKafkaContainer extends
GenericContainer<PekkoConnect
}
public PekkoConnectorsKafkaContainer withExternalZookeeper(String
connectString) {
+ if (useKraft) {
+ throw new IllegalStateException(
+ "ZooKeeper is not supported with Kafka 4 (Confluent Platform 8.x)
images, "
+ + "the broker runs in KRaft mode");
+ }
externalZookeeperConnect = connectString;
return self();
}
@@ -188,7 +226,9 @@ public class PekkoConnectorsKafkaContainer extends
GenericContainer<PekkoConnect
"-c",
"while [ ! -f " + START_STOP_SCRIPT + " ]; do sleep 0.1; done; " +
START_STOP_SCRIPT);
- if (externalZookeeperConnect == null) {
+ if (useKraft) {
+ configureKraft();
+ } else if (externalZookeeperConnect == null) {
addExposedPort(ZOOKEEPER_PORT);
}
if (enableRemoteJmxService) {
@@ -198,6 +238,34 @@ public class PekkoConnectorsKafkaContainer extends
GenericContainer<PekkoConnect
super.doStart();
}
+ private void configureKraft() {
+ if (!getEnvMap().containsKey("CLUSTER_ID")) {
+ withEnv("CLUSTER_ID", DEFAULT_CLUSTER_ID);
+ }
+ // KRaft uses node.id instead of broker.id
+ getEnvMap().remove("KAFKA_BROKER_ID");
+ withEnv("KAFKA_NODE_ID", brokerNum + "");
+ withEnv("KAFKA_PROCESS_ROLES", "broker,controller");
+ withEnv("KAFKA_CONTROLLER_LISTENER_NAMES", "CONTROLLER");
+ String listeners = getEnvMap().get("KAFKA_LISTENERS");
+ if (listeners != null && !listeners.contains("CONTROLLER://")) {
+ withEnv("KAFKA_LISTENERS", listeners + ",CONTROLLER://0.0.0.0:" +
KAFKA_CONTROLLER_PORT);
+ }
+ String protocolMap =
getEnvMap().get("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP");
+ if (protocolMap != null && !protocolMap.contains("CONTROLLER:")) {
+ withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", protocolMap +
",CONTROLLER:PLAINTEXT");
+ }
+ if (!getEnvMap().containsKey("KAFKA_CONTROLLER_QUORUM_VOTERS")) {
+ // a single node quorum, KafkaContainerCluster sets the full quorum for
multi-broker setups
+ String host =
+ getNetwork() != null
+ ? getNetworkAliases().stream().findFirst().orElse("localhost")
+ : "localhost";
+ withEnv(
+ "KAFKA_CONTROLLER_QUORUM_VOTERS", brokerNum + "@" + host + ":" +
KAFKA_CONTROLLER_PORT);
+ }
+ }
+
@Override
protected void containerIsStarting(InspectContainerResponse containerInfo,
boolean reused) {
try {
@@ -213,15 +281,18 @@ public class PekkoConnectorsKafkaContainer extends
GenericContainer<PekkoConnect
}
String command = "#!/bin/bash\n";
- final String zookeeperConnect;
- if (externalZookeeperConnect != null) {
- zookeeperConnect = externalZookeeperConnect;
- } else {
- zookeeperConnect = "localhost:" + ZOOKEEPER_PORT;
- command += "echo 'clientPort=" + ZOOKEEPER_PORT + "' >
zookeeper.properties\n";
- command += "echo 'dataDir=/var/lib/zookeeper/data' >>
zookeeper.properties\n";
- command += "echo 'dataLogDir=/var/lib/zookeeper/log' >>
zookeeper.properties\n";
- command += "zookeeper-server-start zookeeper.properties &\n";
+ if (!useKraft) {
+ final String zookeeperConnect;
+ if (externalZookeeperConnect != null) {
+ zookeeperConnect = externalZookeeperConnect;
+ } else {
+ zookeeperConnect = "localhost:" + ZOOKEEPER_PORT;
+ command += "echo 'clientPort=" + ZOOKEEPER_PORT + "' >
zookeeper.properties\n";
+ command += "echo 'dataDir=/var/lib/zookeeper/data' >>
zookeeper.properties\n";
+ command += "echo 'dataLogDir=/var/lib/zookeeper/log' >>
zookeeper.properties\n";
+ command += "zookeeper-server-start zookeeper.properties &\n";
+ }
+ command += "export KAFKA_ZOOKEEPER_CONNECT='" + zookeeperConnect +
"'\n";
}
List<String> internalIps =
@@ -229,7 +300,6 @@ public class PekkoConnectorsKafkaContainer extends
GenericContainer<PekkoConnect
.map(ContainerNetwork::getIpAddress)
.toList();
- command += "export KAFKA_ZOOKEEPER_CONNECT='" + zookeeperConnect + "'\n";
command +=
"export KAFKA_ADVERTISED_LISTENERS='"
+ Stream.concat(
@@ -249,6 +319,11 @@ public class PekkoConnectorsKafkaContainer extends
GenericContainer<PekkoConnect
command += ". /etc/confluent/docker/bash-config \n";
command += "/etc/confluent/docker/configure \n";
+ if (useKraft) {
+ // formats the KRaft storage directory with the provided CLUSTER_ID
(no-op when
+ // already formatted, so the broker can be stopped and started again)
+ command += "/etc/confluent/docker/ensure \n";
+ }
command += "/etc/confluent/docker/launch \n";
copyFileToContainer(
diff --git a/testkit/src/main/resources/reference.conf
b/testkit/src/main/resources/reference.conf
index c0bc91fb..1cb5fdcc 100644
--- a/testkit/src/main/resources/reference.conf
+++ b/testkit/src/main/resources/reference.conf
@@ -21,6 +21,8 @@ pekko.kafka.testkit.testcontainers {
# we recommend using Confluent Platform docker images and using the same
version across all images
# Confluent publishes images on DockerHub:
https://hub.docker.com/r/confluentinc/cp-kafka/tags
# Kafka versions in Confluent Platform:
https://docs.confluent.io/current/installation/versions-interoperability.html
+ # with Confluent Platform 8.x images (Kafka 4) the brokers run in KRaft mode
and no ZooKeeper
+ # container is started, so the zookeeper-image settings are ignored
zookeeper-image = "confluentinc/cp-zookeeper"
zookeeper-image-tag =
${pekko.kafka.testkit.testcontainers.confluent-platform-version}
kafka-image = "confluentinc/cp-kafka"
diff --git
a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RetentionPeriodSpec.scala
b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RetentionPeriodSpec.scala
index 2dfe0de7..b1ce993e 100644
---
a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RetentionPeriodSpec.scala
+++
b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RetentionPeriodSpec.scala
@@ -32,7 +32,7 @@ import scala.jdk.CollectionConverters._
class RetentionPeriodSpec extends SpecBase with
TestcontainersKafkaPerClassLike {
//
https://docs.confluent.io/current/installation/versions-interoperability.html
private final val confluentPlatformVersion =
- System.getProperty("CONFLUENT_PLATFORM_VERSION", "7.9.2")
+
pekko.kafka.testkit.internal.PekkoConnectorsKafkaContainer.DEFAULT_CONFLUENT_PLATFORM_VERSION
override val testcontainersSettings =
KafkaTestkitTestcontainersSettings(system)
.withKafkaImageTag(confluentPlatformVersion)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
