[GitHub] [kafka] wcarlson5 commented on a change in pull request #8923: KAFKA-6435: KIP-623 Add internal topics option to streamResetter

[GitBox <gi...@apache.org>]

wcarlson5 commented on a change in pull request #8923:
URL: https://github.com/apache/kafka/pull/8923#discussion_r614166802



##########
File path: core/src/main/scala/kafka/tools/StreamsResetter.java
##########
@@ -248,7 +257,7 @@ private void parseArguments(final String[] args) {
             .ofType(String.class)
             .describedAs("file name");
         forceOption = optionParser.accepts("force", "Force the removal of members of the consumer group (intended to remove stopped members if a long session timeout was used). " +
-                "Make sure to shut down all stream applications when this option is specified to avoid unexpected rebalances.");

Review comment:
       unnecessary

##########
File path: core/src/main/scala/kafka/tools/StreamsResetter.java
##########
@@ -225,6 +229,11 @@ private void parseArguments(final String[] args) {
             .ofType(String.class)
             .withValuesSeparatedBy(',')
             .describedAs("list");
+        internalTopicsOption = optionParser.accepts("internal-topics", "Comma-separated list of internal topics to delete. Must be a subset of the internal topics marked for deletion by the default behaviour (do a dry-run without this option to view these topics).")

Review comment:
       line is too long

##########
File path: core/src/main/scala/kafka/tools/StreamsResetter.java
##########
@@ -642,22 +651,38 @@ private boolean isIntermediateTopic(final String topic) {
         return options.valuesOf(intermediateTopicsOption).contains(topic);
     }
 
-    private void maybeDeleteInternalTopics(final Admin adminClient, final boolean dryRun) {
-        System.out.println("Deleting all internal/auto-created topics for application " + options.valueOf(applicationIdOption));
-        final List<String> topicsToDelete = new ArrayList<>();
-        for (final String listing : allTopics) {
-            if (isInternalTopic(listing)) {
-                if (!dryRun) {
-                    topicsToDelete.add(listing);
-                } else {
-                    System.out.println("Topic: " + listing);
-                }
+    private int maybeDeleteInternalTopics(final Admin adminClient, final boolean dryRun) {
+        final List<String> inferredInternalTopics = allTopics.stream()
+                .filter(this::isInferredInternalTopic)
+                .collect(Collectors.toList());
+        final List<String> specifiedInternalTopics = options.valuesOf(internalTopicsOption);
+        final List<String> topicsToDelete;
+
+        if (!specifiedInternalTopics.isEmpty()) {
+            final List<String> notFoundInternalTopics = specifiedInternalTopics.stream()
+                    .filter(topic -> !inferredInternalTopics.contains(topic))
+                    .collect(Collectors.toList());
+            if (!notFoundInternalTopics.isEmpty()) {

Review comment:
       something like `inferredInternalTopics.containsAll(specifiedInternalTopics)` might be easier to understand here




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org