class KsqlResource {

  @POST
  public Response handleKsqlStatements(KsqlRequest request) {
    List<Statement> parsedStatements;
    List<String> statementStrings;
    Map<String, Object> streamsProperties;

    try {
      parsedStatements = ksqlEngine.getStatements(request.getKsql());
      statementStrings = getStatementStrings(request.getKsql());
    } catch (KsqlException e) {
      return Errors.badRequest(e);
    }

    streamsProperties = request.getStreamsProperties();
    if (parsedStatements.size() != statementStrings.size()) {
      return Errors.badRequest(String.format(
          "Size of parsed statements and statement strings differ; %d vs. %d, respectively",
          parsedStatements.size(),
          statementStrings.size()
      ));
    }

    KsqlEntityList result = new KsqlEntityList();
    for (int i = 0; i < parsedStatements.size(); i++) {
      String statementText = statementStrings.get(i);
      try {
        validateStatement(
            result, statementStrings.get(i), parsedStatements.get(i), streamsProperties);
      } catch (KsqlRestException e) {
        throw e;
      } catch (KsqlException e) {
        return Errors.badStatement(e, statementText, result);
      } catch (Exception e) {
        return Errors.serverErrorForStatement(e, statementText, result);
      }
      try {
        result.add(executeStatement(statementText, parsedStatements.get(i), streamsProperties));
      } catch (Exception e) {
        return Errors.serverErrorForStatement(e, statementText, result);
      }
    }
    return Response.ok(result).build();
  }

  public  KsqlResource(
      KsqlEngine ksqlEngine,
      CommandStore commandStore,
      StatementExecutor statementExecutor,
      long distributedCommandResponseTimeout
  );

}