This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new fdc59be8c84 KAFKA-18448 SSL support for ClusterTestExtensions (#22184)
fdc59be8c84 is described below
commit fdc59be8c84af1424b2f51a2df81386927f01a90
Author: Igor Soarez <[email protected]>
AuthorDate: Sun Jul 19 08:22:34 2026 +0100
KAFKA-18448 SSL support for ClusterTestExtensions (#22184)
Mixed protocols - SSL only between brokers and clients, not between
brokers and the quorum or vice versa is not yet supported.
Reviewers: Chia-Ping Tsai <[email protected]>, Ken Huang
<[email protected]>
---
.../kafka/common/test/KafkaClusterTestKit.java | 2 ++
.../org/apache/kafka/common/test/TestKitNodes.java | 17 ++++++---
.../test/junit/RaftClusterInvocationContext.java | 6 ++++
.../apache/kafka/common/test/TestKitNodeTest.java | 23 +++++++-----
.../test/junit/ClusterTestExtensionsTest.java | 41 ++++++++++++++++++++++
5 files changed, 77 insertions(+), 12 deletions(-)
diff --git
a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java
b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java
index 747818a9864..eec40e2bb82 100644
---
a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java
+++
b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java
@@ -251,6 +251,8 @@ public class KafkaClusterTestKit implements AutoCloseable {
props.putIfAbsent(StandardAuthorizer.ALLOW_EVERYONE_IF_NO_ACL_IS_FOUND_CONFIG,
"false");
props.putIfAbsent(StandardAuthorizer.SUPER_USERS_CONFIG,
"User:" + JaasUtils.KAFKA_PLAIN_ADMIN);
sslConfig.forEach(props::putIfAbsent);
+ } else if (securityProtocol.equals(SecurityProtocol.SSL.name)) {
+ sslConfig.forEach(props::putIfAbsent);
}
}
diff --git
a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestKitNodes.java
b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestKitNodes.java
index b1c7fc00e88..573ff79ef92 100644
---
a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestKitNodes.java
+++
b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestKitNodes.java
@@ -162,10 +162,19 @@ public class TestKitNodes {
if (numDisksPerBroker <= 0) {
throw new IllegalArgumentException("Invalid value for
numDisksPerBroker");
}
- // TODO: remove this assertion after
https://issues.apache.org/jira/browse/KAFKA-16680 is finished
- if ((brokerSecurityProtocol != SecurityProtocol.PLAINTEXT &&
brokerSecurityProtocol != SecurityProtocol.SASL_PLAINTEXT &&
brokerSecurityProtocol != SecurityProtocol.SASL_SSL) ||
- (controllerSecurityProtocol != SecurityProtocol.PLAINTEXT &&
controllerSecurityProtocol != SecurityProtocol.SASL_PLAINTEXT &&
brokerSecurityProtocol != SecurityProtocol.SASL_SSL)) {
- throw new IllegalArgumentException("Currently only support
PLAINTEXT / SASL_PLAINTEXT / SASL_SSL security protocol");
+ var supportedProtocols = List.of(
+ SecurityProtocol.PLAINTEXT,
+ SecurityProtocol.SASL_PLAINTEXT,
+ SecurityProtocol.SASL_SSL,
+ SecurityProtocol.SSL
+ );
+ if (!supportedProtocols.contains(brokerSecurityProtocol) ||
!supportedProtocols.contains(controllerSecurityProtocol)) {
+ throw new IllegalArgumentException(String.format("Currently
only support %s security protocol",
+ supportedProtocols.stream().map(sp ->
sp.name).collect(Collectors.joining(" / ")))
+ );
+ }
+ if ((brokerSecurityProtocol == SecurityProtocol.SSL) !=
(controllerSecurityProtocol == SecurityProtocol.SSL)) {
+ throw new IllegalArgumentException("Mixed broker and
controller SSL security protocol configurations are not yet supported");
}
if (baseDirectory == null) {
this.baseDirectory = TestUtils.tempDirectory().toPath();
diff --git
a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java
b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java
index b5d93066232..699adaa3ec7 100644
---
a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java
+++
b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java
@@ -199,6 +199,12 @@ public class RaftClusterInvocationContext implements
TestTemplateInvocationConte
props.putAll(clusterTestKit.sslManager().createClientSslConfig());
props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
}
+ } else if (config().brokerSecurityProtocol() ==
SecurityProtocol.SSL) {
+
props.putIfAbsent(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG,
SecurityProtocol.SSL.name);
+ if (clusterTestKit.sslManager() != null) {
+
props.putAll(clusterTestKit.sslManager().createClientSslConfig());
+
props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
+ }
}
return props;
}
diff --git
a/test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/TestKitNodeTest.java
b/test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/TestKitNodeTest.java
index 4a9dcaf9e5e..adbb5871a46 100644
---
a/test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/TestKitNodeTest.java
+++
b/test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/TestKitNodeTest.java
@@ -32,14 +32,21 @@ public class TestKitNodeTest {
@ParameterizedTest
@EnumSource(SecurityProtocol.class)
public void testSecurityProtocol(SecurityProtocol securityProtocol) {
- if (securityProtocol != SecurityProtocol.PLAINTEXT && securityProtocol
!= SecurityProtocol.SASL_PLAINTEXT && securityProtocol !=
SecurityProtocol.SASL_SSL) {
- assertEquals("Currently only support PLAINTEXT / SASL_PLAINTEXT /
SASL_SSL security protocol",
- assertThrows(IllegalArgumentException.class,
- () -> new
TestKitNodes.Builder().setBrokerSecurityProtocol(securityProtocol).build()).getMessage());
- assertEquals("Currently only support PLAINTEXT / SASL_PLAINTEXT /
SASL_SSL security protocol",
- assertThrows(IllegalArgumentException.class,
- () -> new
TestKitNodes.Builder().setControllerSecurityProtocol(securityProtocol).build()).getMessage());
- }
+ // All security protocols are supported when broker and controller
configurations are consistent
+ new TestKitNodes.Builder()
+ .setBrokerSecurityProtocol(securityProtocol)
+ .setControllerSecurityProtocol(securityProtocol)
+ .build();
+ }
+
+ @Test
+ public void testMixedSslSecurityProtocolIsNotSupported() {
+ assertEquals("Mixed broker and controller SSL security protocol
configurations are not yet supported",
+ assertThrows(IllegalArgumentException.class,
+ () -> new
TestKitNodes.Builder().setBrokerSecurityProtocol(SecurityProtocol.SSL).build()).getMessage());
+ assertEquals("Mixed broker and controller SSL security protocol
configurations are not yet supported",
+ assertThrows(IllegalArgumentException.class,
+ () -> new
TestKitNodes.Builder().setControllerSecurityProtocol(SecurityProtocol.SSL).build()).getMessage());
}
@Test
diff --git
a/test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/junit/ClusterTestExtensionsTest.java
b/test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/junit/ClusterTestExtensionsTest.java
index e2862938f82..e6fa0d5d8c4 100644
---
a/test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/junit/ClusterTestExtensionsTest.java
+++
b/test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/junit/ClusterTestExtensionsTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.common.test.junit;
+import kafka.network.SocketServer;
import kafka.server.ControllerServer;
import kafka.server.KafkaBroker;
@@ -68,6 +69,7 @@ import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -77,6 +79,8 @@ import java.util.Set;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
+import java.util.stream.Collectors;
import scala.jdk.javaapi.CollectionConverters;
@@ -595,4 +599,41 @@ public class ClusterTestExtensionsTest {
assertInstanceOf(SaslAuthenticationException.class,
exception.getCause());
}
}
+
+ @ClusterTest(
+ types = {Type.KRAFT, Type.CO_KRAFT},
+ brokerSecurityProtocol = SecurityProtocol.SSL,
+ controllerSecurityProtocol = SecurityProtocol.SSL,
+ serverProperties = {
+ @ClusterConfigProperty(key =
GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key =
GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
+ }
+ )
+ public void testSsl(ClusterInstance clusterInstance) throws
InterruptedException, ExecutionException {
+ KafkaBroker broker =
clusterInstance.brokers().values().iterator().next();
+ ControllerServer controller =
clusterInstance.controllers().values().iterator().next();
+ Function<SocketServer, String> endpoints = socketServer ->
Collections.list(socketServer.dataPlaneAcceptors().keys())
+ .stream().map(endpoint -> String.format("%s:%s",
endpoint.listener(),
endpoint.securityProtocol())).collect(Collectors.joining(","));
+ assertEquals("EXTERNAL:SSL", endpoints.apply(broker.socketServer()));
+ assertEquals("CONTROLLER:SSL",
endpoints.apply(controller.socketServer()));
+
+ String topic = "ssl-topic";
+ clusterInstance.createTopic(topic, 1, (short) 1);
+ try (Admin admin = clusterInstance.admin()) {
+ Set<String> topics = admin.listTopics().names().get();
+ assertTrue(topics.contains(topic), String.format("%s not included
in %s", topic, topics));
+ }
+
+ try (Producer<byte[], byte[]> producer = clusterInstance.producer()) {
+ producer.send(new ProducerRecord<>(topic, Utils.utf8("key"),
Utils.utf8("value"))).get();
+ producer.flush();
+ }
+ try (Consumer<byte[], byte[]> consumer = clusterInstance.consumer()) {
+ consumer.subscribe(List.of(topic));
+ RaftClusterInvocationContext.waitForCondition(() -> {
+ ConsumerRecords<byte[], byte[]> records =
consumer.poll(Duration.ofMillis(100));
+ return records.count() == 1;
+ }, "Failed to receive message");
+ }
+ }
}
