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

Code before:

                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 && topicPartitions.contains(new TopicPartition(entry.getKey(), partitionInfo.partition())))
                                .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<>();

            consumerGroups.forEach((groupId, groupDescription) -> {
                try {
                    String state = groupDescription.get().groupState().toString();
                    switch (state) {
                        case "Empty":
                        case "Dead":
                            result.put(groupId, resetOffsetsForInactiveGroup(groupId));

Review comment (suggestion):
Please add the filter after `map` to avoid creating extra `TopicPartition` object.
```java
                                .filter(partitionInfo -> partitionInfo.leader() == null)
                                .map(partitionInfo -> new TopicPartition(entry.getKey(), partitionInfo.partition()))
                                .filter(topicPartitions::contains))
```


Code after:
            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<>();

            consumerGroups.forEach((groupId, groupDescription) -> {
                try {
                    String state = groupDescription.get().groupState().toString();
                    switch (state) {
                        case "Empty":
                        case "Dead":
                            result.put(groupId, resetOffsetsForInactiveGroup(groupId));
                            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, Map.of());
                    }
                } catch (InterruptedException ie) {
                    throw new RuntimeException(ie);
                } catch (ExecutionException ee) {
                    if (ee.getCause() instanceof GroupIdNotFoundException) {
                        result.put(groupId, resetOffsetsForInactiveGroup(groupId));
                    } else {
                        throw new RuntimeException(ee);
                    }
                }
            });

            return result;
        }

        private Map<TopicPartition, OffsetAndMetadata> resetOffsetsForInactiveGroup(String groupId) {
            try {
                Collection<TopicPartition> partitionsToReset = getPartitionsToReset(groupId);
                Map<TopicPartition, OffsetAndMetadata> preparedOffsets = prepareOffsetsToReset(groupId, partitionsToReset);

                // Dry-run is the default behavior if --execute is not specified
                boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
                if (!dryRun) {
                    adminClient.alterConsumerGroupOffsets(
                        groupId,
                        preparedOffsets,
                        withTimeoutMs(new AlterConsumerGroupOffsetsOptions())
                    ).all().get();
                }