private Map<TopicPartition, Long> readEndOffsets(
            long maxDurationMs,
            Admin admin,
            Collection<TopicPartition> topicPartitions
    ) throws TimeoutException, InterruptedException, ExecutionException {
        if (topicPartitions.isEmpty()) {
            throw new AssertionError("collection of topic partitions may not be empty");
        }
        Map<TopicPartition, OffsetSpec> offsetSpecMap = topicPartitions.stream().collect(Collectors.toMap(Function.identity(), tp -> OffsetSpec.latest()));
        return admin.listOffsets(offsetSpecMap, new ListOffsetsOptions(IsolationLevel.READ_UNCOMMITTED))
                .all().get(maxDurationMs, TimeUnit.MILLISECONDS)
                .entrySet().stream()
                .collect(Collectors.toMap(
                        Map.Entry::getKey,
                        e -> e.getValue().offset()
                ));
    }
List the latest current offsets for the given {@link Collection} of {@link TopicPartition topic partitions}
@param maxDurationMs the max duration to wait for while fetching metadata from Kafka (in milliseconds)
@param admin the admin client to use for fetching metadata from the Kafka cluster
@param topicPartitions the topic partitions to list end offsets for
@return a {@link Map} containing the latest offset for each requested {@link TopicPartition topic partition}; never null, and never empty
connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java