private Set<TopicPartition> listPartitions(
            long maxDurationMs,
            Admin admin,
            Collection<String> topics
    ) throws TimeoutException, InterruptedException, ExecutionException {
        if (topics.isEmpty()) {
            throw new AssertionError("collection of topics may not be empty");
        }
        return admin.describeTopics(topics)
                .allTopicNames().get(maxDurationMs, TimeUnit.MILLISECONDS)
                .entrySet().stream()
                .flatMap(e -> e.getValue().partitions().stream().map(p -> new TopicPartition(e.getKey(), p.partition())))
                .collect(Collectors.toSet());
    }
List all the known partitions for the given {@link Collection} of topics
@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 topics the topics whose partitions should be listed
@return a {@link Set} of {@link TopicPartition topic partitions} for the given topics; never null, and never empty
connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java