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 d366f759be8 MINOR: Rewrite SaslClientsWithInvalidCredentialsTest and
rename to ConsumerGroupCommandSaslAuthenticationTest (#22582)
d366f759be8 is described below
commit d366f759be87c3f211c2b5ebaedff98a8af59d3a
Author: majialong <[email protected]>
AuthorDate: Sun Jul 19 06:36:22 2026 +0800
MINOR: Rewrite SaslClientsWithInvalidCredentialsTest and rename to
ConsumerGroupCommandSaslAuthenticationTest (#22582)
Migrate `SaslClientsWithInvalidCredentialsTest` to `ClusterInstance` and
rename to `ConsumerGroupCommandSaslAuthenticationTest`.
Reviewers: Chia-Ping Tsai <[email protected]>
---
build.gradle | 12 --
...ConsumerGroupCommandSaslAuthenticationTest.java | 180 +++++++++++++++++++
.../SaslClientsWithInvalidCredentialsTest.java | 193 ---------------------
3 files changed, 180 insertions(+), 205 deletions(-)
diff --git a/build.gradle b/build.gradle
index 96c5f4a370a..2edb462b872 100644
--- a/build.gradle
+++ b/build.gradle
@@ -2647,12 +2647,6 @@ project(':tools') {
configurations {
releaseOnly
- // ApacheDS 2.0.0-M24 pulls in the stale bcprov-jdk15on:1.56, which ships
the same
- // RosstandartObjectIdentifiers class as the modern bcprov-jdk18on we
already depend on
- // but is missing fields referenced by bcpkix-jdk18on:1.84. When
IntelliJ's test runner
- // orders the old JAR first, BC provider registration fails with
NoSuchFieldError. Drop it.
- testCompileClasspath.exclude group: 'org.bouncycastle', module:
'bcprov-jdk15on'
- testRuntimeClasspath.exclude group: 'org.bouncycastle', module:
'bcprov-jdk15on'
}
dependencies {
@@ -2710,12 +2704,6 @@ project(':tools') {
testImplementation(libs.jfreechart) {
exclude group: 'junit', module: 'junit'
}
- testImplementation libs.apachedsCoreApi
- testImplementation libs.apachedsInterceptorKerberos
- testImplementation libs.apachedsProtocolShared
- testImplementation libs.apachedsProtocolKerberos
- testImplementation libs.apachedsProtocolLdap
- testImplementation libs.apachedsLdifPartition
testImplementation testLog4j2Libs
testRuntimeOnly runtimeTestLibs
diff --git
a/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommandSaslAuthenticationTest.java
b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommandSaslAuthenticationTest.java
new file mode 100644
index 00000000000..ea03a1aa1c6
--- /dev/null
+++
b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommandSaslAuthenticationTest.java
@@ -0,0 +1,180 @@
+/*
+ * 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.tools.consumer.group;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterUserScramCredentialsResult;
+import org.apache.kafka.clients.admin.ScramCredentialInfo;
+import org.apache.kafka.clients.admin.ScramMechanism;
+import org.apache.kafka.clients.admin.UserScramCredentialUpsertion;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.GroupProtocol;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.errors.SaslAuthenticationException;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+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.test.TestUtils;
+
+import org.junit.jupiter.api.function.Executable;
+
+import java.io.File;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+
+import static
org.apache.kafka.clients.CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG;
+import static
org.apache.kafka.clients.CommonClientConfigs.SECURITY_PROTOCOL_CONFIG;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_JAAS_CONFIG;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_MECHANISM;
+import static
org.apache.kafka.common.config.internals.BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG;
+import static
org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG;
+import static
org.apache.kafka.metadata.authorizer.StandardAuthorizer.ALLOW_EVERYONE_IF_NO_ACL_IS_FOUND_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ClusterTestDefaults(
+ types = {Type.CO_KRAFT},
+ serverProperties = {
+ @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG,
value = "1"),
+ @ClusterConfigProperty(key = SASL_ENABLED_MECHANISMS_CONFIG, value =
"PLAIN,SCRAM-SHA-256"),
+ @ClusterConfigProperty(key = ALLOW_EVERYONE_IF_NO_ACL_IS_FOUND_CONFIG,
value = "true"),
+ }
+)
+public class ConsumerGroupCommandSaslAuthenticationTest {
+
+ private static final String TOPIC = "topic";
+ private static final String KAFKA_CLIENT_SASL_MECHANISM = "SCRAM-SHA-256";
+ private static final String SCRAM_USER = "scram-user";
+ private static final String SCRAM_PASSWORD = "scram-user-secret";
+ private static final String KAFKA_CLIENT_SASL_JAAS_CONFIG =
+ "org.apache.kafka.common.security.scram.ScramLoginModule required
username=\"" + SCRAM_USER + "\" password=\"" + SCRAM_PASSWORD + "\";";
+
+ private final ClusterInstance cluster;
+
+ public ConsumerGroupCommandSaslAuthenticationTest(ClusterInstance cluster)
{
+ this.cluster = cluster;
+ }
+
+ @ClusterTest(brokerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT)
+ public void
testConsumerGroupServiceWithAuthenticationFailureWithClassicGroupProtocol()
throws Exception {
+
testConsumerGroupServiceWithAuthenticationFailure(GroupProtocol.CLASSIC);
+ }
+
+ @ClusterTest(brokerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT)
+ public void
testConsumerGroupServiceWithAuthenticationFailureWithConsumerGroupProtocol()
throws Exception {
+
testConsumerGroupServiceWithAuthenticationFailure(GroupProtocol.CONSUMER);
+ }
+
+ @ClusterTest(brokerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT)
+ public void
testConsumerGroupServiceWithAuthenticationSuccessWithClassicGroupProtocol()
throws Exception {
+
testConsumerGroupServiceWithAuthenticationSuccess(GroupProtocol.CLASSIC);
+ }
+
+ @ClusterTest(brokerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT)
+ public void
testConsumerGroupServiceWithAuthenticationSuccessWithConsumerGroupProtocol()
throws Exception {
+
testConsumerGroupServiceWithAuthenticationSuccess(GroupProtocol.CONSUMER);
+ }
+
+ private void
testConsumerGroupServiceWithAuthenticationFailure(GroupProtocol groupProtocol)
throws Exception {
+ cluster.createTopic(TOPIC, 1, (short) 1);
+ try (
+ ConsumerGroupCommand.ConsumerGroupService consumerGroupService =
prepareConsumerGroupService();
+ KafkaConsumer<byte[], byte[]> consumer =
createScramConsumer(groupProtocol)
+ ) {
+ consumer.subscribe(List.of(TOPIC));
+ verifyAuthenticationException(consumerGroupService::listGroups);
+ }
+ }
+
+ private void
testConsumerGroupServiceWithAuthenticationSuccess(GroupProtocol groupProtocol)
throws Exception {
+ cluster.createTopic(TOPIC, 1, (short) 1);
+ createScramCredential(SCRAM_USER, SCRAM_PASSWORD);
+ try (
+ ConsumerGroupCommand.ConsumerGroupService consumerGroupService =
prepareConsumerGroupService();
+ KafkaConsumer<byte[], byte[]> consumer =
createScramConsumer(groupProtocol)
+ ) {
+ consumer.subscribe(List.of(TOPIC));
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ consumer.poll(Duration.ofMillis(1000));
+ return true;
+ } catch (SaslAuthenticationException ignored) {
+ return false;
+ }
+ }, "failed to poll data with authentication");
+
+ TestUtils.waitForCondition(
+ () -> consumerGroupService.listConsumerGroups().size() == 1,
+ "failed to find consumer group after successful poll"
+ );
+ }
+ }
+
+ private void createScramCredential(String user, String password) throws
ExecutionException, InterruptedException {
+ try (Admin admin = cluster.admin()) {
+ AlterUserScramCredentialsResult result =
admin.alterUserScramCredentials(List.of(
+ new UserScramCredentialUpsertion(user,
+ new ScramCredentialInfo(ScramMechanism.SCRAM_SHA_256,
4096), password)
+ ));
+ result.all().get();
+ }
+ }
+
+ private KafkaConsumer<byte[], byte[]> createScramConsumer(GroupProtocol
groupProtocol) {
+ Properties props = new Properties();
+ props.put(BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
+ props.put(SECURITY_PROTOCOL_CONFIG,
SecurityProtocol.SASL_PLAINTEXT.name);
+ props.put(SASL_MECHANISM, KAFKA_CLIENT_SASL_MECHANISM);
+ props.put(SASL_JAAS_CONFIG, KAFKA_CLIENT_SASL_JAAS_CONFIG);
+ props.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
+ props.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name());
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+ props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ return new KafkaConsumer<>(props);
+ }
+
+ private ConsumerGroupCommand.ConsumerGroupService
prepareConsumerGroupService() throws IOException {
+ File propsFile = TestUtils.tempFile(
+ "security.protocol=SASL_PLAINTEXT\n" +
+ "sasl.mechanism=" + KAFKA_CLIENT_SASL_MECHANISM + "\n" +
+ "sasl.jaas.config=" + KAFKA_CLIENT_SASL_JAAS_CONFIG);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server",
cluster.bootstrapServers(),
+ "--describe",
+ "--group", "test.group",
+ "--command-config", propsFile.getAbsolutePath()};
+ ConsumerGroupCommandOptions opts =
ConsumerGroupCommandOptions.fromArgs(cgcArgs);
+ return new ConsumerGroupCommand.ConsumerGroupService(opts, Map.of());
+ }
+
+ private void verifyAuthenticationException(Executable action) {
+ long startMs = System.currentTimeMillis();
+ assertThrows(Exception.class, action);
+ long elapsedMs = System.currentTimeMillis() - startMs;
+ assertTrue(elapsedMs <= 5000, "Poll took too long, elapsed=" +
elapsedMs);
+ }
+}
diff --git
a/tools/src/test/java/org/apache/kafka/tools/consumer/group/SaslClientsWithInvalidCredentialsTest.java
b/tools/src/test/java/org/apache/kafka/tools/consumer/group/SaslClientsWithInvalidCredentialsTest.java
deleted file mode 100644
index 3676cf4d482..00000000000
---
a/tools/src/test/java/org/apache/kafka/tools/consumer/group/SaslClientsWithInvalidCredentialsTest.java
+++ /dev/null
@@ -1,193 +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 org.apache.kafka.tools.consumer.group;
-
-import kafka.api.AbstractSaslTest;
-import kafka.security.JaasTestUtils;
-
-import org.apache.kafka.clients.admin.Admin;
-import org.apache.kafka.clients.admin.NewTopic;
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.common.config.SaslConfigs;
-import org.apache.kafka.common.errors.SaslAuthenticationException;
-import org.apache.kafka.common.security.auth.SecurityProtocol;
-import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import org.apache.kafka.metadata.storage.Formatter;
-import org.apache.kafka.test.TestUtils;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.TestInfo;
-import org.junit.jupiter.api.function.Executable;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.MethodSource;
-
-import java.io.File;
-import java.io.IOException;
-import java.time.Duration;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import scala.Option;
-import scala.Some$;
-import scala.collection.Seq;
-import scala.jdk.javaapi.CollectionConverters;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-public class SaslClientsWithInvalidCredentialsTest extends AbstractSaslTest {
- private static final String TOPIC = "topic";
- public static final int NUM_PARTITIONS = 1;
- public static final int BROKER_COUNT = 1;
- public static final String KAFKA_CLIENT_SASL_MECHANISM = "SCRAM-SHA-256";
- private static final Seq<String> KAFKA_SERVER_SASL_MECHANISMS =
CollectionConverters.asScala(List.of(KAFKA_CLIENT_SASL_MECHANISM)).toSeq();
-
- private Consumer<byte[], byte[]> createConsumer() {
- return createConsumer(
- new ByteArrayDeserializer(),
- new ByteArrayDeserializer(),
- new Properties(),
- CollectionConverters.asScala(Set.<String>of()).toList()
- );
- }
-
- @Override
- public SecurityProtocol securityProtocol() {
- return SecurityProtocol.SASL_PLAINTEXT;
- }
-
- @Override
- public Option<Properties> serverSaslProperties() {
- return
Some$.MODULE$.apply(kafkaServerSaslProperties(KAFKA_SERVER_SASL_MECHANISMS,
KAFKA_CLIENT_SASL_MECHANISM));
- }
-
- @Override
- public Option<Properties> clientSaslProperties() {
- return
Some$.MODULE$.apply(kafkaClientSaslProperties(KAFKA_CLIENT_SASL_MECHANISM,
false));
- }
-
- @Override
- public int brokerCount() {
- return 1;
- }
-
- @Override
- public void configureSecurityBeforeServersStart(TestInfo testInfo) {
- super.configureSecurityBeforeServersStart(testInfo);
- }
-
- @Override
- public void addFormatterSettings(Formatter formatter) {
- formatter.setClusterId("XcZZOzUqS4yHOjhMQB6JLQ");
- formatter.setScramArguments(List.of("SCRAM-SHA-256=[name=" +
JaasTestUtils.KAFKA_SCRAM_ADMIN +
- ",password=" + JaasTestUtils.KAFKA_SCRAM_ADMIN_PASSWORD + "]"));
- }
-
- @Override
- public Admin createPrivilegedAdminClient() {
- return createAdminClient(bootstrapServers(listenerName()),
securityProtocol(), trustStoreFile(), clientSaslProperties(),
- KAFKA_CLIENT_SASL_MECHANISM, JaasTestUtils.KAFKA_SCRAM_ADMIN,
JaasTestUtils.KAFKA_SCRAM_ADMIN_PASSWORD);
- }
-
- @BeforeEach
- @Override
- public void setUp(TestInfo testInfo) {
- startSasl(jaasSections(KAFKA_SERVER_SASL_MECHANISMS,
- Some$.MODULE$.apply(KAFKA_CLIENT_SASL_MECHANISM),
- JaasTestUtils.KAFKA_SERVER_CONTEXT_NAME));
- String superuserLoginContext =
jaasAdminLoginModule(KAFKA_CLIENT_SASL_MECHANISM, Option.empty());
- this.superuserClientConfig().put(SaslConfigs.SASL_JAAS_CONFIG,
superuserLoginContext);
- super.setUp(testInfo);
- try (Admin admin = createPrivilegedAdminClient()) {
- admin.createTopics(List.of(
- new NewTopic(TOPIC, NUM_PARTITIONS, (short)
BROKER_COUNT))).all().
- get(5, TimeUnit.MINUTES);
- } catch (ExecutionException | InterruptedException | TimeoutException
e) {
- throw new RuntimeException(e);
- }
- }
-
- @AfterEach
- @Override
- public void tearDown() {
- super.tearDown();
- closeSasl();
- }
-
- // NOTE: Not able to refer
TestInfoUtils#TestWithParameterizedGroupProtocolNames() in the
ParameterizedTest name.
- @ParameterizedTest(name = "{displayName}.groupProtocol={0}")
- @MethodSource("getTestGroupProtocolParametersAll")
- public void testConsumerGroupServiceWithAuthenticationFailure(String
groupProtocol) throws Exception {
- try (
- ConsumerGroupCommand.ConsumerGroupService consumerGroupService =
prepareConsumerGroupService();
- Consumer<byte[], byte[]> consumer = createConsumer()
- ) {
- consumer.subscribe(List.of(TOPIC));
- verifyAuthenticationException(consumerGroupService::listGroups);
- }
- }
-
- // NOTE: Not able to refer
TestInfoUtils#TestWithParameterizedGroupProtocolNames() in the
ParameterizedTest name.
- @ParameterizedTest(name = "{displayName}.groupProtocol={0}")
- @MethodSource("getTestGroupProtocolParametersAll")
- public void testConsumerGroupServiceWithAuthenticationSuccess(String
groupProtocol) throws Exception {
-
createScramCredentialsViaPrivilegedAdminClient(JaasTestUtils.KAFKA_SCRAM_USER_2,
JaasTestUtils.KAFKA_SCRAM_PASSWORD_2);
- try (
- ConsumerGroupCommand.ConsumerGroupService consumerGroupService =
prepareConsumerGroupService();
- Consumer<byte[], byte[]> consumer = createConsumer()
- ) {
- consumer.subscribe(List.of(TOPIC));
-
- TestUtils.waitForCondition(() -> {
- try {
- consumer.poll(Duration.ofMillis(1000));
- return true;
- } catch (SaslAuthenticationException ignored) {
- return false;
- }
- }, "failed to poll data with authentication");
- assertEquals(1, consumerGroupService.listConsumerGroups().size());
- }
- }
-
- private ConsumerGroupCommand.ConsumerGroupService
prepareConsumerGroupService() throws IOException {
- File propsFile = TestUtils.tempFile(
- "security.protocol=SASL_PLAINTEXT\n" +
- "sasl.mechanism=" + KAFKA_CLIENT_SASL_MECHANISM);
-
- String[] cgcArgs = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()),
- "--describe",
- "--group", "test.group",
- "--command-config", propsFile.getAbsolutePath()};
- ConsumerGroupCommandOptions opts =
ConsumerGroupCommandOptions.fromArgs(cgcArgs);
- return new ConsumerGroupCommand.ConsumerGroupService(opts, Map.of());
- }
-
- private void verifyAuthenticationException(Executable action) {
- long startMs = System.currentTimeMillis();
- assertThrows(Exception.class, action);
- long elapsedMs = System.currentTimeMillis() - startMs;
- assertTrue(elapsedMs <= 5000, "Poll took too long, elapsed=" +
elapsedMs);
- }
-}
