class KafkaTopicClientImpl implements KafkaTopicClient {

  @Override
  public void createTopic(
      final String topic,
      final int numPartitions,
      final short replicationFactor,
      final Map<String, ?> configs,
      final CreateTopicsOptions createOptions
  ) {
    if (isTopicExists(topic)) {
      validateTopicProperties(topic, numPartitions, replicationFactor);
      return;
    }

    final short resolvedReplicationFactor = replicationFactor == TopicProperties.DEFAULT_REPLICAS
        ? getDefaultClusterReplication()
        : replicationFactor;

    final NewTopic newTopic = new NewTopic(topic, numPartitions, resolvedReplicationFactor);
    newTopic.configs(toStringConfigs(configs));

    try {
      LOG.info("Creating topic '{}' {}",
          topic,
          (createOptions.shouldValidateOnly()) ? "(ONLY VALIDATE)" : ""
      );

      ExecutorUtil.executeWithRetries(
          () -> adminClient.get().createTopics(
              Collections.singleton(newTopic),
              createOptions
          ).all().get(),
          ExecutorUtil.RetryBehaviour.ON_RETRYABLE);
    } catch (final InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new KafkaResponseGetFailedException(
          "Failed to guarantee existence of topic " + topic, e);

    } catch (final TopicExistsException e) {
      // if the topic already exists, it is most likely because another node just created it.
      // ensure that it matches the partition count and replication factor before returning
      // success
      validateTopicProperties(topic, numPartitions, replicationFactor);

    } catch (final TopicAuthorizationException e) {
      throw new KsqlTopicAuthorizationException(
          AclOperation.CREATE, Collections.singleton(topic));

    } catch (final Exception e) {
      throw new KafkaResponseGetFailedException(
          "Failed to guarantee existence of topic " + topic, e);

    }
  }

  public  KafkaTopicClientImpl(final Supplier<Admin> sharedAdminClient);

}