MINOR: Propagate ClusterInstance#deleteTopic errors and use it in more tests (#23715)

Follow-up to #23699 to apply the same changes to
ClusterInstance#deleteTopic.

ClusterInstance#deleteTopic did not wait for the DeleteTopics result, so
a failed deletion (e.g. the topic did not exist) was either ignored or
hidden by the waitTopicDeletion timeout. It now calls get() and wraps
the cause in a RuntimeException, so callers no longer need to handle
ExecutionException and InterruptedException.

- Use ClusterInstance#deleteTopic in tests that delete a topic and then
immediately wait for the deletion.
- Rename ClusterTestExtensionsTest#testVerifyTopicDeletion to
testDeleteTopic and verify that deleting a non-existent topic fails with
UnknownTopicOrPartitionException.

Reviewers: Ken Huang <s7133700@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/MetricsDuringTopicCreationDeletionTest.java

@@ -31,7 +31,6 @@ import java.io.Closeable;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -125,7 +124,7 @@ public class MetricsDuringTopicCreationDeletionTest {
                 + ", but got: " + underReplicatedPartitionsCount.get());
     }
 
-    private void createAndDeleteTopics() throws InterruptedException, ExecutionException {
+    private void createAndDeleteTopics() {
         for (int i = 1; i <= CREATE_DELETE_ITERATIONS && running; i++) {
             for (String topic : topics) {
                 if (!running) return;

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DeleteTopicTest.java

@@ -78,62 +78,49 @@ public class DeleteTopicTest {
     })
     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);
-            TopicPartition topicPartition = new TopicPartition(testTopic, 0);
-            waitForReplicaCreated(cluster.brokers(), topicPartition, "Replicas were not created");
+        cluster.createTopic(testTopic, 1, (short) 3);
+        TopicPartition topicPartition = new TopicPartition(testTopic, 0);
+        waitForReplicaCreated(cluster.brokers(), topicPartition, "Replicas were not created");
 
-            // Delete topic
-            DeleteTopicsResult deleteResult = admin.deleteTopics(List.of(testTopic));
-            deleteResult.all().get();
-
-            // Wait for topic deletion
-            cluster.waitTopicDeletion(testTopic);
-            // Ensure that the topic directories are hard-deleted
-            TestUtils.waitForCondition(() -> topicDirectories(cluster, testTopic).isEmpty(),
-                "Failed to hard-delete the delete directory");
-        }
+        // Delete topic and wait for topic deletion
+        cluster.deleteTopic(testTopic);
+        // Ensure that the topic directories are hard-deleted
+        TestUtils.waitForCondition(() -> topicDirectories(cluster, testTopic).isEmpty(),
+            "Failed to hard-delete the delete directory");
     }
 
     @ClusterTest(brokers = 1, serverProperties = {
         @ClusterConfigProperty(key = "log.segment.delete.delay.ms", value = "3600000")
     })
     public void testDeleteTopicFilesAfterBrokerRestart(ClusterInstance cluster) throws Exception {
-        try (Admin admin = cluster.admin()) {
-            cluster.createTopic(DEFAULT_TOPIC, 1, (short) 1);
-            TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
-            waitForReplicaCreated(cluster.brokers(), topicPartition, "Replica was not created");
+        cluster.createTopic(DEFAULT_TOPIC, 1, (short) 1);
+        TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
+        waitForReplicaCreated(cluster.brokers(), topicPartition, "Replica was not created");
 
-            admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
-            cluster.waitTopicDeletion(DEFAULT_TOPIC);
-            TestUtils.waitForCondition(() -> {
-                List<File> directories = topicDirectories(cluster, DEFAULT_TOPIC);
-                return directories.size() == 1 && directories.stream().allMatch(dir -> dir.getName().endsWith(DELETE_DIR_SUFFIX));
-            }, "Topic directory was not renamed for deletion");
+        cluster.deleteTopic(DEFAULT_TOPIC);
+        TestUtils.waitForCondition(() -> {
+            List<File> directories = topicDirectories(cluster, DEFAULT_TOPIC);
+            return directories.size() == 1 && directories.stream().allMatch(dir -> dir.getName().endsWith(DELETE_DIR_SUFFIX));
+        }, "Topic directory was not renamed for deletion");
 
-            // The long deletion delay keeps the renamed directories on disk until the broker restarts.
-            cluster.shutdownBroker(0);
-            List<File> deletedDirectories = topicDirectories(cluster, DEFAULT_TOPIC);
-            assertEquals(1, deletedDirectories.size(), "Topic directory should remain on disk until the broker restarts");
-            deletedDirectories.forEach(dir -> assertTrue(dir.getName().endsWith(DELETE_DIR_SUFFIX)));
+        // The long deletion delay keeps the renamed directories on disk until the broker restarts.
+        cluster.shutdownBroker(0);
+        List<File> deletedDirectories = topicDirectories(cluster, DEFAULT_TOPIC);
+        assertEquals(1, deletedDirectories.size(), "Topic directory should remain on disk until the broker restarts");
+        deletedDirectories.forEach(dir -> assertTrue(dir.getName().endsWith(DELETE_DIR_SUFFIX)));
 
-            cluster.restartBroker(0, Map.of(
-                LOG_INITIAL_TASK_DELAY_MS_CONFIG, "100",
-                LOG_DELETE_DELAY_MS_CONFIG, "1000"
-            ));
-            TestUtils.waitForCondition(() -> deletedDirectories.stream().noneMatch(File::exists),
-                "Topic directories were not deleted from disk after the broker restarted");
-        }
+        cluster.restartBroker(0, Map.of(
+            LOG_INITIAL_TASK_DELAY_MS_CONFIG, "100",
+            LOG_DELETE_DELAY_MS_CONFIG, "1000"
+        ));
+        TestUtils.waitForCondition(() -> deletedDirectories.stream().noneMatch(File::exists),
+            "Topic directories were not deleted from disk after the broker restarted");
     }
 
     @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);
