public ProcessLauncher createLauncher()

in twill-yarn/src/main/hadoop20/org/apache/twill/internal/yarn/Hadoop20YarnAppClient.java [75:119]
34 lines of code
3 McCabe index (conditional complexity)

  public ProcessLauncher<ApplicationMasterInfo> createLauncher(TwillSpecification twillSpec,
                                                               @Nullable String schedulerQueue) throws Exception {
    // Request for new application
    final GetNewApplicationResponse response = yarnClient.getNewApplication();
    final ApplicationId appId = response.getApplicationId();

    // Setup the context for application submission
    final ApplicationSubmissionContext appSubmissionContext = Records.newRecord(ApplicationSubmissionContext.class);
    appSubmissionContext.setApplicationId(appId);
    appSubmissionContext.setApplicationName(twillSpec.getName());
    appSubmissionContext.setUser(user);

    if (schedulerQueue != null) {
      appSubmissionContext.setQueue(schedulerQueue);
    }

    // TODO: Make it adjustable through TwillSpec (TWILL-90)
    // Set the resource requirement for AM
    Resource amResource = Records.newRecord(Resource.class);
    amResource.setMemory(Constants.APP_MASTER_MEMORY_MB);
    final Resource capability = adjustMemory(response, amResource);
    ApplicationMasterInfo appMasterInfo = new ApplicationMasterInfo(appId, capability.getMemory(), 1);

    ApplicationSubmitter submitter = new ApplicationSubmitter() {

      @Override
      public ProcessController<YarnApplicationReport> submit(YarnLaunchContext launchContext) {
        ContainerLaunchContext context = launchContext.getLaunchContext();
        addRMToken(context);
        context.setUser(appSubmissionContext.getUser());
        context.setResource(adjustMemory(response, capability));
        appSubmissionContext.setAMContainerSpec(context);

        try {
          yarnClient.submitApplication(appSubmissionContext);
          return new ProcessControllerImpl(yarnClient, appId);
        } catch (YarnRemoteException e) {
          LOG.error("Failed to submit application {}", appId, e);
          throw Throwables.propagate(e);
        }
      }
    };

    return new ApplicationMasterProcessLauncher(appMasterInfo, submitter);
  }