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

    // When:
    final String statement =
        ProcessingLogServerUtils.processingLogStreamCreateStatement(
            config,
            ksqlConfig);

    // Then:
    assertThat(statement, equalTo(
        "CREATE STREAM PROCESSING_LOG_STREAM ("
            + "logger VARCHAR, "
            + "level VARCHAR, "
            + "time BIGINT, "
            + "message STRUCT<"
            + "type INT, "
            + "deserializationError STRUCT<target VARCHAR, errorMessage VARCHAR, recordB64 VARCHAR, cause ARRAY<VARCHAR>, `topic` VARCHAR>, "
            + "recordProcessingError STRUCT<errorMessage VARCHAR, record VARCHAR, cause ARRAY<VARCHAR>>, "
            + "productionError STRUCT<errorMessage VARCHAR>, "
            + "serializationError STRUCT<target VARCHAR, errorMessage VARCHAR, record VARCHAR, cause ARRAY<VARCHAR>, `topic` VARCHAR>, "
            + "kafkaStreamsThreadError STRUCT<errorMessage VARCHAR, threadName VARCHAR, cause ARRAY<VARCHAR>>"
            + ">"
            + ") WITH(KAFKA_TOPIC='processing_log_topic', VALUE_FORMAT='JSON', KEY_FORMAT='KAFKA');"));
  }