public static String processingLogStreamCreateStatement(
      final ProcessingLogConfig config,
      final KsqlConfig ksqlConfig
  ) {
    return processingLogStreamCreateStatement(
        config.getString(ProcessingLogConfig.STREAM_NAME),
        getTopicName(config, ksqlConfig)
    );
  }
@Test
  public void shouldBuildCorrectStreamCreateDDLWithDefaultTopicName() {
    // Given:
    serviceContext.getTopicClient().createTopic(DEFAULT_TOPIC, 1, (short) 1);

    // When:
    final String statement =
        ProcessingLogServerUtils.processingLogStreamCreateStatement(
            new ProcessingLogConfig(
                ImmutableMap.of(
                    ProcessingLogConfig.STREAM_AUTO_CREATE, true,
                    ProcessingLogConfig.STREAM_NAME, STREAM
                )
            ),
            ksqlConfig);

    // Then:
    assertThat(statement,
        containsString("KAFKA_TOPIC='ksql_cluster.ksql_processing_log'"));
  }