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

Code before:
            } else if (opts.options.has(opts.resetToEarliestOpt)) {
                return offsetsUtils.resetToEarliest(partitionsToReset);
            } else if (opts.options.has(opts.resetToLatestOpt)) {
                return offsetsUtils.resetToLatest(partitionsToReset);
            } else if (opts.options.has(opts.resetShiftByOpt)) {
                Map<TopicPartition, OffsetAndMetadata> currentCommittedOffsets = getCommittedOffsets(groupId);
                return offsetsUtils.resetByShiftBy(partitionsToReset, currentCommittedOffsets);
            } else if (opts.options.has(opts.resetToDatetimeOpt)) {
                return offsetsUtils.resetToDateTime(partitionsToReset);
            } else if (opts.options.has(opts.resetByDurationOpt)) {
                return offsetsUtils.resetByDuration(partitionsToReset);
            } else if (offsetsUtils.resetPlanFromFile().isPresent()) {
                return offsetsUtils.resetFromFile(groupId);
            } else if (opts.options.has(opts.resetToCurrentOpt)) {
                Map<TopicPartition, OffsetAndMetadata> currentCommittedOffsets = getCommittedOffsets(groupId);
                return offsetsUtils.resetToCurrent(partitionsToReset, currentCommittedOffsets);
            }

            CommandLineUtils.printUsageAndExit(opts.parser, String.format("Option '%s' requires one of the following scenarios: %s", opts.resetOffsetsOpt, opts.allResetOffsetScenarioOpts));
            return null;
        }

        private void checkAllTopicPartitionsHaveLeader(Collection<TopicPartition> partitionsToReset) {
            List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);
            if (!partitionsWithoutLeader.isEmpty()) {
                // append the TopicPartition list string
                StringBuilder partitionStr = new StringBuilder();
                for (TopicPartition topicPartition : partitionsWithoutLeader) {
                    partitionStr.append(topicPartition.toString()).append(",");
                }
                partitionStr.deleteCharAt(partitionStr.length() - 1);

                // throw exception
                throw new LeaderNotAvailableException("The partitions \"" + partitionStr + "\" have no leader");
            }
        }

        String exportOffsetsToCsv(Map<String, Map<TopicPartition, OffsetAndMetadata>> assignments) {
            boolean isSingleGroupQuery = opts.options.valuesOf(opts.groupOpt).size() == 1;
            ObjectWriter csvWriter = isSingleGroupQuery
                ? CsvUtils.writerFor(CsvUtils.CsvRecordNoGroup.class)
                : CsvUtils.writerFor(CsvUtils.CsvRecordWithGroup.class);

            return assignments.entrySet().stream().flatMap(e -> {
                String groupId = e.getKey();
                Map<TopicPartition, OffsetAndMetadata> partitionInfo = e.getValue();

                return partitionInfo.entrySet().stream().map(e1 -> {
                    TopicPartition k = e1.getKey();
                    OffsetAndMetadata v = e1.getValue();
                    Object csvRecord = isSingleGroupQuery

Review comment (suggestion):
nit: You could do this string concatenation much more neatly with something like
```
String partitionStr = partitionsWithoutLeader.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
```

Code after:
                return offsetsUtils.resetToOffset(partitionsToReset);
            } else if (opts.options.has(opts.resetToEarliestOpt)) {
                return offsetsUtils.resetToEarliest(partitionsToReset);
            } else if (opts.options.has(opts.resetToLatestOpt)) {
                return offsetsUtils.resetToLatest(partitionsToReset);
            } else if (opts.options.has(opts.resetShiftByOpt)) {
                Map<TopicPartition, OffsetAndMetadata> currentCommittedOffsets = getCommittedOffsets(groupId);
                return offsetsUtils.resetByShiftBy(partitionsToReset, currentCommittedOffsets);
            } else if (opts.options.has(opts.resetToDatetimeOpt)) {
                return offsetsUtils.resetToDateTime(partitionsToReset);
            } else if (opts.options.has(opts.resetByDurationOpt)) {
                return offsetsUtils.resetByDuration(partitionsToReset);
            } else if (offsetsUtils.resetPlanFromFile().isPresent()) {
                return offsetsUtils.resetFromFile(groupId);
            } else if (opts.options.has(opts.resetToCurrentOpt)) {
                Map<TopicPartition, OffsetAndMetadata> currentCommittedOffsets = getCommittedOffsets(groupId);
                return offsetsUtils.resetToCurrent(partitionsToReset, currentCommittedOffsets);
            }

            CommandLineUtils.printUsageAndExit(opts.parser, String.format("Option '%s' requires one of the following scenarios: %s", opts.resetOffsetsOpt, opts.allResetOffsetScenarioOpts));
            return null;
        }

        private void checkAllTopicPartitionsValid(Collection<TopicPartition> partitionsToReset) {
            // check the partitions exist
            List<TopicPartition> partitionsNotExistList = filterNonExistentPartitions(partitionsToReset);
            if (!partitionsNotExistList.isEmpty()) {
                String partitionStr = partitionsNotExistList.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
                throw new UnknownTopicOrPartitionException("The partitions \"" + partitionStr + "\" do not exist");
            }

            // check the partitions have leader
            List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);
            if (!partitionsWithoutLeader.isEmpty()) {
                String partitionStr = partitionsWithoutLeader.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
                throw new LeaderNotAvailableException("The partitions \"" + partitionStr + "\" have no leader");
            }
        }

        private List<TopicPartition> filterNonExistentPartitions(Collection<TopicPartition> topicPartitions) {
            // collect all topics
            Set<String> topics = topicPartitions.stream().map(TopicPartition::topic).collect(Collectors.toSet());
            try {
                List<TopicPartition> existPartitions = adminClient.describeTopics(topics).allTopicNames().get().entrySet()
                        .stream()
                        .flatMap(entry -> entry.getValue().partitions().stream()
                                .map(partitionInfo -> new TopicPartition(entry.getKey(), partitionInfo.partition())))
                        .toList();

                return topicPartitions.stream().filter(element -> !existPartitions.contains(element)).toList();
            } catch (InterruptedException | ExecutionException e) {