fix(tools/perf): only delete test topics on reset

Signed-off-by: Ning Yu <[email protected]>

Author: Chillax-0v0

tools/src/main/java/org/apache/kafka/tools/automq/PerfCommand.java

@@ -95,9 +95,9 @@ private void run() {
         TimerUtil timer = new TimerUtil();
 
         if (config.reset) {
-            LOGGER.info("Deleting all topics...");
+            LOGGER.info("Deleting all test topics...");
             int deleted = topicService.deleteTopics();
-            LOGGER.info("Deleted all topics ({} in total), took {} ms", deleted, timer.elapsedAndResetAs(TimeUnit.MILLISECONDS));
+            LOGGER.info("Deleted all test topics ({} in total), took {} ms", deleted, timer.elapsedAndResetAs(TimeUnit.MILLISECONDS));
         }
 
         LOGGER.info("Creating topics...");

tools/src/main/java/org/apache/kafka/tools/automq/perf/TopicService.java

@@ -29,7 +29,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -87,10 +86,11 @@ public List<Topic> createTopics(TopicsConfig config) {
     public int deleteTopics() {
         ListTopicsResult result = admin.listTopics();
         try {
-            Set<String> topics = result.names().get();
-            topics.removeIf(name -> name.startsWith(COMMON_TOPIC_PREFIX));
-            admin.deleteTopics(topics).all().get();
-            return topics.size();
+            List<String> toDelete = result.names().get().stream()
+                .filter(name -> name.startsWith(COMMON_TOPIC_PREFIX))
+                .collect(Collectors.toList());
+            admin.deleteTopics(toDelete).all().get();
+            return toDelete.size();
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
         } catch (ExecutionException ignored) {