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

Code before:
                            subtopology.stateChangelogTopics().keySet().stream()))
                        .filter(topic -> !sourceTopics.contains(topic))
                        .collect(Collectors.toList());
                    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 list and 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);

Review comment (suggestion):
Since the topics are already listed, please remove the `to list` from the error message.

Code after:
                            subtopology.stateChangelogTopics().keySet().stream()))
                        .filter(topic -> !sourceTopics.contains(topic))
                        .collect(Collectors.toList());
                    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);