This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 1f6d195f2b2 NIFI-16158 Switched Kafka IT tests from
confluentinc/cp-kafka to apache/kafka (#11494)
1f6d195f2b2 is described below
commit 1f6d195f2b25baa8835f1b518973bbed4b4af16e
Author: Pierre Villard <[email protected]>
AuthorDate: Thu Jul 30 22:40:02 2026 +0200
NIFI-16158 Switched Kafka IT tests from confluentinc/cp-kafka to
apache/kafka (#11494)
Signed-off-by: David Handermann <[email protected]>
---
.../nifi/connectors/kafkas3/KafkaToS3IT.java | 68 ++++++++--------------
.../nifi/kafka/processors/AbstractKafkaBaseIT.java | 8 +--
.../PublishKafkaAuthSaslPlaintextIT.java | 17 ++----
.../service/Kafka3ConnectionServiceBaseIT.java | 27 +++++----
.../service/Kafka3ConnectionServiceSSLIT.java | 4 +-
.../Kafka3ConnectionServiceSaslPlaintextIT.java | 1 +
.../nifi/kafka/service/KafkaConnectivityIT.java | 6 +-
pom.xml | 6 ++
8 files changed, 62 insertions(+), 75 deletions(-)
diff --git
a/nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-integration-tests/src/test/java/org/apache/nifi/connectors/kafkas3/KafkaToS3IT.java
b/nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-integration-tests/src/test/java/org/apache/nifi/connectors/kafkas3/KafkaToS3IT.java
index 0165d601069..6ed6bede8cd 100644
---
a/nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-integration-tests/src/test/java/org/apache/nifi/connectors/kafkas3/KafkaToS3IT.java
+++
b/nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-integration-tests/src/test/java/org/apache/nifi/connectors/kafkas3/KafkaToS3IT.java
@@ -45,7 +45,7 @@ import org.junit.jupiter.api.Test;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.wait.strategy.Wait;
-import org.testcontainers.kafka.ConfluentKafkaContainer;
+import org.testcontainers.kafka.KafkaContainer;
import org.testcontainers.localstack.LocalStackContainer;
import org.testcontainers.utility.DockerImageName;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
@@ -78,9 +78,11 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class KafkaToS3IT {
+ private static final String KAFKA_IMAGE_NAME =
System.getProperty("kafka.docker.image", "apache/kafka:4.3.1");
+
private static ConnectorTestRunner runner;
private static Network network;
- private static ConfluentKafkaContainer kafkaContainer;
+ private static KafkaContainer kafkaContainer;
private static GenericContainer<?> schemaRegistryContainer;
private static LocalStackContainer localStackContainer;
private static S3Client s3Client;
@@ -90,50 +92,30 @@ public class KafkaToS3IT {
private static final String S3_REGION = "us-west-2";
- // JAAS configuration for Kafka broker SASL/PLAIN authentication.
- // The 'username' and 'password' fields are credentials the broker uses
for inter-broker communication.
- // The 'user_<username>="<password>"' entries define client users that can
authenticate to this broker.
- // In this setup:
- // - Broker uses 'admin' / 'admin-secret' for inter-broker communication
(though we use PLAINTEXT for that)
- // - Clients can authenticate using 'testuser' / 'testpassword' on the
SASL listener with PLAIN mechanism
- private static final String JAAS_CONFIG_CONTENT = """
- KafkaServer {
- org.apache.kafka.common.security.plain.PlainLoginModule required
- username="admin"
- password="admin-secret"
- user_%s="%s";
- };
- """.formatted(SCRAM_USERNAME, SCRAM_PASSWORD);
-
-
@BeforeAll
public static void setupTestContainers() {
network = Network.newNetwork();
- kafkaContainer = new
ConfluentKafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.8.0"))
+ kafkaContainer = new
KafkaContainer(DockerImageName.parse(KAFKA_IMAGE_NAME))
.withNetwork(network)
- .withNetworkAliases("kafka")
+ .withListener("kafka:19092")
.withStartupTimeout(Duration.ofSeconds(10))
- .withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
"CONTROLLER:PLAINTEXT,BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SASL:SASL_PLAINTEXT")
- .withEnv("KAFKA_LISTENERS",
"CONTROLLER://0.0.0.0:9094,BROKER://0.0.0.0:9092,PLAINTEXT://0.0.0.0:19092,SASL://0.0.0.0:9093")
- .withEnv("KAFKA_ADVERTISED_LISTENERS",
"BROKER://kafka:9092,PLAINTEXT://kafka:19092,SASL://localhost:9093")
- .withEnv("KAFKA_CONTROLLER_LISTENER_NAMES", "CONTROLLER")
- .withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER")
+ .withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
"BROKER:PLAINTEXT,PLAINTEXT:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT")
.withEnv("KAFKA_SASL_ENABLED_MECHANISMS", "PLAIN")
+ .withEnv("KAFKA_LISTENER_NAME_PLAINTEXT_SASL_ENABLED_MECHANISMS",
"PLAIN")
+ .withEnv("KAFKA_LISTENER_NAME_PLAINTEXT_PLAIN_SASL_JAAS_CONFIG",
String.format(
+ "%s required user_%s=\"%s\";",
+
org.apache.kafka.common.security.plain.PlainLoginModule.class.getName(),
+ SCRAM_USERNAME,
+ SCRAM_PASSWORD
+ ))
.withEnv("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0")
.withEnv("KAFKA_GROUP_MIN_SESSION_TIMEOUT_MS", "1000")
.withEnv("KAFKA_GROUP_MAX_SESSION_TIMEOUT_MS", "60000")
.withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1")
.withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
- .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
- .withEnv("KAFKA_OPTS",
"-Djava.security.auth.login.config=/tmp/kafka_jaas.conf")
- .withCommand(
- "sh", "-c",
- "echo '" + JAAS_CONFIG_CONTENT + "' > /tmp/kafka_jaas.conf &&
" +
- "/etc/confluent/docker/run"
- );
-
- kafkaContainer.setPortBindings(List.of("9093:9093"));
+ .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1");
+
kafkaContainer.start();
schemaRegistryContainer = new
GenericContainer<>(DockerImageName.parse("confluentinc/cp-schema-registry:7.8.0"))
@@ -209,7 +191,7 @@ public class KafkaToS3IT {
private void createKafkaTopics(final String... topicNames) throws
ExecutionException, InterruptedException {
final Properties adminProps = new Properties();
- adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9093");
+ adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaContainer.getBootstrapServers());
adminProps.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG,
"SASL_PLAINTEXT");
adminProps.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
adminProps.put(SaslConfigs.SASL_JAAS_CONFIG, String.format(
@@ -229,7 +211,7 @@ public class KafkaToS3IT {
private void produceRecordsToTopic(final String topicName, final String...
records) throws ExecutionException, InterruptedException {
final Properties producerProps = new Properties();
- producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9093");
+ producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaContainer.getBootstrapServers());
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
producerProps.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG,
"SASL_PLAINTEXT");
@@ -255,7 +237,7 @@ public class KafkaToS3IT {
private void produceAvroRecordsToTopic(final String topicName, final
GenericRecord... records) throws ExecutionException, InterruptedException {
final Properties producerProps = new Properties();
- producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9093");
+ producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaContainer.getBootstrapServers());
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
KafkaAvroSerializer.class.getName());
producerProps.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG,
"SASL_PLAINTEXT");
@@ -297,7 +279,7 @@ public class KafkaToS3IT {
);
final Map<String, String> kafkaServerConfig = Map.of(
- "Kafka Brokers", "localhost:9093",
+ "Kafka Brokers", kafkaContainer.getBootstrapServers(),
"Security Protocol", "SASL_PLAINTEXT",
"SASL Mechanism", "PLAIN",
"Username", SCRAM_USERNAME
@@ -359,7 +341,7 @@ public class KafkaToS3IT {
);
final Map<String, String> kafkaServerConfig = Map.of(
- "Kafka Brokers", "localhost:9093",
+ "Kafka Brokers", kafkaContainer.getBootstrapServers(),
"Security Protocol", "SASL_PLAINTEXT",
"SASL Mechanism", "PLAIN",
"Username", SCRAM_USERNAME
@@ -453,7 +435,7 @@ public class KafkaToS3IT {
produceAvroRecordsToTopic("avro-topic", record1, record2);
final Map<String, String> kafkaConnectionConfig = Map.of(
- "Kafka Brokers", "localhost:9093",
+ "Kafka Brokers", kafkaContainer.getBootstrapServers(),
"Security Protocol", "SASL_PLAINTEXT",
"SASL Mechanism", "PLAIN",
"Username", SCRAM_USERNAME,
@@ -524,7 +506,7 @@ public class KafkaToS3IT {
produceAvroRecordsToTopic("user-events", record1, record2, record3);
final Map<String, String> kafkaConnectionConfig = Map.of(
- "Kafka Brokers", "localhost:9093",
+ "Kafka Brokers", kafkaContainer.getBootstrapServers(),
"Security Protocol", "SASL_PLAINTEXT",
"SASL Mechanism", "PLAIN",
"Username", SCRAM_USERNAME,
@@ -627,7 +609,7 @@ public class KafkaToS3IT {
// Configure Connector to consume from JSON Kafka topic and write to
S3 in JSON format, but with an invalid S3 endpoint.
// This will cause the data to remain queued, since PutS3Object will
fail to write the data.
final Map<String, String> kafkaServerConfig = Map.of(
- "Kafka Brokers", "localhost:9093",
+ "Kafka Brokers", kafkaContainer.getBootstrapServers(),
"Security Protocol", "SASL_PLAINTEXT",
"SASL Mechanism", "PLAIN",
"Username", SCRAM_USERNAME
@@ -714,7 +696,7 @@ public class KafkaToS3IT {
);
final Map<String, String> kafkaConnectionWithSchemaRegistry = Map.of(
- "Kafka Brokers", "localhost:9093",
+ "Kafka Brokers", kafkaContainer.getBootstrapServers(),
"Security Protocol", "SASL_PLAINTEXT",
"SASL Mechanism", "PLAIN",
"Username", SCRAM_USERNAME,
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/AbstractKafkaBaseIT.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/AbstractKafkaBaseIT.java
index da2f9069ce7..7c780596dc6 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/AbstractKafkaBaseIT.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/AbstractKafkaBaseIT.java
@@ -29,7 +29,7 @@ import org.apache.nifi.serialization.RecordReaderFactory;
import org.apache.nifi.serialization.RecordSetWriterFactory;
import org.apache.nifi.util.TestRunner;
import org.junit.jupiter.api.BeforeAll;
-import org.testcontainers.kafka.ConfluentKafkaContainer;
+import org.testcontainers.kafka.KafkaContainer;
import org.testcontainers.utility.DockerImageName;
import java.time.Duration;
@@ -38,7 +38,7 @@ import java.util.Properties;
public abstract class AbstractKafkaBaseIT {
- protected static final String IMAGE_NAME = "confluentinc/cp-kafka:7.8.6";
// January 2026
+ protected static final String IMAGE_NAME =
System.getProperty("kafka.docker.image", "apache/kafka:4.3.1");
protected static final Integer MESSAGE_MAX_BYTES = 2097152;
@@ -53,11 +53,11 @@ public abstract class AbstractKafkaBaseIT {
protected static final Duration DURATION_POLL = Duration.ofSeconds(3);
- protected static final ConfluentKafkaContainer kafkaContainer;
+ protected static final KafkaContainer kafkaContainer;
// NIFI-11259 - single testcontainers Kafka instance needed for all module
integration tests
static {
- kafkaContainer = new
ConfluentKafkaContainer(DockerImageName.parse(IMAGE_NAME))
+ kafkaContainer = new KafkaContainer(DockerImageName.parse(IMAGE_NAME))
.withEnv(getEnvironmentIntegration());
kafkaContainer.start();
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/PublishKafkaAuthSaslPlaintextIT.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/PublishKafkaAuthSaslPlaintextIT.java
index de28ed5593a..11c1cf5279b 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/PublishKafkaAuthSaslPlaintextIT.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/PublishKafkaAuthSaslPlaintextIT.java
@@ -34,11 +34,10 @@ import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
-import org.testcontainers.kafka.ConfluentKafkaContainer;
+import org.testcontainers.kafka.KafkaContainer;
import org.testcontainers.utility.DockerImageName;
import java.util.Collections;
@@ -51,7 +50,6 @@ import java.util.UUID;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
-@Disabled("circle back to this")
@TestMethodOrder(MethodOrderer.MethodName.class)
public class PublishKafkaAuthSaslPlaintextIT {
private static final String SERVICE_ID =
Kafka3ConnectionService.class.getSimpleName();
@@ -61,27 +59,22 @@ public class PublishKafkaAuthSaslPlaintextIT {
private static final String USERNAME = "nifi";
private static final String PASSWORD = UUID.randomUUID().toString();
- private static ConfluentKafkaContainer kafka;
+ private static KafkaContainer kafka;
@BeforeAll
static void beforeAll() {
- kafka = new
ConfluentKafkaContainer(DockerImageName.parse(AbstractKafkaBaseIT.IMAGE_NAME))
+ kafka = new
KafkaContainer(DockerImageName.parse(AbstractKafkaBaseIT.IMAGE_NAME))
.withEnv(getEnvironmentSaslPlaintext());
kafka.start();
}
/**
* Environment to be provided to docker container to enable SASL
authentication.
- * <p>
- * Disable this test for now:
- * <ul>
- * <li><a
href="https://github.com/testcontainers/testcontainers-java/issues/3899";>Kafka
SASL mechanism</a></li>
- * <li><a
href="https://github.com/testcontainers/testcontainers-java/issues/6423";>Kafka
SASL mechanism</a></li>
- * </ul>
*/
private static Map<String, String> getEnvironmentSaslPlaintext() {
final Map<String, String> environment = new LinkedHashMap<>();
- environment.put("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
"BROKER:PLAINTEXT,PLAINTEXT:SASL_PLAINTEXT");
+ environment.put("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
"BROKER:PLAINTEXT,PLAINTEXT:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT");
+ environment.put("KAFKA_SASL_ENABLED_MECHANISMS", "PLAIN");
environment.put("KAFKA_LISTENER_NAME_PLAINTEXT_SASL_ENABLED_MECHANISMS",
"PLAIN");
environment.put("KAFKA_LISTENER_NAME_PLAINTEXT_PLAIN_SASL_JAAS_CONFIG",
String.format(
"%s required user_%s=\"%s\";",
PlainLoginModule.class.getName(), USERNAME, PASSWORD));
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceBaseIT.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceBaseIT.java
index 2960aa4bcb9..114d91bbf41 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceBaseIT.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceBaseIT.java
@@ -51,7 +51,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.TestInstance.Lifecycle;
import org.junit.jupiter.api.io.TempDir;
-import org.testcontainers.kafka.ConfluentKafkaContainer;
+import org.testcontainers.kafka.KafkaContainer;
import org.testcontainers.utility.DockerImageName;
import org.testcontainers.utility.MountableFile;
@@ -93,7 +93,7 @@ public class Kafka3ConnectionServiceBaseIT {
// This Base class executes its tests with Ssl off and Sasl off.
// There are subclasses which execute these same tests and enable Ssl or
Sasl
- public static final String IMAGE_NAME = "confluentinc/cp-kafka:7.8.6"; //
January 2026
+ public static final String IMAGE_NAME =
System.getProperty("kafka.docker.image", "apache/kafka:4.3.1");
private static final String DELIVERY_TIMEOUT_MS_KEY =
"delivery.timeout.ms";
private static final String DELIVERY_TIMEOUT_MS_VALUE = "60000";
@@ -121,7 +121,8 @@ public class Kafka3ConnectionServiceBaseIT {
private static final int POLLING_ATTEMPTS = 3;
- private static final Set<String> fileLocationNames =
Set.of("KAFKA_SSL_KEYSTORE_LOCATION", "KAFKA_SSL_TRUSTSTORE_LOCATION");
+ protected static final String CONTAINER_KEY_STORE_PATH =
"/tmp/kafka-keystore.p12";
+ protected static final String CONTAINER_TRUST_STORE_PATH =
"/tmp/kafka-truststore.p12";
protected static final String TEST_USERNAME = "nifi";
protected static final String TEST_PASSWORD = UUID.randomUUID().toString();
@@ -143,7 +144,7 @@ public class Kafka3ConnectionServiceBaseIT {
protected TestRunner runner;
- private ConfluentKafkaContainer kafkaContainer;
+ private KafkaContainer kafkaContainer;
private Kafka3ConnectionService service;
@@ -170,7 +171,7 @@ public class Kafka3ConnectionServiceBaseIT {
trustStore.store(outputStream, KEY_STORE_PASSWORD.toCharArray());
}
- kafkaContainer = new
ConfluentKafkaContainer(DockerImageName.parse(IMAGE_NAME));
+ kafkaContainer = new KafkaContainer(DockerImageName.parse(IMAGE_NAME));
initializeContainer();
kafkaContainer.start();
}
@@ -194,14 +195,18 @@ public class Kafka3ConnectionServiceBaseIT {
}
protected void initializeContainer() {
- Map<String, String> environment = getKafkaContainerConfigProperties();
+ final Map<String, String> environment =
getKafkaContainerConfigProperties();
kafkaContainer.withEnv(environment);
- // For each property which is a file location, copy the file into the
kafka container.
- environment.entrySet().stream()
- .filter(e -> fileLocationNames.contains(e.getKey()))
- .forEach(e -> kafkaContainer.withCopyFileToContainer(
- MountableFile.forHostPath(e.getValue()),
e.getValue()));
+ final String keyStoreLocation =
environment.get("KAFKA_SSL_KEYSTORE_LOCATION");
+ if (keyStoreLocation != null) {
+
kafkaContainer.withCopyFileToContainer(MountableFile.forHostPath(keyStorePath),
keyStoreLocation);
+ }
+
+ final String trustStoreLocation =
environment.get("KAFKA_SSL_TRUSTSTORE_LOCATION");
+ if (trustStoreLocation != null) {
+
kafkaContainer.withCopyFileToContainer(MountableFile.forHostPath(trustStorePath),
trustStoreLocation);
+ }
}
protected Map<String, String> getKafkaContainerConfigProperties() {
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceSSLIT.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceSSLIT.java
index fd1e64033c9..cb8356dff0a 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceSSLIT.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceSSLIT.java
@@ -47,11 +47,11 @@ public class Kafka3ConnectionServiceSSLIT extends
Kafka3ConnectionServiceBaseIT
protected Map<String, String> getKafkaContainerConfigProperties() {
final Map<String, String> properties = new
LinkedHashMap<>(super.getKafkaContainerConfigProperties());
properties.put("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
"BROKER:SSL,PLAINTEXT:SSL,CONTROLLER:SSL");
- properties.put("KAFKA_SSL_KEYSTORE_LOCATION", keyStorePath.toString());
+ properties.put("KAFKA_SSL_KEYSTORE_LOCATION",
CONTAINER_KEY_STORE_PATH);
properties.put("KAFKA_SSL_KEYSTORE_TYPE", keyStoreType);
properties.put("KAFKA_SSL_KEYSTORE_PASSWORD", KEY_STORE_PASSWORD);
properties.put("KAFKA_SSL_KEY_PASSWORD", KEY_PASSWORD);
- properties.put("KAFKA_SSL_TRUSTSTORE_LOCATION",
trustStorePath.toString());
+ properties.put("KAFKA_SSL_TRUSTSTORE_LOCATION",
CONTAINER_TRUST_STORE_PATH);
properties.put("KAFKA_SSL_TRUSTSTORE_TYPE", keyStoreType);
properties.put("KAFKA_SSL_TRUSTSTORE_PASSWORD", KEY_STORE_PASSWORD);
properties.put("KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND", "false");
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceSaslPlaintextIT.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceSaslPlaintextIT.java
index a64fdb81a46..94fad2dbe6e 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceSaslPlaintextIT.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceSaslPlaintextIT.java
@@ -31,6 +31,7 @@ public class Kafka3ConnectionServiceSaslPlaintextIT extends
Kafka3ConnectionServ
protected Map<String, String> getKafkaContainerConfigProperties() {
final Map<String, String> properties = new
LinkedHashMap<>(super.getKafkaContainerConfigProperties());
properties.put("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
"BROKER:PLAINTEXT,PLAINTEXT:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT");
+ properties.put("KAFKA_SASL_ENABLED_MECHANISMS", "PLAIN");
properties.put("KAFKA_LISTENER_NAME_PLAINTEXT_SASL_ENABLED_MECHANISMS",
"PLAIN");
properties.put("KAFKA_LISTENER_NAME_PLAINTEXT_PLAIN_SASL_JAAS_CONFIG",
getJaasConfigKafkaContainer(TEST_USERNAME, TEST_PASSWORD));
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/KafkaConnectivityIT.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/KafkaConnectivityIT.java
index 0eb1730c88c..e1c9a861cfd 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/KafkaConnectivityIT.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/KafkaConnectivityIT.java
@@ -32,7 +32,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.testcontainers.kafka.ConfluentKafkaContainer;
+import org.testcontainers.kafka.KafkaContainer;
import org.testcontainers.utility.DockerImageName;
import java.time.Duration;
@@ -50,11 +50,11 @@ public class KafkaConnectivityIT {
private static final String TEST_TOPIC = "nifi-" +
System.currentTimeMillis();
- private static ConfluentKafkaContainer kafka;
+ private static KafkaContainer kafka;
@BeforeAll
static void beforeAll() {
- kafka = new
ConfluentKafkaContainer(DockerImageName.parse(Kafka3ConnectionServiceBaseIT.IMAGE_NAME));
+ kafka = new
KafkaContainer(DockerImageName.parse(Kafka3ConnectionServiceBaseIT.IMAGE_NAME));
kafka.start();
}
diff --git a/pom.xml b/pom.xml
index d6631d759b9..3aee00c7556 100644
--- a/pom.xml
+++ b/pom.xml
@@ -151,6 +151,7 @@
<!-- Data formats and serialization -->
<kafka-clients.version>4.3.1</kafka-clients.version>
+ <kafka.docker.image>apache/kafka:4.3.1</kafka.docker.image>
<avro.version>1.12.1</avro.version>
<calcite.version>1.42.0</calcite.version>
<com.github.luben.zstd-jni.version>1.5.7-12</com.github.luben.zstd-jni.version>
@@ -1139,6 +1140,11 @@
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-failsafe-plugin</artifactId>
+ <configuration>
+ <systemPropertyVariables>
+
<kafka.docker.image>${kafka.docker.image}</kafka.docker.image>
+ </systemPropertyVariables>
+ </configuration>
<executions>
<execution>
<goals>
