public static void main(String[] args) throws Exception {
    Thread.setDefaultUncaughtExceptionHandler(
        new SamzaUncaughtExceptionHandler(() -> {
          log.info("Exiting process now.");
          System.exit(1);
        }));

    String containerId = System.getenv(ShellCommandConfig.ENV_CONTAINER_ID);
    System.out.println(String.format("Container ID: %s", containerId));

    String coordinatorUrl = System.getenv(ShellCommandConfig.ENV_COORDINATOR_URL);
    System.out.println(String.format("Coordinator URL: %s", coordinatorUrl));

    Optional<String> executionEnvContainerId =
        Optional.ofNullable(System.getenv(ShellCommandConfig.ENV_EXECUTION_ENV_CONTAINER_ID));
    Optional<String> samzaEpochId = Optional.ofNullable(System.getenv(EnvironmentVariables.SAMZA_EPOCH_ID));

    int delay = new Random().nextInt(SamzaContainer.DEFAULT_READ_JOBMODEL_DELAY_MS()) + 1;
    JobModel jobModel = SamzaContainer.readJobModel(coordinatorUrl, delay);
    Config config = jobModel.getConfig();

    // this call is also in ContainerLaunchUtil, but adding this here allows more logs to get handled by Samza loggers
    LoggingContextHolder.INSTANCE.setConfig(config);
    log.info(String.format("Got container ID: %s", containerId));
    log.info(String.format("Got coordinator URL: %s", coordinatorUrl));

    JobConfig jobConfig = new JobConfig(config);
    String jobName = jobConfig.getName()
        .orElseThrow(() -> new SamzaException(String.format("Config %s is missing", JobConfig.JOB_NAME)));
    String jobId = jobConfig.getJobId();

    ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc =
        ApplicationDescriptorUtil.getAppDescriptor(ApplicationUtil.fromConfig(config), config);

    ContainerLaunchUtil.run(appDesc, jobName, jobId, containerId, executionEnvContainerId, samzaEpochId,
        jobModel);
  }
Launches and manages the lifecycle for {@link SamzaContainer}s in YARN.
samza-core/src/main/java/org/apache/samza/runtime/LocalContainerRunner.java