public static Config computeSamzaConfigs(Boolean localRunner, Config config) {
    Map<String, String> newConfig = new HashMap<>();

    // TODO: Introduce an API to return a dsl string containing one or more sql statements
    List<String> dslStmts = SamzaSqlDslConverter.fetchSqlFromConfig(config);

    // This is needed because the SQL file may not be available in all the node managers.
    String sqlJson = SamzaSqlApplicationConfig.serializeSqlStmts(dslStmts);
    newConfig.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, sqlJson);

    List<String> inputSystemStreams = new LinkedList<>();
    List<String> outputSystemStreams = new LinkedList<>();

    SamzaSqlApplicationConfig.populateSystemStreamsAndGetRelRoots(dslStmts, config, inputSystemStreams,
        outputSystemStreams);

    SqlIOResolver ioResolver = SamzaSqlApplicationConfig.createIOResolver(config);

    // Populate stream to system mapping config for input and output system streams
    for (String source : inputSystemStreams) {
      SqlIOConfig inputSystemStreamConfig = ioResolver.fetchSourceInfo(source);
      newConfig.put(String.format(CFG_FMT_SAMZA_STREAM_SYSTEM, inputSystemStreamConfig.getStreamId()),
          inputSystemStreamConfig.getSystemName());
      newConfig.putAll(inputSystemStreamConfig.getConfig());
    }

    for (String sink : outputSystemStreams) {
      SqlIOConfig outputSystemStreamConfig = ioResolver.fetchSinkInfo(sink);
      newConfig.put(String.format(CFG_FMT_SAMZA_STREAM_SYSTEM, outputSystemStreamConfig.getStreamId()),
          outputSystemStreamConfig.getSystemName());
      newConfig.putAll(outputSystemStreamConfig.getConfig());
    }

    newConfig.putAll(config);

    if (localRunner) {
      newConfig.put(ApplicationConfig.APP_RUNNER_CLASS, LocalApplicationRunner.class.getName());
    } else {
      newConfig.put(ApplicationConfig.APP_RUNNER_CLASS, RemoteApplicationRunner.class.getName());
    }

    LOG.info("New Samza configs: " + newConfig);
    return new MapConfig(newConfig);
  }
NOTE: This constructor is called from {@link ApplicationRunners} through reflection.
Please refrain from updating the signature or removing this constructor unless the caller has changed the interface.
samza-sql/src/main/java/org/apache/samza/sql/runner/SamzaSqlApplicationRunner.java