SamzaSqlApplicationRunner implements ApplicationRunner { public static Config computeSamzaConfigs(Boolean localRunner, Config config) { Map<String, String> newConfig = new HashMap<>(); List<String> dslStmts = SamzaSqlDslConverter.fetchSqlFromConfig(config); 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); 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); } @SuppressWarnings("unused") /* used via reflection */  SamzaSqlApplicationRunner(SamzaApplication app, Config config);  SamzaSqlApplicationRunner(Boolean isLocalRunner, Config config); private  SamzaSqlApplicationRunner(SamzaApplication app, Boolean localRunner, Config config); static Config computeSamzaConfigs(Boolean localRunner, Config config); void runAndWaitForFinish(); void runAndWaitForFinish(ExternalContext externalContext); @Override void run(ExternalContext externalContext); @Override void kill(); @Override ApplicationStatus status(); @Override void waitForFinish(); @Override boolean waitForFinish(Duration timeout); static final String CFG_FMT_SAMZA_STREAM_SYSTEM; }
@Test public void testComputeSamzaConfigs() { Map<String, String> configs = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(10); String sql1 = "Insert into testavro.outputTopic(id,long_value) select id, MyTest(id) as long_value from testavro.SIMPLE1"; configs.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, sql1); configs.put(ApplicationConfig.APP_RUNNER_CLASS, SamzaSqlApplicationRunner.class.getName()); MapConfig samzaConfig = new MapConfig(configs); Config newConfigs = SamzaSqlApplicationRunner.computeSamzaConfigs(true, samzaConfig); Assert.assertEquals(newConfigs.get(ApplicationConfig.APP_RUNNER_CLASS), LocalApplicationRunner.class.getName()); Assert.assertEquals(newConfigs.size(), configs.size() + 5); newConfigs = SamzaSqlApplicationRunner.computeSamzaConfigs(false, samzaConfig); Assert.assertEquals(newConfigs.get(ApplicationConfig.APP_RUNNER_CLASS), RemoteApplicationRunner.class.getName()); Assert.assertEquals(newConfigs.size(), configs.size() + 5); }