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); // *sigh*
      return (ApplicationRunner) constructor.newInstance(userApp, config);
    } catch (ConfigException ce) {
      // this is thrown due to invalid app.runner.class configuration
      throw ce;
    } catch (Exception e) {
      // other types of exception during class loading and construction of new instance
      throw new ConfigException(String.format("Could not load ApplicationRunner class %s", appRunnerClassName), e);
    }
  }
Get the {@link ApplicationRunner} that runs the {@code userApp}

@param userApp the user application object
@param config the configuration for this application
@return the {@link ApplicationRunner} object that will run the {@code userApp}
samza-core/src/main/java/org/apache/samza/runtime/ApplicationRunners.java