From e91e66ce5ee8d6df673fc7be99f8693e7fe2de3f Mon Sep 17 00:00:00 2001 From: Yen-Ting Li Date: Mon, 5 Oct 2026 23:12:35 +0800 Subject: [PATCH] Propagate ClusterInstance#deleteTopic errors and use it in more tests --- .../kafka/clients/admin/DeleteTopicTest.java | 67 +++++++------------ .../kafka/common/test/ClusterInstance.java | 11 ++- .../test/junit/ClusterTestExtensionsTest.java | 8 ++- .../apache/kafka/tools/TopicCommandTest.java | 6 +- .../group/DescribeConsumerGroupTest.java | 10 +-- 5 files changed, 44 insertions(+), 58 deletions(-) diff --git a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DeleteTopicTest.java b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DeleteTopicTest.java index 5ed4f1f59ff56..3bc7ea0e2abf3 100644 --- a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DeleteTopicTest.java +++ b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DeleteTopicTest.java @@ -69,26 +69,17 @@ public class DeleteTopicTest { @ClusterTest public void testCreateAndDeleteTopic(ClusterInstance cluster) throws Exception { String testTopic = "test-topic"; - try (Admin admin = cluster.admin()) { - // Create a test topic - cluster.createTopic(testTopic, 1, (short) 3); - - // Delete topic - DeleteTopicsResult deleteResult = admin.deleteTopics(List.of(testTopic)); - deleteResult.all().get(); + // Create a test topic + cluster.createTopic(testTopic, 1, (short) 3); - // Wait for topic deletion - cluster.waitTopicDeletion(testTopic); - } + // Delete topic and wait for topic deletion + cluster.deleteTopic(testTopic); } @ClusterTest public void testDeleteTopicWithAllAliveReplicas(ClusterInstance cluster) throws Exception { - try (Admin admin = cluster.admin()) { - cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment); - admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get(); - cluster.waitTopicDeletion(DEFAULT_TOPIC); - } + cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment); + cluster.deleteTopic(DEFAULT_TOPIC); } @ClusterTest @@ -209,15 +200,12 @@ public void testAddPartitionDuringDeleteTopic(ClusterInstance cluster) throws Ex @ClusterTest public void testRecreateTopicAfterDeletion(ClusterInstance cluster) throws Exception { - try (Admin admin = cluster.admin()) { - cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment); - TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0); - admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get(); - cluster.waitTopicDeletion(DEFAULT_TOPIC); - // re-create topic on same replicas - cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment); - waitForReplicaCreated(cluster.brokers(), topicPartition, "Replicas for topic " + DEFAULT_TOPIC + " not created."); - } + cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment); + TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0); + cluster.deleteTopic(DEFAULT_TOPIC); + // re-create topic on same replicas + cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment); + waitForReplicaCreated(cluster.brokers(), topicPartition, "Replicas for topic " + DEFAULT_TOPIC + " not created."); } @ClusterTest @@ -248,23 +236,20 @@ public void testDeleteNonExistingTopic(ClusterInstance cluster) throws Exception @ClusterConfigProperty(key = "log.cleaner.dedupe.buffer.size", value = "1048577") }) public void testDeleteTopicWithCleaner(ClusterInstance cluster) throws Exception { - try (Admin admin = cluster.admin()) { - cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment); - TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0); - // for simplicity, we are validating cleaner offsets on a single broker - KafkaBroker server = cluster.brokers().values().stream().findFirst().orElseThrow(); - TestUtils.waitForCondition(() -> server.logManager().getLog(topicPartition, false).isPresent(), - "Replicas for topic test not created."); - UnifiedLog log = server.logManager().getLog(topicPartition, false).get(); - writeDups(100, 3, log); - // force roll the segment so that cleaner can work on it - server.logManager().getLog(topicPartition, false).get().roll(Optional.empty()); - // wait for cleaner to clean - server.logManager().cleaner().awaitCleaned(topicPartition, 0, 60000); - admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get(); - - cluster.waitTopicDeletion(DEFAULT_TOPIC); - } + cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment); + TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0); + // for simplicity, we are validating cleaner offsets on a single broker + KafkaBroker server = cluster.brokers().values().stream().findFirst().orElseThrow(); + TestUtils.waitForCondition(() -> server.logManager().getLog(topicPartition, false).isPresent(), + "Replicas for topic test not created."); + UnifiedLog log = server.logManager().getLog(topicPartition, false).get(); + writeDups(100, 3, log); + // force roll the segment so that cleaner can work on it + server.logManager().getLog(topicPartition, false).get().roll(Optional.empty()); + // wait for cleaner to clean + server.logManager().cleaner().awaitCleaned(topicPartition, 0, 60000); + + cluster.deleteTopic(DEFAULT_TOPIC); } @ClusterTest diff --git a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java index 2fd63d771ef9d..76b3f5c4fca6d 100644 --- a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java +++ b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java @@ -356,12 +356,17 @@ default void createTopicWithAssignment(String topicName, Map topic.name().equals(testTopic) )); + + assertThrows(UnknownTopicOrPartitionException.class, () -> clusterInstance.deleteTopic("non-existent-topic")); } } diff --git a/tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java index 9f0b148eaa431..2d90d60835d02 100644 --- a/tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java +++ b/tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java @@ -389,8 +389,7 @@ public void testCreate(ClusterInstance clusterInstance) throws InterruptedExcept Assertions.assertTrue(adminClient.listTopics().names().get().contains(testTopicName), "Admin client didn't see the created topic. It saw: " + adminClient.listTopics().names().get()); - adminClient.deleteTopics(List.of(testTopicName)); - clusterInstance.waitTopicDeletion(testTopicName); + clusterInstance.deleteTopic(testTopicName); Assertions.assertTrue(adminClient.listTopics().names().get().isEmpty(), "Admin client see the created topic. It saw: " + adminClient.listTopics().names().get()); } @@ -420,8 +419,7 @@ public void testCreateWithDefaults(ClusterInstance clusterInstance) throws Inter Assertions.assertEquals(defaultNumPartitions, partitions.size(), "Unequal partition size: " + partitions.size()); Assertions.assertEquals(defaultReplicationFactor, (short) partitions.get(0).replicas().size(), "Unequal replication factor: " + partitions.get(0).replicas().size()); - adminClient.deleteTopics(List.of(testTopicName)); - clusterInstance.waitTopicDeletion(testTopicName); + clusterInstance.deleteTopic(testTopicName); Assertions.assertTrue(adminClient.listTopics().names().get().isEmpty(), "Admin client see the created topic. It saw: " + adminClient.listTopics().names().get()); } diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java index dc275fed0d2d0..da9da69c75f73 100644 --- a/tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java +++ b/tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java @@ -86,6 +86,8 @@ @ClusterConfigProperty(key = GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, value = "1000"), @ClusterConfigProperty(key = CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG, value = "500"), @ClusterConfigProperty(key = CONSUMER_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG, value = "500"), + @ClusterConfigProperty(key = "log.initial.task.delay.ms", value = "100"), + @ClusterConfigProperty(key = "log.segment.delete.delay.ms", value = "1000"), } ) public class DescribeConsumerGroupTest { @@ -439,7 +441,7 @@ public void testDescribeAllExistingGroups(ClusterInstance clusterInstance) throw } // remove previous consumer groups, so we can have a clean cluster for next consumer group protocol test. deleteConsumerGroups(groups); - deleteTopic(topic); + clusterInstance.deleteTopic(topic); } } } @@ -1182,12 +1184,6 @@ private void deleteConsumerGroups(Collection groupIds) { } } - private void deleteTopic(String topic) { - try (Admin admin = Admin.create(Map.of(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers()))) { - Assertions.assertDoesNotThrow(() -> admin.deleteTopics(List.of(topic)).topicNameValues().get(topic).get()); - } - } - private AutoCloseable consumerGroupClosable(GroupProtocol protocol, String groupId, Set topicPartitions, Map customConfigs) { Map configs = composeConfigs( groupId,