@Override
  public NonQueryResult executeNonQuery(ExecutionContext context, List<String> statement) throws ExecutorException {
    int execId = execIdSeq.incrementAndGet();
    Map<String, String> staticConfigs = fetchSamzaSqlConfig(execId);
    staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(formatSqlStmts(statement)));

    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);

    return new NonQueryResult(execId);
  }
Samza implementation of Executor for Samza SQL Shell.
samza-sql-shell/src/main/java/org/apache/samza/sql/client/impl/SamzaExecutor.java