public static void ensureTopic(
      final String name,
      final KsqlConfig ksqlConfig,
      final KafkaTopicClient topicClient
  ) {
    if (topicClient.isTopicExists(name)) {
      validateTopicConfig(name, ksqlConfig, topicClient);
      return;
    }

    final short replicationFactor = ksqlConfig
        .getShort(KsqlConfig.KSQL_INTERNAL_TOPIC_REPLICAS_PROPERTY);

    if (replicationFactor < 2) {
      log.warn("Creating topic {} with replication factor of {} which is less than 2. "
              + "This is not advisable in a production environment. ",
          name, replicationFactor);
    }

    final short minInSyncReplicas = ksqlConfig
        .getShort(KsqlConfig.KSQL_INTERNAL_TOPIC_MIN_INSYNC_REPLICAS_PROPERTY);

    topicClient.createTopic(
        name,
        INTERNAL_TOPIC_PARTITION_COUNT,
        replicationFactor,
        ImmutableMap.<String, Object>builder()
            .putAll(INTERNAL_TOPIC_CONFIG)
            .put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, minInSyncReplicas)
            .build()
    );
  }
@Test
  public void shouldFailIfTopicExistsWithInvalidNPartitions() {
    // Given:
    whenTopicExistsWith(2, NREPLICAS);

    // When:
    assertThrows(
        IllegalStateException.class,
        () -> KsqlInternalTopicUtils.ensureTopic(TOPIC_NAME, ksqlConfig, topicClient)
    );
  }