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

    // Then:
    final InOrder inOrder = Mockito.inOrder(ksqlResource, serverState);
    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));
    inOrder.verify(serverState).setReady();
  }