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

Code before:
                : opts.options.valuesOf(opts.groupOpt);
            if (!groupIds.isEmpty()) {
                Map<String, KafkaFuture<StreamsGroupDescription>> streamsGroups = adminClient.describeStreamsGroups(
                    groupIds
                ).describedGroups();

                streamsGroups.forEach((groupId, groupDescription) -> {
                    try {
                        String state = groupDescription.get().groupState().toString();
                        switch (state) {
                            case "Empty":
                            case "Dead":
                                // reset offsets in source topics
                                result.put(groupId, resetOffsetsForInactiveGroup(groupId, dryRun));
                                // delete internal topics
                                if (!dryRun) {
                                    List<String> internalTopics = getInternalTopicsToBeDeleted(groupId);
                                    if (!internalTopics.isEmpty()) {
                                        try {
                                            adminClient.deleteTopics(internalTopics).all().get();
                                        } catch (InterruptedException | ExecutionException e) {
                                            if (e.getCause() instanceof UnknownTopicOrPartitionException) {
                                                printError("Deleting internal topics for group '" + groupId + "' failed because the topics do not exist.", Optional.empty());
                                            } else if (e.getCause() instanceof UnsupportedVersionException) {
                                                printError("Deleting internal topics is not supported by the broker version.\n" +
                                                    "Topics: (" + String.join(",", internalTopics) + ").\n" +
                                                    "Use 'kafka-topics.sh' to delete the group's internal topics.", Optional.of(e.getCause()));
                                            } else {
                                                printError("Deleting internal topics for group '" + groupId + "' failed due to " + e.getMessage(), Optional.of(e));
                                            }
                                        }
                                    }
                                }
                                break;
                            default:
                                printError("Assignments can only be reset if the group '" + groupId + "' is inactive, but the current state is " + state + ".", Optional.empty());
                                result.put(groupId, Collections.emptyMap());
                        }
                    } catch (InterruptedException ie) {
                        throw new RuntimeException(ie);
                    } catch (ExecutionException ee) {
                        if (ee.getCause() instanceof GroupIdNotFoundException) {
                            result.put(groupId, resetOffsetsForInactiveGroup(groupId, dryRun));
                        } else {
                            throw new RuntimeException(ee);
                        }
                    }
                });
            }
            return result;
        }

Review comment (suggestion):
Add `Internal` here as well?! Like the other error message

Code after:
                : opts.options.valuesOf(opts.groupOpt);
            if (!groupIds.isEmpty()) {
                Map<String, KafkaFuture<StreamsGroupDescription>> streamsGroups = adminClient.describeStreamsGroups(
                    groupIds
                ).describedGroups();

                streamsGroups.forEach((groupId, groupDescription) -> {
                    try {
                        String state = groupDescription.get().groupState().toString();
                        switch (state) {
                            case "Empty":
                            case "Dead":
                                // reset offsets in source topics
                                result.put(groupId, resetOffsetsForInactiveGroup(groupId, dryRun));
                                // delete internal topics
                                if (!dryRun) {
                                    List<String> internalTopics = getInternalTopicsToBeDeleted(groupId);
                                    if (!internalTopics.isEmpty()) {
                                        try {
                                            adminClient.deleteTopics(internalTopics).all().get();
                                        } catch (InterruptedException | ExecutionException e) {
                                            if (e.getCause() instanceof UnknownTopicOrPartitionException) {
                                                printError("Deleting internal topics for group '" + groupId + "' failed because the topics do not exist.", Optional.empty());
                                            } else if (e.getCause() instanceof UnsupportedVersionException) {
                                                printError("Deleting 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()));
                                            } else {
                                                printError("Deleting internal topics for group '" + groupId + "' failed due to " + e.getMessage(), Optional.of(e));
                                            }
                                        }
                                    }
                                }
                                break;
                            default:
                                printError("Assignments can only be reset if the group '" + groupId + "' is inactive, but the current state is " + state + ".", Optional.empty());
                                result.put(groupId, Collections.emptyMap());
                        }
                    } catch (InterruptedException ie) {
                        throw new RuntimeException(ie);
                    } catch (ExecutionException ee) {
                        if (ee.getCause() instanceof GroupIdNotFoundException) {
                            result.put(groupId, resetOffsetsForInactiveGroup(groupId, dryRun));
                        } else {
                            throw new RuntimeException(ee);
                        }
                    }
                });
            }
            return result;
        }