FrankYang0529 commented on code in PR #17671:
URL: https://github.com/apache/kafka/pull/17671#discussion_r1899597568
##########
test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java:
##########
@@ -331,4 +342,105 @@ 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
InterruptedException {
+ Assertions.assertEquals(SecurityProtocol.SASL_PLAINTEXT,
clusterInstance.config().brokerSecurityProtocol());
+
+ // default ClusterInstance#admin helper with admin credentials
+ try (Admin admin = clusterInstance.admin()) {
+ assertDoesNotThrow(() ->
admin.describeAcls(AclBindingFilter.ANY).values().get());
+ }
+ String topic = "sasl-plaintext-topic";
+ Assertions.assertDoesNotThrow(() -> clusterInstance.createTopic(topic,
1, (short) 1));
+ try (Producer<byte[], byte[]> producer = clusterInstance.producer()) {
+ assertDoesNotThrow(() -> producer.send(new ProducerRecord<>(topic,
Utils.utf8("key"), Utils.utf8("value"))).get());
+ assertDoesNotThrow(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(
Review Comment:
Hi @chia7712, thanks for the review. Yes, we already set default user to
`admin`, `consumer`, and `producer` helper function automatically. Here, we
would like to test developer overrides `sasl.jaas.config` with non-admin
credentials. We also test overriding with unknown credential, so we can make
sure the cluster instance with sasl plaintext works.
--
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]
