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:
us...@infra.apache.org
