@VisibleForTesting
  void startKsql(final KsqlConfig ksqlConfigWithPort) {
    cleanupOldState();
    initialize(ksqlConfigWithPort);
  }
@Test
  public void shouldNotCreateLogStreamIfAutoCreateNotConfigured() {
    // Given:
    when(processingLogConfig.getBoolean(ProcessingLogConfig.STREAM_AUTO_CREATE))
        .thenReturn(false);

    // When:
    app.startKsql(ksqlConfig);

    // Then:
    verify(ksqlResource, never()).handleKsqlStatements(
        securityContext,
        new KsqlRequest(logCreateStatement, Collections.emptyMap(), Collections.emptyMap(), null)
    );
  }