public Set<String> createTopicsWithRetry(NewTopic topicDescription, long timeoutMs, long backOffMs, Time time) {
        Timer timer = time.timer(timeoutMs);
        do {
            try {
                return createTopics(topicDescription);
            } catch (ConnectException e) {
                if (timer.notExpired() && retryableTopicCreationException(e)) {
                    log.info("'{}' topic creation failed due to '{}', retrying, {}ms remaining",
                            topicDescription.name(), e.getMessage(), timer.remainingMs());
                } else {
                    throw e;
                }
            }
            timer.sleep(backOffMs);
        } while (timer.notExpired());
        throw new TimeoutException("Timeout expired while trying to create topic(s)");
    }
Implements a retry logic around creating topic(s) in case it'd fail due to
specific type of exceptions, see {@link TopicAdmin#retryableTopicCreationException(ConnectException)}

@param topicDescription the specifications of the topic
@param timeoutMs        Timeout in milliseconds
@param backOffMs        Time for delay after initial failed attempt in milliseconds
@param time             {@link Time} instance
@return the names of the topics that were created by this operation; never null but possibly empty,
the same as {@link TopicAdmin#createTopics(NewTopic...)}
connect/runtime/src/main/java/org/apache/kafka/connect/util/TopicAdmin.java