@VisibleForTesting
  void startKsql(final KsqlConfig ksqlConfigWithPort) {
    cleanupOldState();
    initialize(ksqlConfigWithPort);
  }
@Test
  public void shouldNotCreateLogStreamIfAlreadyExists() {
    // Given:

    final StreamsList streamsList =
        new StreamsList(
            LIST_STREAMS_SQL,
            Collections.singletonList(new SourceInfo.Stream(LOG_STREAM_NAME, "", "", "", false)));
    when(response.getEntity()).thenReturn(new KsqlEntityList(Collections.singletonList(streamsList)));
    
    when(ksqlResource.handleKsqlStatements(
        any(),
        any())
    ).thenReturn(response);

    // When:
    app.startKsql(ksqlConfig);

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