@VisibleForTesting
  ApplicationId setupAndSubmitApplication() throws IOException, YarnException {
    YarnClientApplication gobblinYarnApp = this.yarnClient.createApplication();
    ApplicationSubmissionContext appSubmissionContext = gobblinYarnApp.getApplicationSubmissionContext();
    appSubmissionContext.setApplicationType(GOBBLIN_YARN_APPLICATION_TYPE);
    ApplicationId applicationId = appSubmissionContext.getApplicationId();

    GetNewApplicationResponse newApplicationResponse = gobblinYarnApp.getNewApplicationResponse();
    // Set up resource type requirements for ApplicationMaster
    Resource resource = prepareContainerResource(newApplicationResponse);

    // Add lib jars, and jars and files that the ApplicationMaster need as LocalResources
    Map<String, LocalResource> appMasterLocalResources = addAppMasterLocalResources(applicationId);

    ContainerLaunchContext amContainerLaunchContext = Records.newRecord(ContainerLaunchContext.class);
    amContainerLaunchContext.setLocalResources(appMasterLocalResources);
    amContainerLaunchContext.setEnvironment(YarnHelixUtils.getEnvironmentVariables(this.yarnConfiguration));
    amContainerLaunchContext.setCommands(Lists.newArrayList(buildApplicationMasterCommand(resource.getMemory())));
    if (UserGroupInformation.isSecurityEnabled()) {
      setupSecurityTokens(amContainerLaunchContext);
    }

    // Setup the application submission context
    appSubmissionContext.setApplicationName(this.applicationName);
    appSubmissionContext.setResource(resource);
    appSubmissionContext.setQueue(this.appQueueName);
    appSubmissionContext.setPriority(Priority.newInstance(0));
    appSubmissionContext.setAMContainerSpec(amContainerLaunchContext);

    // Also setup container local resources by copying local jars and files the container need to HDFS
    addContainerLocalResources(applicationId);

    // Submit the application
    LOGGER.info("Submitting application " + applicationId);
    this.yarnClient.submitApplication(appSubmissionContext);

    LOGGER.info("Application successfully submitted and accepted");
    ApplicationReport applicationReport = this.yarnClient.getApplicationReport(applicationId);
    LOGGER.info("Application Name: " + applicationReport.getName());
    LOGGER.info("Application Tracking URL: " + applicationReport.getTrackingUrl());
    LOGGER.info("Application User: " + applicationReport.getUser() + " Queue: " + applicationReport.getQueue());

    return applicationId;
  }
@ VisibleForTesting ApplicationId setupAndSubmitApplication ( ) throws IOException , YarnException { YarnClientApplication gobblinYarnApp = this . yarnClient . createApplication ( ) ; ApplicationSubmissionContext appSubmissionContext = gobblinYarnApp . getApplicationSubmissionContext ( ) ; appSubmissionContext . setApplicationType ( GOBBLIN_YARN_APPLICATION_TYPE ) ; ApplicationId applicationId = appSubmissionContext . getApplicationId ( ) ; GetNewApplicationResponse newApplicationResponse = gobblinYarnApp . getNewApplicationResponse ( ) ; Resource resource = prepareContainerResource ( newApplicationResponse ) ; Map < String , LocalResource > appMasterLocalResources = addAppMasterLocalResources ( applicationId ) ; ContainerLaunchContext amContainerLaunchContext = Records . newRecord ( ContainerLaunchContext . class ) ; amContainerLaunchContext . setLocalResources ( appMasterLocalResources ) ; amContainerLaunchContext . setEnvironment ( YarnHelixUtils . getEnvironmentVariables ( this . yarnConfiguration ) ) ; amContainerLaunchContext . setCommands ( Lists . newArrayList ( buildApplicationMasterCommand ( resource . getMemory ( ) ) ) ) ; if ( UserGroupInformation . isSecurityEnabled ( ) ) { setupSecurityTokens ( amContainerLaunchContext ) ; } appSubmissionContext . setApplicationName ( this . applicationName ) ; appSubmissionContext . setResource ( resource ) ; appSubmissionContext . setQueue ( this . appQueueName ) ; appSubmissionContext . setPriority ( Priority . newInstance ( 0 ) ) ; appSubmissionContext . setAMContainerSpec ( amContainerLaunchContext ) ; addContainerLocalResources ( applicationId ) ; LOGGER . info ( "Submitting application " + applicationId ) ; this . yarnClient . submitApplication ( appSubmissionContext ) ; LOGGER . info ( "Application successfully submitted and accepted" ) ; ApplicationReport applicationReport = this . yarnClient . getApplicationReport ( applicationId ) ; LOGGER . info ( "Application Name: " + applicationReport . getName ( ) ) ; LOGGER . info ( "Application Tracking URL: " + applicationReport . getTrackingUrl ( ) ) ; LOGGER . info ( "Application User: " + applicationReport . getUser ( ) + " Queue: " + applicationReport . getQueue ( ) ) ; return applicationId ; }
Setup and submit the Gobblin Yarn application.

@throws IOException if there's anything wrong setting up and submitting the Yarn application
@throws YarnException if there's anything wrong setting up and submitting the Yarn application