@Override
  public boolean createTopic(
      final String topic,
      final int numPartitions,
      final short replicationFactor,
      final Map<String, ?> configs,
      final CreateTopicsOptions createOptions
  ) {
    final Optional<Long> retentionMs = KafkaTopicClient.getRetentionMs(configs);

    if (isTopicExists(topic)) {
      validateTopicProperties(topic, numPartitions, replicationFactor, retentionMs);
      return false;
    }

    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);
      return true;

    } 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, replication factor, and retention
      // before returning success
      validateTopicProperties(topic, numPartitions, replicationFactor, retentionMs);
      return false;

    } 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);

    }
  }
@Test
  public void shouldNotCreateTopicIfItAlreadyExistsWithMatchingDetails() {
    // Given:
    givenTopicExists("someTopic", 3, 2);
    givenTopicConfigs(
        "someTopic",
        overriddenConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "8640000000")
    );

    // When:
    kafkaTopicClient.createTopic("someTopic", 3, (short) 2, configs);

    // Then:
    verify(adminClient, never()).createTopics(any(), any());
  }