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 shouldFailOnDropStatement() {
    // Given:
    givenQueryFileParsesTo(
        PreparedStatement.of("DROP Test",
            new DropStream(SOME_NAME, false, false))
    );

    // When:
    final Exception e = assertThrows(
        KsqlStatementException.class,
        () -> standaloneExecutor.startAsync()
    );

    // Then:
    assertThat(e.getMessage(), containsString("Unsupported statement. "
        + "Only the following statements are supporting in standalone mode:\n"
        + "CREAETE STREAM AS SELECT\n"
        + "CREATE STREAM\n"
        + "CREATE TABLE\n"
        + "CREATE TABLE AS SELECT\n"
        + "INSERT INTO\n"
        + "REGISTER TYPE\n"
        + "SET\n"
        + "UNSET"));
  }