ApplicationRunners { public static ApplicationRunner getApplicationRunner(SamzaApplication<?> userApp, Config config) { String appRunnerClassName = new ApplicationConfig(config).getAppRunnerClass(); try { Class<?> runnerClass = Class.forName(appRunnerClassName); if (!ApplicationRunner.class.isAssignableFrom(runnerClass)) { throw new ConfigException( String.format("Class %s does not extend ApplicationRunner properly", appRunnerClassName)); } Constructor<?> constructor = runnerClass.getConstructor(SamzaApplication.class, Config.class); return (ApplicationRunner) constructor.newInstance(userApp, config); } catch (ConfigException ce) { throw ce; } catch (Exception e) { throw new ConfigException(String.format("Could not load ApplicationRunner class %s", appRunnerClassName), e); } } private  ApplicationRunners(); static ApplicationRunner getApplicationRunner(SamzaApplication<?> userApp, Config config);  }
@Test public void testGetAppRunner() { Map<String, String> configMap = new HashMap<>(); configMap.put(ApplicationConfig.APP_RUNNER_CLASS, MockApplicationRunner.class.getName()); Config config = new MapConfig(configMap); StreamApplication app = mock(StreamApplication.class); ApplicationRunner appRunner = ApplicationRunners.getApplicationRunner(app, config); assertTrue(appRunner instanceof MockApplicationRunner); }