FrankYang0529 commented on code in PR #17671:
URL: https://github.com/apache/kafka/pull/17671#discussion_r1906171247
##########
test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java:
##########
@@ -331,4 +346,143 @@ public void testControllerListenerName(ClusterInstance
cluster) throws Execution
assertEquals(1,
admin.describeMetadataQuorum().quorumInfo().get().nodes().size());
}
}
+
+ @ClusterTest(
+ types = {Type.KRAFT, Type.CO_KRAFT},
+ brokerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT,
+ controllerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT,
+ serverProperties = {
+ @ClusterConfigProperty(key =
GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key =
GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
+ }
+ )
+ public void testSaslPlaintext(ClusterInstance clusterInstance) throws
CancellationException, ExecutionException, InterruptedException {
+ Assertions.assertEquals(SecurityProtocol.SASL_PLAINTEXT,
clusterInstance.config().brokerSecurityProtocol());
+
+ // default ClusterInstance#admin helper with admin credentials
+ try (Admin admin = clusterInstance.admin()) {
+ admin.describeAcls(AclBindingFilter.ANY).values().get();
+ }
+ String topic = "sasl-plaintext-topic";
+ clusterInstance.createTopic(topic, 1, (short) 1);
+ 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));
+ TestUtils.waitForCondition(() -> {
+ ConsumerRecords<byte[], byte[]> records =
consumer.poll(Duration.ofMillis(100));
+ return records.count() == 1;
+ }, "Failed to receive message");
+ }
+
+ // client with non-admin credentials
+ Map<String, Object> nonAdminConfig = Map.of(
+ SaslConfigs.SASL_JAAS_CONFIG,
+ String.format(
+ "org.apache.kafka.common.security.plain.PlainLoginModule
required username=\"%s\" password=\"%s\";",
+ JaasUtils.KAFKA_PLAIN_USER1,
JaasUtils.KAFKA_PLAIN_USER1_PASSWORD
+ )
+ );
+ try (Admin admin = clusterInstance.admin(nonAdminConfig)) {
+ ExecutionException exception = assertThrows(
+ ExecutionException.class,
+ () -> admin.describeAcls(AclBindingFilter.ANY).values().get()
+ );
+ assertInstanceOf(ClusterAuthorizationException.class,
exception.getCause());
+ }
+ try (Producer<byte[], byte[]> producer =
clusterInstance.producer(nonAdminConfig)) {
+ ExecutionException exception = assertThrows(
+ ExecutionException.class,
+ () -> producer.send(new ProducerRecord<>(topic,
Utils.utf8("key"), Utils.utf8("value"))).get()
+ );
+ assertInstanceOf(TopicAuthorizationException.class,
exception.getCause());
+ }
+ try (Consumer<byte[], byte[]> consumer =
clusterInstance.consumer(nonAdminConfig)) {
+ consumer.subscribe(List.of(topic));
+ AtomicBoolean hasException = new AtomicBoolean(false);
+ TestUtils.waitForCondition(() -> {
+ if (hasException.get()) {
+ return true;
+ }
+ try {
+ consumer.poll(Duration.ofMillis(100));
+ } catch (TopicAuthorizationException e) {
+ hasException.set(true);
+ }
+ return false;
+ }, "Failed to get exception");
+ }
+
+ // client with unknown credentials
+ Map<String, Object> unknownUserConfig = Map.of(
+ SaslConfigs.SASL_JAAS_CONFIG,
+ String.format(
+ "org.apache.kafka.common.security.plain.PlainLoginModule
required username=\"%s\" password=\"%s\";",
+ "unknown", "unknown"
+ )
+ );
+ try (Admin admin = clusterInstance.admin(unknownUserConfig)) {
+ ExecutionException exception = assertThrows(
+ ExecutionException.class,
+ () -> admin.describeAcls(AclBindingFilter.ANY).values().get()
+ );
+ assertInstanceOf(SaslAuthenticationException.class,
exception.getCause());
+ }
+ try (Producer<byte[], byte[]> producer =
clusterInstance.producer(unknownUserConfig)) {
+ ExecutionException exception = assertThrows(
+ ExecutionException.class,
+ () -> producer.send(new ProducerRecord<>(topic,
Utils.utf8("key"), Utils.utf8("value"))).get()
+ );
+ assertInstanceOf(SaslAuthenticationException.class,
exception.getCause());
+ }
+ try (Consumer<byte[], byte[]> consumer =
clusterInstance.consumer(unknownUserConfig)) {
+ consumer.subscribe(List.of(topic));
+ AtomicBoolean hasException = new AtomicBoolean(false);
+ TestUtils.waitForCondition(() -> {
+ if (hasException.get()) {
+ return true;
+ }
+ try {
+ consumer.poll(Duration.ofMillis(100));
+ } catch (SaslAuthenticationException e) {
+ hasException.set(true);
+ }
+ return false;
+ }, "Failed to get exception");
+ }
+ }
+
+ @ClusterTest(
+ types = {Type.KRAFT, Type.CO_KRAFT},
+ brokerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT,
+ controllerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT,
+ serverProperties = {
+ @ClusterConfigProperty(key =
GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key =
GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
+ }
+ )
+ public void testSaslPlaintextWithController(ClusterInstance
clusterInstance) throws CancellationException, ExecutionException,
InterruptedException {
+ // test with admin
+ try (Admin admin = clusterInstance.admin(Map.of(), true)) {
+ admin.describeAcls(AclBindingFilter.ANY).values().get();
+ }
+
+ // test with non-admin
+ Map<String, Object> nonAdminConfig = Map.of(
+ SaslConfigs.SASL_JAAS_CONFIG,
+ String.format(
+ "org.apache.kafka.common.security.plain.PlainLoginModule
required username=\"%s\" password=\"%s\";",
+ JaasUtils.KAFKA_PLAIN_USER1,
JaasUtils.KAFKA_PLAIN_USER1_PASSWORD
+ )
+ );
+ try (Admin admin = clusterInstance.admin(nonAdminConfig, true)) {
+ ExecutionException exception = assertThrows(
+ ExecutionException.class,
+ () -> admin.describeAcls(AclBindingFilter.ANY, new
DescribeAclsOptions().timeoutMs(5000)).values().get()
+ );
+ assertInstanceOf(TimeoutException.class, exception.getCause());
Review Comment:
Thanks for review. I will handle it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