-        }
+    public void testDeleteTopicWithAllAliveReplicas(ClusterInstance cluster) {
+        cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment);
+        cluster.deleteTopic(DEFAULT_TOPIC);
     }
 
     @ClusterTest
@@ -254,15 +241,12 @@ public class DeleteTopicTest {
 
     @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
@@ -293,23 +277,20 @@ public class DeleteTopicTest {
         @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.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.waitTopicDeletion(DEFAULT_TOPIC);
-        }
+        cluster.deleteTopic(DEFAULT_TOPIC);
     }
 
     @ClusterTest

test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java

@@ -361,12 +361,15 @@ public interface ClusterInstance {
      * Deletes a topic and waits for the deletion to complete.
      *
      * @param topicName The name of the topic to delete
-     * @throws InterruptedException If the operation is interrupted
      */
-    default void deleteTopic(String topicName) throws InterruptedException, ExecutionException {
+    default void deleteTopic(String topicName) {
         try (Admin admin = admin()) {
-            admin.deleteTopics(List.of(topicName));
+            admin.deleteTopics(List.of(topicName)).all().get();
             waitTopicDeletion(topicName);
+        } catch (InterruptedException e) {
+            throw new InterruptException(e);
+        } catch (ExecutionException e) {
+            throw new RuntimeException("Failed to delete topic " + topicName, e.getCause());
         }
     }
 

test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/junit/ClusterTestExtensionsTest.java

@@ -41,6 +41,7 @@ import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.errors.ClusterAuthorizationException;
 import org.apache.kafka.common.errors.SaslAuthenticationException;
 import org.apache.kafka.common.errors.TopicAuthorizationException;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
@@ -292,15 +293,17 @@ public class ClusterTestExtensionsTest {
 
 
     @ClusterTest(types = {Type.CO_KRAFT, Type.KRAFT}, brokers = 4)
-    public void testVerifyTopicDeletion(ClusterInstance clusterInstance) throws Exception {
+    public void testDeleteTopic(ClusterInstance clusterInstance) throws Exception {
         String testTopic = "testTopic";
         clusterInstance.createTopic(testTopic, 1, (short) 1);
         try (Admin admin = clusterInstance.admin()) {
-            admin.deleteTopics(List.of(testTopic));
-            clusterInstance.waitTopicDeletion(testTopic);
+            clusterInstance.deleteTopic(testTopic);
             Assertions.assertTrue(admin.listTopics().listings().get().stream().noneMatch(
                     topic -> topic.name().equals(testTopic)
             ));
+
+            assertInstanceOf(UnknownTopicOrPartitionException.class,
+                assertThrows(RuntimeException.class, () -> clusterInstance.deleteTopic("non-existent-topic")).getCause());
         }
     }
 

tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java

@@ -380,8 +380,7 @@ public class TopicCommandTest {
             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());
         }
@@ -405,8 +404,7 @@ public class TopicCommandTest {
             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());
         }

tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java

@@ -439,7 +439,7 @@ public class DescribeConsumerGroupTest {
                 }
                 // remove previous consumer groups, so we can have a clean cluster for next consumer group protocol test.
                 deleteConsumerGroups(groups);
-                deleteTopic(topic);
+                clusterInstance.deleteTopic(topic);
             }
         }
     }
@@ -1174,12 +1174,6 @@ public class DescribeConsumerGroupTest {
         }
     }
 
-    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<TopicPartition> topicPartitions, Map<String, Object> customConfigs) {
         Map<String, Object> configs = composeConfigs(
                 groupId,