Repository: apache/kafka
File: tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java (Java)

Code before:
                    internalTopics.removeIf(topic -> {
                        if (!isInferredInternalTopic(topic, description.groupId())) {
                            printError("The internal topic '" + topic + "' is not inferred as internal " +
                                "and thus will not be deleted with the group '" + description.groupId() + "'.", Optional.empty());
                            return true;
                        }
                        return false;
                    });
                    if (!internalTopics.isEmpty()) {
                        groupToInternalTopics.put(description.groupId(), internalTopics);
                    }
                }
            } 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()));
                } else {
                    printError("Retrieving internal topics failed due to " + e.getMessage(), Optional.of(e));
                }
            }
            return groupToInternalTopics;
        }

        private Map<TopicPartition, OffsetAndMetadata> resetOffsetsForInactiveGroup(String groupId, boolean dryRun) {
            try {
                Collection<TopicPartition> partitionsToReset = getPartitionsToReset(groupId);
                Map<TopicPartition, OffsetAndMetadata> preparedOffsets = prepareOffsetsToReset(groupId, partitionsToReset);
                if (!dryRun) {
                    adminClient.alterStreamsGroupOffsets(
                        groupId,
                        preparedOffsets
                    ).all().get();
                }

                return preparedOffsets;
            } catch (InterruptedException ie) {

Review comment (suggestion):
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 + ").");
```

Code after:
                    internalTopics.removeIf(topic -> {
                        if (!isInferredInternalTopic(topic, description.groupId())) {
                            printError("The internal topic '" + topic + "' is not inferred as internal " +
                                "and thus will not be deleted with the group '" + description.groupId() + "'.", Optional.empty());
                            return true;
                        }
                        return false;
                    });
                    if (!internalTopics.isEmpty()) {
                        groupToInternalTopics.put(description.groupId(), internalTopics);
                    }
                }
            } catch (InterruptedException | ExecutionException e) {
                if (e.getCause() instanceof UnsupportedVersionException) {
                    try {
                        // Retrieve internal topic list if possible, and add the list of topic names to error message
                        Set<String> allTopics = adminClient.listTopics().names().get();
                        List<String> internalTopics = allTopics.stream()
                            .filter(topic -> groupIds.stream().anyMatch(groupId -> isInferredInternalTopic(topic, groupId)))
                            .collect(Collectors.toList());
                        printError("Retrieving internal topics is not supported by the broker version.\n" +
                            "Internal topics: (" + String.join(",", internalTopics) + ").\n" +
                            "Use 'kafka-topics.sh' to delete the group's internal topics.", Optional.of(e.getCause()));
                    } catch (InterruptedException | ExecutionException 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()));
                    }
                } else {
                    printError("Retrieving internal topics failed due to " + e.getMessage(), Optional.of(e));
                }
            }
            return groupToInternalTopics;
        }

        private Map<TopicPartition, OffsetAndMetadata> resetOffsetsForInactiveGroup(String groupId, boolean dryRun) {
            try {
                Collection<TopicPartition> partitionsToReset = getPartitionsToReset(groupId);
                Map<TopicPartition, OffsetAndMetadata> preparedOffsets = prepareOffsetsToReset(groupId, partitionsToReset);
                if (!dryRun) {
                    adminClient.alterStreamsGroupOffsets(
                        groupId,
                        preparedOffsets
                    ).all().get();
                }

                return preparedOffsets;
            } catch (InterruptedException ie) {
                throw new RuntimeException(ie);
            } catch (ExecutionException ee) {
                Throwable cause = ee.getCause();
                if (cause instanceof KafkaException) {