@Test(enabled=false, groups = { "disabledOnTravis" }, dependsOnMethods = "testCreateHelixCluster") public void testSetupAndSubmitApplication() throws Exception { HelixUtils.createGobblinHelixCluster( this.config.getString(GobblinClusterConfigurationKeys.ZK_CONNECTION_STRING_KEY), this.config.getString(GobblinClusterConfigurationKeys.HELIX_CLUSTER_NAME_KEY)); this.gobblinYarnAppLauncher.startYarnClient(); this.applicationId = this.gobblinYarnAppLauncher.setupAndSubmitApplication(); int i; for (i = 0; i < 120; i++) { if (yarnClient.getApplicationReport(applicationId).getYarnApplicationState() == YarnApplicationState.RUNNING) { break; } Thread.sleep(1000); } Assert.assertTrue(i < 120, "timed out waiting for RUNNING state"); Thread.sleep(10000); Assert.assertEquals(yarnClient.getApplicationReport(applicationId).getYarnApplicationState(), YarnApplicationState.RUNNING, "Application may have aborted"); }
@VisibleForTesting ApplicationId setupAndSubmitApplication() throws IOException, YarnException { YarnClientApplication gobblinYarnApp = this.yarnClient.createApplication(); ApplicationSubmissionContext appSubmissionContext = gobblinYarnApp.getApplicationSubmissionContext(); appSubmissionContext.setApplicationType(GOBBLIN_YARN_APPLICATION_TYPE); appSubmissionContext.setMaxAppAttempts(ConfigUtils.getInt(config, GobblinYarnConfigurationKeys.APP_MASTER_MAX_ATTEMPTS_KEY, GobblinYarnConfigurationKeys.DEFAULT_APP_MASTER_MAX_ATTEMPTS_KEY)); 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(applicationId.toString(), resource.getMemory()))); Map<ApplicationAccessType, String> acls = new HashMap<>(1); acls.put(ApplicationAccessType.VIEW_APP, this.appViewAcl); amContainerLaunchContext.setApplicationACLs(acls); 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 " + sanitizeApplicationId(applicationId.toString())); 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; }
GobblinYarnAppLauncher { @VisibleForTesting ApplicationId setupAndSubmitApplication() throws IOException, YarnException { YarnClientApplication gobblinYarnApp = this.yarnClient.createApplication(); ApplicationSubmissionContext appSubmissionContext = gobblinYarnApp.getApplicationSubmissionContext(); appSubmissionContext.setApplicationType(GOBBLIN_YARN_APPLICATION_TYPE); appSubmissionContext.setMaxAppAttempts(ConfigUtils.getInt(config, GobblinYarnConfigurationKeys.APP_MASTER_MAX_ATTEMPTS_KEY, GobblinYarnConfigurationKeys.DEFAULT_APP_MASTER_MAX_ATTEMPTS_KEY)); 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(applicationId.toString(), resource.getMemory()))); Map<ApplicationAccessType, String> acls = new HashMap<>(1); acls.put(ApplicationAccessType.VIEW_APP, this.appViewAcl); amContainerLaunchContext.setApplicationACLs(acls); 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 " + sanitizeApplicationId(applicationId.toString())); 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; } }
GobblinYarnAppLauncher { @VisibleForTesting ApplicationId setupAndSubmitApplication() throws IOException, YarnException { YarnClientApplication gobblinYarnApp = this.yarnClient.createApplication(); ApplicationSubmissionContext appSubmissionContext = gobblinYarnApp.getApplicationSubmissionContext(); appSubmissionContext.setApplicationType(GOBBLIN_YARN_APPLICATION_TYPE); appSubmissionContext.setMaxAppAttempts(ConfigUtils.getInt(config, GobblinYarnConfigurationKeys.APP_MASTER_MAX_ATTEMPTS_KEY, GobblinYarnConfigurationKeys.DEFAULT_APP_MASTER_MAX_ATTEMPTS_KEY)); 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(applicationId.toString(), resource.getMemory()))); Map<ApplicationAccessType, String> acls = new HashMap<>(1); acls.put(ApplicationAccessType.VIEW_APP, this.appViewAcl); amContainerLaunchContext.setApplicationACLs(acls); 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 " + sanitizeApplicationId(applicationId.toString())); 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; } GobblinYarnAppLauncher(Config config, YarnConfiguration yarnConfiguration); }
GobblinYarnAppLauncher { @VisibleForTesting ApplicationId setupAndSubmitApplication() throws IOException, YarnException { YarnClientApplication gobblinYarnApp = this.yarnClient.createApplication(); ApplicationSubmissionContext appSubmissionContext = gobblinYarnApp.getApplicationSubmissionContext(); appSubmissionContext.setApplicationType(GOBBLIN_YARN_APPLICATION_TYPE); appSubmissionContext.setMaxAppAttempts(ConfigUtils.getInt(config, GobblinYarnConfigurationKeys.APP_MASTER_MAX_ATTEMPTS_KEY, GobblinYarnConfigurationKeys.DEFAULT_APP_MASTER_MAX_ATTEMPTS_KEY)); 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(applicationId.toString(), resource.getMemory()))); Map<ApplicationAccessType, String> acls = new HashMap<>(1); acls.put(ApplicationAccessType.VIEW_APP, this.appViewAcl); amContainerLaunchContext.setApplicationACLs(acls); 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 " + sanitizeApplicationId(applicationId.toString())); 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; } GobblinYarnAppLauncher(Config config, YarnConfiguration yarnConfiguration); void launch(); synchronized void stop(); @Subscribe void handleApplicationReportArrivalEvent(ApplicationReportArrivalEvent applicationReportArrivalEvent); @Subscribe void handleGetApplicationReportFailureEvent(
      GetApplicationReportFailureEvent getApplicationReportFailureEvent); static void main(String[] args); }
GobblinYarnAppLauncher { @VisibleForTesting ApplicationId setupAndSubmitApplication() throws IOException, YarnException { YarnClientApplication gobblinYarnApp = this.yarnClient.createApplication(); ApplicationSubmissionContext appSubmissionContext = gobblinYarnApp.getApplicationSubmissionContext(); appSubmissionContext.setApplicationType(GOBBLIN_YARN_APPLICATION_TYPE); appSubmissionContext.setMaxAppAttempts(ConfigUtils.getInt(config, GobblinYarnConfigurationKeys.APP_MASTER_MAX_ATTEMPTS_KEY, GobblinYarnConfigurationKeys.DEFAULT_APP_MASTER_MAX_ATTEMPTS_KEY)); 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(applicationId.toString(), resource.getMemory()))); Map<ApplicationAccessType, String> acls = new HashMap<>(1); acls.put(ApplicationAccessType.VIEW_APP, this.appViewAcl); amContainerLaunchContext.setApplicationACLs(acls); 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 " + sanitizeApplicationId(applicationId.toString())); 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; } GobblinYarnAppLauncher(Config config, YarnConfiguration yarnConfiguration); void launch(); synchronized void stop(); @Subscribe void handleApplicationReportArrivalEvent(ApplicationReportArrivalEvent applicationReportArrivalEvent); @Subscribe void handleGetApplicationReportFailureEvent(
      GetApplicationReportFailureEvent getApplicationReportFailureEvent); static void main(String[] args); static final String GOBBLIN_YARN_CONFIG_OUTPUT_PATH; static final String GOBBLIN_YARN_APP_LAUNCHER_MODE; static final String DEFAULT_GOBBLIN_YARN_APP_LAUNCHER_MODE; static final String AZKABAN_APP_LAUNCHER_MODE_KEY; }