public void startAsync() {
    try {
      udfLoader.load();
      ProcessingLogServerUtils.maybeCreateProcessingLogTopic(
          serviceContext.getTopicClient(),
          processingLogConfig,
          ksqlConfig);
      if (processingLogConfig.getBoolean(ProcessingLogConfig.STREAM_AUTO_CREATE)) {
        log.warn("processing log auto-create is enabled, but this is not supported "
            + "for headless mode.");
      }
      rocksDBConfigSetterHandler.accept(ksqlConfig);
      processesQueryFile(readQueriesFile(queriesFile));
      showWelcomeMessage();
      final Properties properties = new Properties();
      ksqlConfig.originals().forEach((key, value) -> {
        if (nonNull(value)) {
          properties.put(key, value.toString());
        }
      });
      versionChecker.start(KsqlModuleType.SERVER, properties);
    } catch (final Exception e) {
      log.error("Failed to start KSQL Server with query file: " + queriesFile, e);
      throw e;
    }
  }
@Test
  public void shouldSupportTopicInference() {
    // Given:
    givenQueryFileParsesTo(PREPARED_CSAS);

    when(sandBoxTopicInjector.inject(argThat(configured(equalTo(PREPARED_CSAS)))))
        .thenReturn((ConfiguredStatement) CSAS_CFG_WITH_TOPIC);

    // When:
    standaloneExecutor.startAsync();

    // Then:
    verify(sandBox).execute(sandBoxServiceContext, CSAS_CFG_WITH_TOPIC);
  }