public static String processingLogTopic(
      final ProcessingLogConfig config,
      final KsqlConfig ksqlConfig
  ) {
    final String topicNameConfig = config.getString(ProcessingLogConfig.TOPIC_NAME);
    if (topicNameConfig.equals(ProcessingLogConfig.TOPIC_NAME_NOT_SET)) {
      return String.format(
          "%s%s",
          ksqlConfig.getString(KsqlConfig.KSQL_SERVICE_ID_CONFIG),
          ProcessingLogConfig.TOPIC_NAME_DEFAULT_SUFFIX
      );
    } else {
      return topicNameConfig;
    }
  }
@Test
  public void shouldReturnProcessingLogTopic() {
    // Given/When
    final ProcessingLogConfig processingLogConfig = new ProcessingLogConfig(ImmutableMap.of());
    final String processingLogTopic = ReservedInternalTopics.processingLogTopic(
        processingLogConfig, ksqlConfig);

    // Then
    assertThat(processingLogTopic, is("default_ksql_processing_log"));
  }