frankvicky commented on code in PR #20157:
URL: https://github.com/apache/kafka/pull/20157#discussion_r2202450911
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -830,6 +831,21 @@ Map<String, List<String>>
retrieveInternalTopics(List<String> groupIds) {
}
} catch (InterruptedException | ExecutionException e) {
if (e.getCause() instanceof UnsupportedVersionException) {
+ List<String> internalTopics = new ArrayList<>();
+ try {
+ Set<String> allTopics =
adminClient.listTopics().names().get();
+ for (String topic : allTopics) {
+ for (String groupId : groupIds) {
+ if (isInferredInternalTopic(topic, groupId)) {
+ internalTopics.add(topic);
+ break;
+ }
+ }
+ }
+ System.out.println("Internal Topics: (" +
String.join(",", internalTopics) + ").");
+ } catch (InterruptedException | ExecutionException ex) {
+ printError("Retrieving internal topics failed due to "
+ ex.getMessage(), Optional.of(ex));
+ }
printError("Retrieving internal topics is not supported by
the broker version. " +
"Use 'kafka-topics.sh' to list and delete the group's
internal topics.", Optional.of(e.getCause()));
Review Comment:
Should we combine the internal topics message into `printError`?
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -830,6 +831,21 @@ Map<String, List<String>>
retrieveInternalTopics(List<String> groupIds) {
}
} catch (InterruptedException | ExecutionException e) {
if (e.getCause() instanceof UnsupportedVersionException) {
+ List<String> internalTopics = new ArrayList<>();
+ try {
+ Set<String> allTopics =
adminClient.listTopics().names().get();
+ for (String topic : allTopics) {
+ for (String groupId : groupIds) {
+ if (isInferredInternalTopic(topic, groupId)) {
+ internalTopics.add(topic);
+ break;
+ }
+ }
+ }
+ System.out.println("Internal Topics: (" +
String.join(",", internalTopics) + ").");
Review Comment:
We could use the stream api to align the style with this whole class.
For example:
```suggestion
String internalTopics;
try {
Set<String> allTopics =
adminClient.listTopics().names().get();
internalTopics = allTopics.stream()
.filter(topic ->
groupIds.stream().anyMatch(groupId -> isInferredInternalTopic(topic, groupId)))
.collect(Collectors.joining(","));
System.out.println("Internal Topics: (" +
internalTopics + ").");
```
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
