aravind-kvs commented on code in PR #22575:
URL: https://github.com/apache/kafka/pull/22575#discussion_r3951598793
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerFailureHandlingTest.java:
##########
@@ -215,19 +216,58 @@ public void testSendAfterClosed(ClusterInstance
clusterInstance) throws Interrup
assertThrows(IllegalStateException.class, () ->
producer3.send(record));
}
+ /**
+ * Test that sending to an internal topic throws InvalidTopicException
+ * when auto.create.topics.enable=true and the internal topic already
exists.
+ */
+ @ClusterTest(serverProperties = {
+ @ClusterConfigProperty(key = AUTO_CREATE_TOPICS_ENABLE_CONFIG, value =
"true")
+ })
+ public void
testCannotSendToInternalTopicWhenAutoCreateTrueAndTopicExists(ClusterInstance
clusterInstance) throws Exception {
+ createInternalTopic(clusterInstance);
+ assertSendToInternalTopicFails(clusterInstance,
InvalidTopicException.class);
+ }
+
+ /**
+ * Test that sending to an internal topic throws InvalidTopicException
+ * when auto.create.topics.enable=false and the internal topic already
exists.
+ */
@ClusterTest
- public void testCannotSendToInternalTopic(ClusterInstance clusterInstance)
throws InterruptedException {
- Map<String, String> topicConfig = clusterInstance.brokers().get(0)
- .groupCoordinator()
- .groupMetadataTopicConfigs();
- clusterInstance.createTopic(Topic.GROUP_METADATA_TOPIC_NAME, 1,
(short) 1, topicConfig);
+ public void
testCannotSendToInternalTopicWhenAutoCreateFalseAndTopicExists(ClusterInstance
clusterInstance) throws Exception {
+ createInternalTopic(clusterInstance);
+ assertSendToInternalTopicFails(clusterInstance,
InvalidTopicException.class);
+ }
- try (Producer<byte[], byte[]> producer =
clusterInstance.producer(producerConfig(1))) {
- Exception thrown = assertThrows(ExecutionException.class,
- () -> producer.send(new
ProducerRecord<>(Topic.GROUP_METADATA_TOPIC_NAME, "test".getBytes(),
- "test".getBytes())).get());
- assertInstanceOf(InvalidTopicException.class, thrown.getCause(),
- () -> "Unexpected exception while sending to an invalid
topic " + thrown.getCause());
+ /**
+ * Test that sending to an internal topic throws InvalidTopicException
+ * when auto.create.topics.enable=true and the internal topic does not
exist.
+ * The broker should auto-create the internal topic even though the send
is rejected.
+ */
+ @ClusterTest(serverProperties = {
+ @ClusterConfigProperty(key = AUTO_CREATE_TOPICS_ENABLE_CONFIG, value =
"true"),
+ @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG,
value = "1")
+ })
+ public void
testCannotSendToInternalTopicWhenAutoCreateTrueAndTopicNotExists(ClusterInstance
clusterInstance) throws Exception {
+ assertSendToInternalTopicFails(clusterInstance,
InvalidTopicException.class);
+
+ // verify the broker auto-created the internal topic
+ clusterInstance.waitTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME, 1);
+ }
+
+ /**
+ * Test that sending to an internal topic throws TimeoutException
+ * when auto.create.topics.enable=false and the internal topic does not
exist.
+ * The broker returns UNKNOWN_TOPIC_OR_PARTITION as a recoverable error
causing the producer to time out.
+ */
+ @ClusterTest(serverProperties = {
+ @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG,
value = "1")
+ })
+ public void
testCannotSendToInternalTopicWhenAutoCreateFalseAndTopicNotExists(ClusterInstance
clusterInstance) throws Exception {
+ assertSendToInternalTopicFails(clusterInstance,
TimeoutException.class);
+
+ // verify the internal topic was not auto-created
+ try (Admin admin = clusterInstance.admin()) {
+
assertFalse(admin.listTopics().names().get().contains(Topic.GROUP_METADATA_TOPIC_NAME));
Review Comment:
Added `ListTopicsOptions` with `listInternal(true)`🙂
--
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]
