@Override
  public QueryResult executeQuery(ExecutionContext context, String statement) throws ExecutorException {
    outputData.clear();

    int execId = execIdSeq.incrementAndGet();
    Map<String, String> staticConfigs = fetchSamzaSqlConfig(execId);
    List<String> sqlStmts = formatSqlStmts(Collections.singletonList(statement));
    staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts));

    SamzaSqlApplicationRunner runner;
    try {
      runner = new SamzaSqlApplicationRunner(true, new MapConfig(staticConfigs));
      runner.run(null);
    } catch (SamzaException ex) {
      throw new ExecutorException(ex);
    }
    executions.put(execId, runner);
    LOG.debug("Executing sql. Id ", execId);

    SqlSchema resultSchema = null;
    // TODO: after fixing the TODO in generateResultSchema function, we can uncomment the following piece of code.
    // resultSchema = generateResultSchema(new MapConfig(staticConfigs));
    return new QueryResult(execId, resultSchema);
  }
Samza implementation of Executor for Samza SQL Shell.
samza-sql-shell/src/main/java/org/apache/samza/sql/client/impl/SamzaExecutor.java