@VisibleForTesting
  void startKsql(final KsqlConfig ksqlConfigWithPort) {
    cleanupOldState();
    initialize(ksqlConfigWithPort);
  }
@Test
  public void shouldStartCommandStoreAndCommandRunnerBeforeCreatingLogStream() {
    // When:
    app.startKsql(ksqlConfig);

    // Then:
    final InOrder inOrder = Mockito.inOrder(commandQueue, commandRunner, ksqlResource);
    inOrder.verify(commandQueue).start();
    inOrder.verify(commandRunner).processPriorCommands(queryCleanupArgumentCaptor.capture());
    inOrder.verify(commandRunner).start();
    inOrder.verify(ksqlResource).handleKsqlStatements(
        securityContextArgumentCaptor.capture(),
        eq(new KsqlRequest(logCreateStatement, Collections.emptyMap(), Collections.emptyMap(), null))
    );
    assertThat(securityContextArgumentCaptor.getValue().getUserPrincipal(), is(Optional.empty()));
    assertThat(securityContextArgumentCaptor.getValue().getServiceContext(), is(serviceContext));
  }