mingyen066 commented on code in PR #22575:
URL: https://github.com/apache/kafka/pull/22575#discussion_r3507380694
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerFailureHandlingTest.java:
##########
@@ -215,22 +216,110 @@ 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 {
+ // explicitly create __consumer_offsets to satisfy the "topic exists"
precondition
Review Comment:
The four new tests repeat a lot of boilerplate —
`...TrueAndTopicExists/...FalseAndTopicExists` share nearly identical
topic-creation code, and all four repeat the same send+assert block with only
the expected exception differing. Worth extracting two helpers:
```java
private void createInternalTopic(ClusterInstance clusterInstance) throws
Exception {
try (Admin admin = clusterInstance.admin()) {
Map<String, String> topicConfig = clusterInstance.brokers().get(0)
.groupCoordinator()
.groupMetadataTopicConfigs();
admin.createTopics(List.of(new
NewTopic(Topic.GROUP_METADATA_TOPIC_NAME, 1, (short) 1).configs(topicConfig)));
clusterInstance.waitTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME,
1);
}
}
private void assertSendToInternalTopicFails(ClusterInstance clusterInstance,
Class<? extends Throwable> expectedCause) {
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(expectedCause, thrown.getCause(),
() -> "Expected " + expectedCause.getSimpleName() + " but
got " + thrown.getCause());
}
}
```
Then each test shrinks to:
```java
@ClusterTest(serverProperties = {@ClusterConfigProperty(key =
AUTO_CREATE_TOPICS_ENABLE_CONFIG, value = "true")})
public void
testCannotSendToInternalTopicWhenAutoCreateTrueAndTopicExists(ClusterInstance
clusterInstance) throws Exception {
createInternalTopic(clusterInstance);
assertSendToInternalTopicFails(clusterInstance,
InvalidTopicException.class);
}
```
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerFailureHandlingTest.java:
##########
@@ -215,22 +216,110 @@ 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 {
+ // explicitly create __consumer_offsets to satisfy the "topic exists"
precondition
+ try (Admin admin = clusterInstance.admin()) {
+ Map<String, String> topicConfig = clusterInstance.brokers().get(0)
+ .groupCoordinator()
+ .groupMetadataTopicConfigs();
+ admin.createTopics(List.of(new
NewTopic(Topic.GROUP_METADATA_TOPIC_NAME, 1, (short) 1).configs(topicConfig)));
+ clusterInstance.waitTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME,
1);
+ }
+
+ 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(),
+ () -> "Expected InvalidTopicException but got " +
thrown.getCause());
+ }
+ }
+
+ /**
+ * 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 {
+ public void
testCannotSendToInternalTopicWhenAutoCreateFalseAndTopicExists(ClusterInstance
clusterInstance) throws Exception {
+ // explicitly create __consumer_offsets to satisfy the "topic exists"
precondition
try (Admin admin = clusterInstance.admin()) {
Map<String, String> topicConfig = clusterInstance.brokers().get(0)
- .groupCoordinator()
- .groupMetadataTopicConfigs();
+ .groupCoordinator()
+ .groupMetadataTopicConfigs();
admin.createTopics(List.of(new
NewTopic(Topic.GROUP_METADATA_TOPIC_NAME, 1, (short) 1).configs(topicConfig)));
- clusterInstance.waitTopicDeletion(Topic.GROUP_METADATA_TOPIC_NAME);
+ clusterInstance.waitTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME,
1);
}
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(),
+ () -> 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());
+ () -> "Expected InvalidTopicException but got " +
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 {
+ clusterInstance.deleteTopic(Topic.GROUP_METADATA_TOPIC_NAME);
Review Comment:
`clusterInstance.deleteTopic(Topic.GROUP_METADATA_TOPIC_NAME)` is a no-op
here — `__consumer_offsets` doesn't exist yet at this point, and `deleteTopic`
discards the result without awaiting it. Looks like leftover from the old test;
safe to drop.
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerFailureHandlingTest.java:
##########
@@ -215,22 +216,110 @@ 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 {
+ // explicitly create __consumer_offsets to satisfy the "topic exists"
precondition
+ try (Admin admin = clusterInstance.admin()) {
+ Map<String, String> topicConfig = clusterInstance.brokers().get(0)
+ .groupCoordinator()
+ .groupMetadataTopicConfigs();
+ admin.createTopics(List.of(new
NewTopic(Topic.GROUP_METADATA_TOPIC_NAME, 1, (short) 1).configs(topicConfig)));
+ clusterInstance.waitTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME,
1);
+ }
+
+ 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(),
+ () -> "Expected InvalidTopicException but got " +
thrown.getCause());
+ }
+ }
+
+ /**
+ * 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 {
+ public void
testCannotSendToInternalTopicWhenAutoCreateFalseAndTopicExists(ClusterInstance
clusterInstance) throws Exception {
+ // explicitly create __consumer_offsets to satisfy the "topic exists"
precondition
try (Admin admin = clusterInstance.admin()) {
Map<String, String> topicConfig = clusterInstance.brokers().get(0)
- .groupCoordinator()
- .groupMetadataTopicConfigs();
+ .groupCoordinator()
+ .groupMetadataTopicConfigs();
admin.createTopics(List.of(new
NewTopic(Topic.GROUP_METADATA_TOPIC_NAME, 1, (short) 1).configs(topicConfig)));
- clusterInstance.waitTopicDeletion(Topic.GROUP_METADATA_TOPIC_NAME);
+ clusterInstance.waitTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME,
1);
}
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(),
+ () -> 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());
+ () -> "Expected InvalidTopicException but got " +
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 {
+ clusterInstance.deleteTopic(Topic.GROUP_METADATA_TOPIC_NAME);
+
+ 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(),
+ () -> "Expected InvalidTopicException but got " +
thrown.getCause());
+ }
+
+ // 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 {
+ clusterInstance.deleteTopic(Topic.GROUP_METADATA_TOPIC_NAME);
Review Comment:
ditto
--
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]
