public static Optional<String> maybeCreateProcessingLogTopic(
      final KafkaTopicClient topicClient,
      final ProcessingLogConfig config,
      final KsqlConfig ksqlConfig) {
    if (!config.getBoolean(ProcessingLogConfig.TOPIC_AUTO_CREATE)) {
      return Optional.empty();
    }
    final String topicName = getTopicName(config, ksqlConfig);
    final int nPartitions =
        config.getInt(ProcessingLogConfig.TOPIC_PARTITIONS);
    final short nReplicas =
        config.getShort(ProcessingLogConfig.TOPIC_REPLICATION_FACTOR);
    try {
      topicClient.createTopic(topicName, nPartitions, nReplicas);
    } catch (final KafkaTopicExistsException e) {
      if (e.getPartitionOrReplicaMismatch()) {
        LOGGER.warn(String.format("Log topic %s already exists", topicName), e);
      } else {
        LOGGER.info(String.format("Log topic %s already exists", topicName), e);
      }
    }
    return Optional.of(topicName);
  }
@Test
  public void shouldNotCreateLogTopicIfNotConfigured() {
    // Given:
    final ProcessingLogConfig config = new ProcessingLogConfig(
        ImmutableMap.of(ProcessingLogConfig.TOPIC_AUTO_CREATE, false)
    );

    // When:
    final Optional<String> createdTopic = ProcessingLogServerUtils.maybeCreateProcessingLogTopic(
        spyTopicClient,
        config,
        ksqlConfig);

    // Then:
    assertThat(createdTopic.isPresent(), is(false));
    verifyNoMoreInteractions(spyTopicClient);
  }