@SuppressWarnings("CyclomaticComplexity")
        private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId, Collection<TopicPartition> partitionsToReset) {
            if (opts.options.has(opts.resetToOffsetOpt)) {
                long offset = opts.options.valueOf(opts.resetToOffsetOpt);
                return checkOffsetsRange(partitionsToReset.stream().collect(Collectors.toMap(Function.identity(), tp -> offset)))
                    .entrySet().stream().collect(Collectors.toMap(Entry::getKey, e -> new OffsetAndMetadata(e.getValue())));
            } else if (opts.options.has(opts.resetToEarliestOpt)) {
                Map<TopicPartition, LogOffsetResult> logStartOffsets = getLogStartOffsets(partitionsToReset);
                return partitionsToReset.stream().collect(Collectors.toMap(Function.identity(), topicPartition -> {
                    LogOffsetResult logOffsetResult = logStartOffsets.get(topicPartition);

                    if (!(logOffsetResult instanceof LogOffset)) {
                        ToolsUtils.printUsageAndExit(opts.parser, "Error getting starting offset of topic partition: " + topicPartition);
                        return null;
                    }

                    return new OffsetAndMetadata(((LogOffset) logOffsetResult).value);
                }));
            } else if (opts.options.has(opts.resetToLatestOpt)) {
                Map<TopicPartition, LogOffsetResult> logEndOffsets = getLogEndOffsets(partitionsToReset);
                return partitionsToReset.stream().collect(Collectors.toMap(Function.identity(), topicPartition -> {
                    LogOffsetResult logOffsetResult = logEndOffsets.get(topicPartition);

                    if (!(logOffsetResult instanceof LogOffset)) {
                        ToolsUtils.printUsageAndExit(opts.parser, "Error getting ending offset of topic partition: " + topicPartition);
                        return null;
                    }

                    return new OffsetAndMetadata(((LogOffset) logOffsetResult).value);
                }));
            } else if (opts.options.has(opts.resetShiftByOpt)) {
                Map<TopicPartition, OffsetAndMetadata> currentCommittedOffsets = getCommittedOffsets(groupId);
                Map<TopicPartition, Long> requestedOffsets = partitionsToReset.stream().collect(Collectors.toMap(Function.identity(), topicPartition -> {
                    long shiftBy = opts.options.valueOf(opts.resetShiftByOpt);
                    OffsetAndMetadata currentOffset = currentCommittedOffsets.get(topicPartition);

                    if (currentOffset == null) {
                        throw new IllegalArgumentException("Cannot shift offset for partition " + topicPartition + " since there is no current committed offset");
                    }

                    return currentOffset.offset() + shiftBy;
                }));
                return checkOffsetsRange(requestedOffsets).entrySet().stream()
                    .collect(Collectors.toMap(Entry::getKey, e -> new OffsetAndMetadata(e.getValue())));
            } else if (opts.options.has(opts.resetToDatetimeOpt)) {
                try {
                    long timestamp = Utils.getDateTime(opts.options.valueOf(opts.resetToDatetimeOpt));
                    Map<TopicPartition, LogOffsetResult> logTimestampOffsets = getLogTimestampOffsets(partitionsToReset, timestamp);
                    return partitionsToReset.stream().collect(Collectors.toMap(Function.identity(), topicPartition -> {
                        LogOffsetResult logTimestampOffset = logTimestampOffsets.get(topicPartition);

                        if (!(logTimestampOffset instanceof LogOffset)) {
                            ToolsUtils.printUsageAndExit(opts.parser, "Error getting offset by timestamp of topic partition: " + topicPartition);
                            return null;
                        }

                        return new OffsetAndMetadata(((LogOffset) logTimestampOffset).value);
                    }));
                } catch (ParseException e) {
                    throw new RuntimeException(e);
                }
            } else if (opts.options.has(opts.resetByDurationOpt)) {
                String duration = opts.options.valueOf(opts.resetByDurationOpt);
                Duration durationParsed = Duration.parse(duration);
                Instant now = Instant.now();
                durationParsed.negated().addTo(now);
                long timestamp = now.minus(durationParsed).toEpochMilli();
                Map<TopicPartition, LogOffsetResult> logTimestampOffsets = getLogTimestampOffsets(partitionsToReset, timestamp);
                return partitionsToReset.stream().collect(Collectors.toMap(Function.identity(), topicPartition -> {
                    LogOffsetResult logTimestampOffset = logTimestampOffsets.get(topicPartition);

                    if (!(logTimestampOffset instanceof LogOffset)) {
                        ToolsUtils.printUsageAndExit(opts.parser, "Error getting offset by timestamp of topic partition: " + topicPartition);
                        return null;
                    }

                    return new OffsetAndMetadata(((LogOffset) logTimestampOffset).value);
                }));
            } else if (resetPlanFromFile().isPresent()) {
                return resetPlanFromFile().map(resetPlan -> {
                    Map<TopicPartition, OffsetAndMetadata> resetPlanForGroup = resetPlan.get(groupId);

                    if (resetPlanForGroup == null) {
                        printError("No reset plan for group " + groupId + " found", Optional.empty());
                        return Collections.<TopicPartition, OffsetAndMetadata>emptyMap();
                    }

                    Map<TopicPartition, Long> requestedOffsets = resetPlanForGroup.keySet().stream().collect(Collectors.toMap(
                        Function.identity(),
                        topicPartition -> resetPlanForGroup.get(topicPartition).offset()));

                    return checkOffsetsRange(requestedOffsets).entrySet().stream()
                        .collect(Collectors.toMap(Entry::getKey, e -> new OffsetAndMetadata(e.getValue())));
                }).orElseGet(Collections::emptyMap);
            } else if (opts.options.has(opts.resetToCurrentOpt)) {
                Map<TopicPartition, OffsetAndMetadata> currentCommittedOffsets = getCommittedOffsets(groupId);
                Collection<TopicPartition> partitionsToResetWithCommittedOffset = new ArrayList<>();
                Collection<TopicPartition> partitionsToResetWithoutCommittedOffset = new ArrayList<>();

                for (TopicPartition topicPartition : partitionsToReset) {
                    if (currentCommittedOffsets.containsKey(topicPartition))
                        partitionsToResetWithCommittedOffset.add(topicPartition);
                    else
                        partitionsToResetWithoutCommittedOffset.add(topicPartition);
                }

                Map<TopicPartition, OffsetAndMetadata> preparedOffsetsForPartitionsWithCommittedOffset = partitionsToResetWithCommittedOffset.stream()
                    .collect(Collectors.toMap(Function.identity(), topicPartition -> {
                        OffsetAndMetadata committedOffset = currentCommittedOffsets.get(topicPartition);

                        if (committedOffset == null) {
                            throw new IllegalStateException("Expected a valid current offset for topic partition: " + topicPartition);
                        }

                        return new OffsetAndMetadata(committedOffset.offset());
                    }));

                Map<TopicPartition, OffsetAndMetadata> preparedOffsetsForPartitionsWithoutCommittedOffset = getLogEndOffsets(partitionsToResetWithoutCommittedOffset)
                    .entrySet().stream().collect(Collectors.toMap(Entry::getKey, e -> {
                        if (!(e.getValue() instanceof LogOffset)) {
                            ToolsUtils.printUsageAndExit(opts.parser, "Error getting ending offset of topic partition: " + e.getKey());
                            return null;
                        }

                        return new OffsetAndMetadata(((LogOffset) e.getValue()).value);
                    }));

                preparedOffsetsForPartitionsWithCommittedOffset.putAll(preparedOffsetsForPartitionsWithoutCommittedOffset);

                return preparedOffsetsForPartitionsWithCommittedOffset;
            }

            ToolsUtils.printUsageAndExit(opts.parser, String.format("Option '%s' requires one of the following scenarios: %s", opts.resetOffsetsOpt, opts.allResetOffsetScenarioOpts));
            return null;
        }
Returns states of the specified consumer groups and partition assignment states
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java