TreeMap<String, Entry<Optional<ConsumerGroupState>, Optional<Collection<PartitionAssignmentState>>>> collectGroupsOffsets(Collection<String> groupIds) throws Exception {
            Map<String, ConsumerGroupDescription> consumerGroups = describeConsumerGroups(groupIds);
            TreeMap<String, Entry<Optional<ConsumerGroupState>, Optional<Collection<PartitionAssignmentState>>>> groupOffsets = new TreeMap<>();

            consumerGroups.forEach((groupId, consumerGroup) -> {
                ConsumerGroupState state = consumerGroup.state();
                Map<TopicPartition, OffsetAndMetadata> committedOffsets = getCommittedOffsets(groupId);
                // The admin client returns `null` as a value to indicate that there is not committed offset for a partition.
                Function<TopicPartition, Optional<Long>> getPartitionOffset = tp -> Optional.ofNullable(committedOffsets.get(tp)).map(OffsetAndMetadata::offset);
                List<TopicPartition> assignedTopicPartitions = new ArrayList<>();
                Comparator<MemberDescription> comparator =
                    Comparator.<MemberDescription>comparingInt(m -> m.assignment().topicPartitions().size()).reversed();
                List<PartitionAssignmentState> rowsWithConsumer = new ArrayList<>();
                consumerGroup.members().stream().filter(m -> !m.assignment().topicPartitions().isEmpty())
                    .sorted(comparator)
                    .forEach(consumerSummary -> {
                        Set<TopicPartition> topicPartitions = consumerSummary.assignment().topicPartitions();
                        assignedTopicPartitions.addAll(topicPartitions);
                        rowsWithConsumer.addAll(collectConsumerAssignment(
                            groupId,
                            Optional.of(consumerGroup.coordinator()),
                            topicPartitions,
                            getPartitionOffset,
                            Optional.of(consumerSummary.consumerId()),
                            Optional.of(consumerSummary.host()),
                            Optional.of(consumerSummary.clientId()))
                        );
                    });
                Map<TopicPartition, OffsetAndMetadata> unassignedPartitions = new HashMap<>();
                committedOffsets.entrySet().stream().filter(e -> !assignedTopicPartitions.contains(e.getKey()))
                    .forEach(e -> unassignedPartitions.put(e.getKey(), e.getValue()));
                Collection<PartitionAssignmentState> rowsWithoutConsumer = !unassignedPartitions.isEmpty()
                    ? collectConsumerAssignment(
                        groupId,
                        Optional.of(consumerGroup.coordinator()),
                        unassignedPartitions.keySet(),
                        getPartitionOffset,
                        Optional.of(MISSING_COLUMN_VALUE),
                        Optional.of(MISSING_COLUMN_VALUE),
                        Optional.of(MISSING_COLUMN_VALUE))
                    : Collections.emptyList();

                rowsWithConsumer.addAll(rowsWithoutConsumer);

                groupOffsets.put(groupId, new SimpleImmutableEntry<>(Optional.of(state), Optional.of(rowsWithConsumer)));
            });

            return groupOffsets;
        }
Returns states of the specified consumer groups and partition assignment states
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java