Repository: apache/kafka
File: tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java (Java)

Code before:
            // prepare data for partitions with leaders
            topicPartitions.removeAll(nonLeaderTopicPartitions);
            List<PartitionAssignmentState> existLeaderAssignments = offsetsUtils.getLogEndOffsets(topicPartitions).entrySet().stream().map(logEndOffsetResult -> {
                if (logEndOffsetResult.getValue() instanceof OffsetsUtils.LogOffset)
                    return getDescribePartitionResult.apply(
                        logEndOffsetResult.getKey(),
                        Optional.of(((OffsetsUtils.LogOffset) logEndOffsetResult.getValue()).value())
                    );
                else if (logEndOffsetResult.getValue() instanceof OffsetsUtils.Unknown)
                    return getDescribePartitionResult.apply(logEndOffsetResult.getKey(), Optional.empty());
                else if (logEndOffsetResult.getValue() instanceof OffsetsUtils.Ignore)
                    return null;

                throw new IllegalStateException("Unknown LogOffset subclass: " + logEndOffsetResult.getValue());
            }).toList();

            // prepare data for partitions without leaders
            List<PartitionAssignmentState> noneLeaderAssignments = nonLeaderTopicPartitions.stream()
                    .map(tp -> getDescribePartitionResult.apply(tp, Optional.empty())).toList();

            // concat the data and then sort them
            return Stream.concat(existLeaderAssignments.stream(), noneLeaderAssignments.stream())
                    .sorted(Comparator.<PartitionAssignmentState, String>comparing(
                            state -> state.topic.orElse(""), String::compareTo)
                            .thenComparingInt(state -> state.partition.orElse(-1)))
                    .collect(Collectors.toList());
        }

        private Set<TopicPartition> filterNoneLeaderPartitions(List<TopicPartition> topicPartitions) {
            // collect all topics
            Set<String> topics = topicPartitions.stream().map(TopicPartition::topic).collect(Collectors.toSet());

            try {
                return adminClient.describeTopics(topics).allTopicNames().get().entrySet()
                        .stream()
                        .flatMap(entry -> entry.getValue().partitions().stream()
                                .filter(partitionInfo -> partitionInfo.leader() == null)
                                .map(partitionInfo -> new TopicPartition(entry.getKey(), partitionInfo.partition())))
                        .collect(Collectors.toSet());
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        Map<String, Map<TopicPartition, OffsetAndMetadata>> resetOffsets() {
            List<String> groupIds = opts.options.has(opts.allGroupsOpt)
                ? listConsumerGroups()
                : opts.options.valuesOf(opts.groupOpt);

            Map<String, KafkaFuture<ConsumerGroupDescription>> consumerGroups = adminClient.describeConsumerGroups(
                groupIds,

Review comment (refactor):
I don't think we need to convert it to a set here, since AdminClient.describeTopics doesn’t handle duplicate topic names anyway https://github.com/apache/kafka/blob/a5a54dc32b41a791646feee579207323c02ffa5b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java#L2320

Code after:
                        logEndOffsetResult.getKey(),
                        Optional.of(((OffsetsUtils.LogOffset) logEndOffsetResult.getValue()).value())
                    );
                else if (logEndOffsetResult.getValue() instanceof OffsetsUtils.Unknown)
                    return getDescribePartitionResult.apply(logEndOffsetResult.getKey(), Optional.empty());
                else if (logEndOffsetResult.getValue() instanceof OffsetsUtils.Ignore)
                    return null;

                throw new IllegalStateException("Unknown LogOffset subclass: " + logEndOffsetResult.getValue());
            }).toList();

            // prepare data for partitions without leaders
            List<PartitionAssignmentState> noneLeaderAssignments = topicPartitionsWithoutLeader.stream()
                    .map(tp -> getDescribePartitionResult.apply(tp, Optional.empty())).toList();

            // concat the data and then sort them
            return Stream.concat(existLeaderAssignments.stream(), noneLeaderAssignments.stream())
                    .sorted(Comparator.<PartitionAssignmentState, String>comparing(
                            state -> state.topic.orElse(""), String::compareTo)
                            .thenComparingInt(state -> state.partition.orElse(-1)))
                    .collect(Collectors.toList());
        }

        private List<TopicPartition> filterNoneLeaderPartitions(Collection<TopicPartition> topicPartitions) {
            // collect all topics
            Set<String> topics = topicPartitions.stream().map(TopicPartition::topic).collect(Collectors.toSet());

            try {
                return adminClient.describeTopics(topics).allTopicNames().get().entrySet()
                        .stream()
                        .flatMap(entry -> entry.getValue().partitions().stream()
                                .filter(partitionInfo -> partitionInfo.leader() == null)
                                .map(partitionInfo -> new TopicPartition(entry.getKey(), partitionInfo.partition())))
                        .toList();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        Map<String, Map<TopicPartition, OffsetAndMetadata>> resetOffsets() {
            List<String> groupIds = opts.options.has(opts.allGroupsOpt)
                ? listConsumerGroups()
                : opts.options.valuesOf(opts.groupOpt);

            Map<String, KafkaFuture<ConsumerGroupDescription>> consumerGroups = adminClient.describeConsumerGroups(
                groupIds,
                withTimeoutMs(new DescribeConsumerGroupsOptions())
            ).describedGroups();

            Map<String, Map<TopicPartition, OffsetAndMetadata>> result = new HashMap<>();