You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2026/08/18 11:26:55 UTC
(kafka) branch trunk updated: KAFKA-19042 Rewrite SaslPlaintextConsumerTest from Scala to Java (#22918)
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 1e0a3b59222 KAFKA-19042 Rewrite SaslPlaintextConsumerTest from Scala to Java (#22918)
1e0a3b59222 is described below
commit 1e0a3b5922280aedfb2b1bbc42dbac5367545328
Author: Ken Huang <s7...@gmail.com>
AuthorDate: Tue Aug 18 19:26:45 2026 +0800
KAFKA-19042 Rewrite SaslPlaintextConsumerTest from Scala to Java (#22918)
Rewrite SaslPlaintextConsumerTest from Scala to Java
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
build.gradle | 21 +++
.../import-control-clients-integration-tests.xml | 1 +
.../consumer/SaslPlaintextConsumerTest.java | 193 +++++++++++++++++++++
.../kafka/api/SaslPlaintextConsumerTest.scala | 35 ----
.../apache/kafka/common/test/ClusterInstance.java | 29 +++-
.../kafka/common/test/KafkaClusterTestKit.java | 30 +++-
.../org/apache/kafka/common/test/TestUtils.java | 2 +-
.../common/test/junit/ClusterTestExtensions.java | 7 +-
.../test/junit/RaftClusterInvocationContext.java | 19 +-
9 files changed, 276 insertions(+), 61 deletions(-)
diff --git a/build.gradle b/build.gradle
index 39b63f2d2dd..cc33dae7de4 100644
--- a/build.gradle
+++ b/build.gradle
@@ -2204,6 +2204,14 @@ project(':clients:clients-integration-tests') {
archivesName = "kafka-clients-integration-tests"
}
+ configurations {
+ // To prevent a UniqueResourceException due the same resource existing in both
+ // org.apache.directory.api/api-all and org.apache.directory.api/api-ldap-schema-data
+ testImplementation.exclude module: 'api-ldap-schema-data'
+ testCompileClasspath.exclude group: 'org.bouncycastle', module: 'bcprov-jdk15on'
+ testRuntimeClasspath.exclude group: 'org.bouncycastle', module: 'bcprov-jdk15on'
+ }
+
dependencies {
testImplementation libs.metrics
testImplementation libs.slf4jApi
@@ -2212,6 +2220,7 @@ project(':clients:clients-integration-tests') {
testImplementation project(':metadata')
testImplementation project(':raft')
testImplementation project(':server')
+ testImplementation testFixtures(project(':server'))
testImplementation project(':storage')
testImplementation testFixtures(project(':storage'))
testImplementation testFixtures(project(':clients'))
@@ -2222,6 +2231,18 @@ project(':clients:clients-integration-tests') {
implementation project(':transaction-coordinator')
testImplementation project(':test-common:test-common-util')
+ testImplementation libs.bcpkix
+ testImplementation(libs.apacheda) {
+ exclude group: 'xml-apis', module: 'xml-apis'
+ exclude module: 'mina-core'
+ }
+ testImplementation libs.apachedsCoreApi
+ testImplementation libs.apachedsInterceptorKerberos
+ testImplementation libs.apachedsProtocolShared
+ testImplementation libs.apachedsProtocolKerberos
+ testImplementation libs.apachedsProtocolLdap
+ testImplementation libs.apachedsLdifPartition
+ testImplementation libs.apachedsJdbmPartition
testImplementation libs.junitJupiter
testRuntimeOnly runtimeTestLibs
diff --git a/checkstyle/import-control-clients-integration-tests.xml b/checkstyle/import-control-clients-integration-tests.xml
index 2044bd7538d..4acc1094ffb 100644
--- a/checkstyle/import-control-clients-integration-tests.xml
+++ b/checkstyle/import-control-clients-integration-tests.xml
@@ -26,6 +26,7 @@
<allow pkg="org.apache.kafka"/>
<allow pkg="kafka"/>
<allow pkg="org.slf4j"/>
+ <allow pkg="javax.security.auth.login"/>
<allow pkg="com.yammer.metrics.core" />
diff --git a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/SaslPlaintextConsumerTest.java b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/SaslPlaintextConsumerTest.java
new file mode 100644
index 00000000000..a0747ec0a1e
--- /dev/null
+++ b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/SaslPlaintextConsumerTest.java
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer;
+
+import org.apache.kafka.clients.ClientsTestUtils;
+import org.apache.kafka.common.security.JaasUtils;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.security.authenticator.LoginManager;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.security.JaasTestUtils;
+import org.apache.kafka.security.minikdc.MiniKdc;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+
+import java.io.File;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+import javax.security.auth.login.Configuration;
+
+import static org.apache.kafka.clients.ClientsTestUtils.BaseConsumerTestcase.testClusterResourceListener;
+import static org.apache.kafka.clients.ClientsTestUtils.BaseConsumerTestcase.testCoordinatorFailover;
+import static org.apache.kafka.clients.ClientsTestUtils.BaseConsumerTestcase.testSimpleConsumption;
+import static org.apache.kafka.clients.CommonClientConfigs.MAX_POLL_INTERVAL_MS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_PROTOCOL_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.clients.consumer.SaslPlaintextConsumerTest.MECHANISMS;
+import static org.apache.kafka.common.config.internals.BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG;
+import static org.apache.kafka.common.config.internals.BrokerSecurityConfigs.SASL_MECHANISM_INTER_BROKER_PROTOCOL_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG;
+import static org.apache.kafka.security.JaasTestUtils.KAFKA_CLIENT_PRINCIPAL_UNQUALIFIED_NAME;
+import static org.apache.kafka.security.JaasTestUtils.KAFKA_CLIENT_PRINCIPAL_UNQUALIFIED_NAME_2;
+import static org.apache.kafka.security.JaasTestUtils.KAFKA_SERVER_PRINCIPAL_UNQUALIFIED_NAME;
+
+@ClusterTestDefaults(
+ types = {Type.KRAFT},
+ brokers = ClientsTestUtils.BaseConsumerTestcase.BROKER_COUNT,
+ serverProperties = {
+ @ClusterConfigProperty(key = OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, value = "100"),
+ @ClusterConfigProperty(key = SASL_ENABLED_MECHANISMS_CONFIG, value = MECHANISMS),
+ @ClusterConfigProperty(key = SASL_MECHANISM_INTER_BROKER_PROTOCOL_CONFIG, value = MECHANISMS),
+ }
+)
+public class SaslPlaintextConsumerTest {
+
+ private final ClusterInstance cluster;
+ public static final String MECHANISMS = "GSSAPI";
+ private static Properties kdcConf = MiniKdc.createConfig();
+ private static MiniKdc kdc;
+
+ public SaslPlaintextConsumerTest(ClusterInstance cluster) {
+ this.cluster = cluster;
+ }
+
+ @BeforeAll
+ public static void setup() throws Exception {
+ // Important if tests leak consumers, producers or brokers
+ LoginManager.closeAll();
+ File workingDir = TestUtils.tempDirectory();
+ Optional<File> serverKeytabFile = Optional.of(TestUtils.tempFile());
+ Optional<File> clientKeytabFile = Optional.of(TestUtils.tempFile());
+ List<JaasTestUtils.JaasSection> jaasSections = List.of(
+ JaasTestUtils.kafkaServerSection(
+ JaasTestUtils.KAFKA_SERVER_CONTEXT_NAME,
+ List.of(MECHANISMS),
+ serverKeytabFile
+ ), JaasTestUtils.kafkaClientSection(
+ Optional.of(MECHANISMS),
+ clientKeytabFile
+ )
+ );
+ // init MiniKdc
+ kdc = new MiniKdc(kdcConf, workingDir);
+ kdc.start();
+ kdc.createPrincipal(
+ serverKeytabFile.get(),
+ List.of(KAFKA_SERVER_PRINCIPAL_UNQUALIFIED_NAME + "/localhost")
+ );
+ kdc.createPrincipal(
+ clientKeytabFile.get(),
+ List.of(KAFKA_CLIENT_PRINCIPAL_UNQUALIFIED_NAME, KAFKA_CLIENT_PRINCIPAL_UNQUALIFIED_NAME_2)
+ );
+ // writeJaasConfigurationToFile
+ File file = JaasTestUtils.writeJaasContextsToFile(jaasSections);
+ System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, file.getAbsolutePath());
+ // This will cause a reload of the Configuration singleton when `getConfiguration` is called
+ Configuration.setConfiguration(null);
+ }
+
+ @BeforeEach
+ public void beforeEach() throws InterruptedException {
+ cluster.createTopic(ClientsTestUtils.BaseConsumerTestcase.TOPIC, 2, (short) ClientsTestUtils.BaseConsumerTestcase.BROKER_COUNT);
+ }
+
+ @AfterAll
+ public static void teardown() {
+ if (kdc != null)
+ kdc.stop();
+ // Important if tests leak consumers, producers or brokers
+ LoginManager.closeAll();
+ System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM);
+ Configuration.setConfiguration(null);
+ }
+
+ @ClusterTest(
+ brokerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT
+ )
+ public void testClassicConsumerSimpleConsumption() throws InterruptedException {
+ testSimpleConsumption(cluster, Map.of(
+ GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name().toLowerCase(Locale.ROOT))
+ );
+ }
+
+ @ClusterTest(
+ brokerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT
+ )
+ public void testAsyncConsumerSimpleConsumption() throws InterruptedException {
+ testSimpleConsumption(cluster, Map.of(
+ GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name().toLowerCase(Locale.ROOT))
+ );
+ }
+
+ @ClusterTest(
+ brokerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT
+ )
+ public void testClassicConsumerClusterResourceListener() throws InterruptedException {
+ testClusterResourceListener(cluster, Map.of(
+ GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name().toLowerCase(Locale.ROOT))
+ );
+ }
+
+ @ClusterTest(
+ brokerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT
+ )
+ public void testAsyncConsumerClusterResourceListener() throws InterruptedException {
+ testClusterResourceListener(cluster, Map.of(
+ GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name().toLowerCase(Locale.ROOT))
+ );
+ }
+
+ @ClusterTest(
+ brokerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT
+ )
+ public void testClassicConsumerCoordinatorFailover() throws InterruptedException {
+ Map<String, Object> config = Map.of(
+ GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name().toLowerCase(Locale.ROOT),
+ SESSION_TIMEOUT_MS_CONFIG, 5001,
+ HEARTBEAT_INTERVAL_MS_CONFIG, 1000,
+ // Use higher poll timeout to avoid consumer leaving the group due to timeout
+ MAX_POLL_INTERVAL_MS_CONFIG, 15000
+ );
+ testCoordinatorFailover(cluster, config);
+ }
+
+ @ClusterTest(
+ brokerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT
+ )
+ public void testAsyncConsumeCoordinatorFailover() throws InterruptedException {
+ Map<String, Object> config = Map.of(
+ GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name().toLowerCase(Locale.ROOT),
+ // Use higher poll timeout to avoid consumer leaving the group due to timeout
+ MAX_POLL_INTERVAL_MS_CONFIG, 15000
+ );
+ testCoordinatorFailover(cluster, config);
+ }
+}
diff --git a/core/src/test/scala/integration/kafka/api/SaslPlaintextConsumerTest.scala b/core/src/test/scala/integration/kafka/api/SaslPlaintextConsumerTest.scala
deleted file mode 100644
index ff2b7952cd3..00000000000
--- a/core/src/test/scala/integration/kafka/api/SaslPlaintextConsumerTest.scala
+++ /dev/null
@@ -1,35 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE
- * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file
- * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
- * License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
- * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations under the License.
- */
-package kafka.api
-
-import org.apache.kafka.common.security.auth.SecurityProtocol
-import org.apache.kafka.security.JaasTestUtils
-import org.junit.jupiter.api.{AfterEach, BeforeEach, TestInfo, Timeout}
-
-@Timeout(600)
-class SaslPlaintextConsumerTest extends BaseConsumerTest with SaslSetup {
- override protected def securityProtocol = SecurityProtocol.SASL_PLAINTEXT
-
- @BeforeEach
- override def setUp(testInfo: TestInfo): Unit = {
- startSasl(jaasSections(Seq("GSSAPI"), Some("GSSAPI"), JaasTestUtils.KAFKA_SERVER_CONTEXT_NAME))
- super.setUp(testInfo)
- }
-
- @AfterEach
- override def tearDown(): Unit = {
- super.tearDown()
- closeSasl()
- }
-
-}
diff --git a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java
index dacf3f27241..3d28aa32727 100644
--- a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java
+++ b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java
@@ -39,6 +39,7 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.acl.AccessControlEntry;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
@@ -203,19 +204,31 @@ public interface ClusterInstance {
default Map<String, Object> setClientSaslConfig(Map<String, Object> configs) {
Map<String, Object> props = new HashMap<>(configs);
if (config().brokerSecurityProtocol() == SecurityProtocol.SASL_PLAINTEXT) {
+ String mechanism = clientSaslMechanism();
props.putIfAbsent(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_PLAINTEXT.name);
- props.putIfAbsent(SaslConfigs.SASL_MECHANISM, "PLAIN");
- props.putIfAbsent(
- SaslConfigs.SASL_JAAS_CONFIG,
- String.format(
- "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";",
- JaasUtils.KAFKA_PLAIN_ADMIN, JaasUtils.KAFKA_PLAIN_ADMIN_PASSWORD
- )
- );
+ props.putIfAbsent(SaslConfigs.SASL_MECHANISM, mechanism);
+ if (mechanism.equalsIgnoreCase("PLAIN")) {
+ props.putIfAbsent(
+ SaslConfigs.SASL_JAAS_CONFIG,
+ String.format(
+ "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";",
+ JaasUtils.KAFKA_PLAIN_ADMIN, JaasUtils.KAFKA_PLAIN_ADMIN_PASSWORD
+ )
+ );
+ }
}
return props;
}
+ default String clientSaslMechanism() {
+ String mechanisms = config().serverProperties().get(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG);
+ if (mechanisms == null) return "PLAIN";
+ for (String m : mechanisms.split(",")) {
+ if (m.trim().equalsIgnoreCase("PLAIN")) return "PLAIN";
+ }
+ return mechanisms.split(",")[0].trim();
+ }
+
Map<String, Object> setClientSslConfig(Map<String, Object> configs);
default Admin admin(Map<String, Object> configs) {
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 ec2fa992d9c..eb68ef03288 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
@@ -240,25 +240,39 @@ public class KafkaClusterTestKit implements AutoCloseable {
props.putIfAbsent(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, "PLAIN");
props.putIfAbsent(BrokerSecurityConfigs.SASL_MECHANISM_INTER_BROKER_PROTOCOL_CONFIG, "PLAIN");
props.putIfAbsent(KRaftConfigs.SASL_MECHANISM_CONTROLLER_PROTOCOL_CONFIG, "PLAIN");
- props.putIfAbsent(ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG, StandardAuthorizer.class.getName());
- props.putIfAbsent(StandardAuthorizer.ALLOW_EVERYONE_IF_NO_ACL_IS_FOUND_CONFIG, "false");
- props.putIfAbsent(StandardAuthorizer.SUPER_USERS_CONFIG, "User:" + JaasUtils.KAFKA_PLAIN_ADMIN);
+ maybeSetPlainAuthorizerProps(props);
} else if (securityProtocol.equals(SecurityProtocol.SASL_SSL.name)) {
props.putIfAbsent(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, "PLAIN");
props.putIfAbsent(BrokerSecurityConfigs.SASL_MECHANISM_INTER_BROKER_PROTOCOL_CONFIG, "PLAIN");
props.putIfAbsent(KRaftConfigs.SASL_MECHANISM_CONTROLLER_PROTOCOL_CONFIG, "PLAIN");
- props.putIfAbsent(ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG, StandardAuthorizer.class.getName());
- props.putIfAbsent(StandardAuthorizer.ALLOW_EVERYONE_IF_NO_ACL_IS_FOUND_CONFIG, "false");
- props.putIfAbsent(StandardAuthorizer.SUPER_USERS_CONFIG, "User:" + JaasUtils.KAFKA_PLAIN_ADMIN);
+ maybeSetPlainAuthorizerProps(props);
sslConfig.forEach(props::putIfAbsent);
} else if (securityProtocol.equals(SecurityProtocol.SSL.name)) {
sslConfig.forEach(props::putIfAbsent);
}
}
+ private void maybeSetPlainAuthorizerProps(Map<String, Object> props) {
+ if (isPlainSaslMechanism(props)) {
+ props.putIfAbsent(ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG, StandardAuthorizer.class.getName());
+ props.putIfAbsent(StandardAuthorizer.ALLOW_EVERYONE_IF_NO_ACL_IS_FOUND_CONFIG, "false");
+ props.putIfAbsent(StandardAuthorizer.SUPER_USERS_CONFIG, "User:" + JaasUtils.KAFKA_PLAIN_ADMIN);
+ }
+ }
+
+ private boolean isPlainSaslMechanism(Map<String, Object> props) {
+ Object mechanism = props.get(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG);
+ if (mechanism == null) return true;
+ for (String m : mechanism.toString().split(",")) {
+ if (m.trim().equalsIgnoreCase("PLAIN")) return true;
+ }
+ return false;
+ }
+
private Optional<File> maybeSetupJaasFile() throws Exception {
- if (brokerSecurityProtocol.equals(SecurityProtocol.SASL_PLAINTEXT.name) ||
- brokerSecurityProtocol.equals(SecurityProtocol.SASL_SSL.name)) {
+ if ((brokerSecurityProtocol.equals(SecurityProtocol.SASL_PLAINTEXT.name) ||
+ brokerSecurityProtocol.equals(SecurityProtocol.SASL_SSL.name)) &&
+ isPlainSaslMechanism(configProps)) {
File file = JaasUtils.writeJaasContextsToFile(Set.of(
new JaasUtils.JaasSection(JaasUtils.KAFKA_SERVER_CONTEXT_NAME,
List.of(
diff --git a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestUtils.java b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestUtils.java
index 16ed515ccd0..4f6a78891c9 100644
--- a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestUtils.java
+++ b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestUtils.java
@@ -42,7 +42,7 @@ class TestUtils {
/* A consistent random number generator to make tests repeatable */
public static final Random SEEDED_RANDOM = new Random(192348092834L);
-
+
public static final String LETTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
public static final String DIGITS = "0123456789";
public static final String LETTERS_AND_DIGITS = LETTERS + DIGITS;
diff --git a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/ClusterTestExtensions.java b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/ClusterTestExtensions.java
index 8bb1ae1509e..26f2927e096 100644
--- a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/ClusterTestExtensions.java
+++ b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/ClusterTestExtensions.java
@@ -109,11 +109,16 @@ public class ClusterTestExtensions implements TestTemplateInvocationContextProvi
private static final String PROCESS_REAPER_THREAD_PREFIX = "process reaper";
private static final String RMI_THREAD_PREFIX = "RMI";
private static final String JDK_INTERNAL_CLEANERIMPL_THREAD_PREFIX = "Cleaner-";
+ // Apache MINA I/O threads spawned by MiniKdc (via ApacheDS). They live for the KDC's lifetime,
+ // which a SASL/GSSAPI test manages across all its @ClusterTest methods (started in @BeforeAll,
+ // stopped in @AfterAll), so they must not be treated as per-test leaks.
+ private static final String MINA_NIO_PROCESSOR_THREAD_PREFIX = "NioProcessor";
private static final String DETECT_THREAD_LEAK_KEY = "detectThreadLeak";
private static final Set<String> SKIPPED_THREAD_PREFIX = Set.of(METRICS_METER_TICK_THREAD_PREFIX, SCALA_THREAD_PREFIX,
FORK_JOIN_POOL_THREAD_PREFIX, JUNIT_THREAD_PREFIX, ATTACH_LISTENER_THREAD_PREFIX, PROCESS_REAPER_THREAD_PREFIX,
- RMI_THREAD_PREFIX, SystemTimer.SYSTEM_TIMER_THREAD_PREFIX, JDK_INTERNAL_CLEANERIMPL_THREAD_PREFIX);
+ RMI_THREAD_PREFIX, SystemTimer.SYSTEM_TIMER_THREAD_PREFIX, JDK_INTERNAL_CLEANERIMPL_THREAD_PREFIX,
+ MINA_NIO_PROCESSOR_THREAD_PREFIX);
@Override
public boolean supportsTestTemplate(ExtensionContext context) {
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 75757b9f4e1..772e7c49f86 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
@@ -186,15 +186,18 @@ public class RaftClusterInvocationContext implements TestTemplateInvocationConte
public Map<String, Object> setClientSslConfig(Map<String, Object> configs) {
Map<String, Object> props = new HashMap<>(configs);
if (config().brokerSecurityProtocol() == SecurityProtocol.SASL_SSL) {
+ String mechanism = clientSaslMechanism();
props.putIfAbsent(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_SSL.name);
- props.putIfAbsent(SaslConfigs.SASL_MECHANISM, "PLAIN");
- props.putIfAbsent(
- SaslConfigs.SASL_JAAS_CONFIG,
- String.format(
- "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";",
- JaasUtils.KAFKA_PLAIN_ADMIN, JaasUtils.KAFKA_PLAIN_ADMIN_PASSWORD
- )
- );
+ props.putIfAbsent(SaslConfigs.SASL_MECHANISM, mechanism);
+ if (mechanism.equalsIgnoreCase("PLAIN")) {
+ props.putIfAbsent(
+ SaslConfigs.SASL_JAAS_CONFIG,
+ String.format(
+ "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";",
+ JaasUtils.KAFKA_PLAIN_ADMIN, JaasUtils.KAFKA_PLAIN_ADMIN_PASSWORD
+ )
+ );
+ }
if (clusterTestKit.sslManager() != null) {
props.putAll(clusterTestKit.sslManager().createClientSslConfig());
props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");