public Map<String, Optional<TopicDescription>> describeTopics(Set<String> topicNames) {
        Map<String, Optional<TopicDescription>> results = new HashMap<>();
        log.info("Describing topics {}", topicNames);
        try (Admin admin = createAdminClient()) {
            DescribeTopicsResult result = admin.describeTopics(topicNames);
            Map<String, KafkaFuture<TopicDescription>> byName = result.topicNameValues();
            for (Map.Entry<String, KafkaFuture<TopicDescription>> entry : byName.entrySet()) {
                String topicName = entry.getKey();
                try {
                    TopicDescription desc = entry.getValue().get();
                    results.put(topicName, Optional.of(desc));
                    log.info("Found topic {} : {}", topicName, desc);
                } catch (ExecutionException e) {
                    Throwable cause = e.getCause();
                    if (cause instanceof UnknownTopicOrPartitionException) {
                        results.put(topicName, Optional.empty());
                        log.info("Found non-existent topic {}", topicName);
                        continue;
                    }
                    throw new AssertionError("Could not describe topic(s)" + topicNames, e);
                }
            }
        } catch (Exception e) {
            throw new AssertionError("Could not describe topic(s) " + topicNames, e);
        }
        log.info("Found topics {}", results);
        return results;
    }
Get the topic descriptions of the named topics. The value of the map entry will be empty
if the topic does not exist.

@param topicNames the names of the topics to describe
@return the map of optional {@link TopicDescription} keyed by the topic name
connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java