private Map<TopicPartition, LogOffsetResult> getLogTimestampOffsets(Collection<TopicPartition> topicPartitions, long timestamp) {
            try {
                Map<TopicPartition, OffsetSpec> timestampOffsets = topicPartitions.stream()
                    .collect(Collectors.toMap(Function.identity(), tp -> OffsetSpec.forTimestamp(timestamp)));

                Map<TopicPartition, ListOffsetsResultInfo> offsets = adminClient.listOffsets(
                    timestampOffsets,
                    withTimeoutMs(new ListOffsetsOptions())
                ).all().get();

                Map<TopicPartition, ListOffsetsResultInfo> successfulOffsetsForTimes = new HashMap<>();
                Map<TopicPartition, ListOffsetsResultInfo> unsuccessfulOffsetsForTimes = new HashMap<>();

                offsets.forEach((tp, offsetsResultInfo) -> {
                    if (offsetsResultInfo.offset() != ListOffsetsResponse.UNKNOWN_OFFSET)
                        successfulOffsetsForTimes.put(tp, offsetsResultInfo);
                    else
                        unsuccessfulOffsetsForTimes.put(tp, offsetsResultInfo);
                });

                Map<TopicPartition, LogOffsetResult> successfulLogTimestampOffsets = successfulOffsetsForTimes.entrySet().stream()
                    .collect(Collectors.toMap(Entry::getKey, e -> new LogOffset(e.getValue().offset())));

                unsuccessfulOffsetsForTimes.forEach((tp, offsetResultInfo) ->
                    System.out.println("\nWarn: Partition " + tp.partition() + " from topic " + tp.topic() +
                    " is empty. Falling back to latest known offset."));

                successfulLogTimestampOffsets.putAll(getLogEndOffsets(unsuccessfulOffsetsForTimes.keySet()));

                return successfulLogTimestampOffsets;
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        }
Returns states of the specified consumer groups and partition assignment states
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java