private boolean retryableTopicCreationException(ConnectException e) {
        // createTopics wraps the exception into ConnectException
        // to retry the creation, it should be an ExecutionException from future get which was caused by InvalidReplicationFactorException
        // or can be a TimeoutException
        Throwable cause = e.getCause();
        while (cause != null) {
            final Throwable finalCause = cause;
            if (CAUSES_TO_RETRY_TOPIC_CREATION.stream().anyMatch(exceptionClass -> exceptionClass.isInstance(finalCause))) {
                return true;
            }
            cause = cause.getCause();
        }
        return false;
    }
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