chia7712 commented on code in PR #17671:
URL: https://github.com/apache/kafka/pull/17671#discussion_r1899527500
##########
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));
Review Comment:
Since the APIs in question only throw runtime exceptions, there's no need to
use `assertDoesNotThrow`, which is typically employed to convert checked
exceptions into runtime exceptions.
##########
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:
the helper methods of `clusterInstance` should be able configure the
`sasl.jaas.config` automatically - that is why we add those helper to
`clusterInstance`
--
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]
