@Test(groups = {"gobblin.yarn", "disabledOnTravis"}, dependsOnMethods = "testReleasedContainerCache") public void testBuildContainerCommand() throws Exception { Config modifiedConfig = this.config .withValue(GobblinYarnConfigurationKeys.CONTAINER_JVM_MEMORY_OVERHEAD_MBS_KEY, ConfigValueFactory.fromAnyRef("10")) .withValue(GobblinYarnConfigurationKeys.CONTAINER_JVM_MEMORY_XMX_RATIO_KEY, ConfigValueFactory.fromAnyRef("0.8")); TestYarnService yarnService = new TestYarnService(modifiedConfig, "testApp2", "appId2", this.clusterConf, FileSystem.getLocal(new Configuration()), this.eventBus); ContainerId containerId = ContainerId.newInstance(ApplicationAttemptId.newInstance(ApplicationId.newInstance(1, 0), 0), 0); Resource resource = Resource.newInstance(2048, 1); Container container = Container.newInstance(containerId, null, null, resource, null, null); String command = yarnService.buildContainerCommand(container, "helixInstance1"); Assert.assertTrue(command.contains("-Xmx1628")); }
@VisibleForTesting protected String buildContainerCommand(Container container, String helixInstanceName) { String containerProcessName = GobblinYarnTaskRunner.class.getSimpleName(); StringBuilder containerCommand = new StringBuilder() .append(ApplicationConstants.Environment.JAVA_HOME.$()).append("/bin/java") .append(" -Xmx").append((int) (container.getResource().getMemory() * this.jvmMemoryXmxRatio) - this.jvmMemoryOverheadMbs).append("M") .append(" -D").append(GobblinYarnConfigurationKeys.JVM_USER_TIMEZONE_CONFIG).append("=").append(this.containerTimezone) .append(" -D").append(GobblinYarnConfigurationKeys.GOBBLIN_YARN_CONTAINER_LOG_DIR_NAME).append("=").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR) .append(" -D").append(GobblinYarnConfigurationKeys.GOBBLIN_YARN_CONTAINER_LOG_FILE_NAME).append("=").append(containerProcessName).append(".").append(ApplicationConstants.STDOUT) .append(" ").append(JvmUtils.formatJvmArguments(this.containerJvmArgs)) .append(" ").append(GobblinYarnTaskRunner.class.getName()) .append(" --").append(GobblinClusterConfigurationKeys.APPLICATION_NAME_OPTION_NAME) .append(" ").append(this.applicationName) .append(" --").append(GobblinClusterConfigurationKeys.APPLICATION_ID_OPTION_NAME) .append(" ").append(this.applicationId) .append(" --").append(GobblinClusterConfigurationKeys.HELIX_INSTANCE_NAME_OPTION_NAME) .append(" ").append(helixInstanceName); if (!Strings.isNullOrEmpty(this.helixInstanceTags)) { containerCommand.append(" --").append(GobblinClusterConfigurationKeys.HELIX_INSTANCE_TAGS_OPTION_NAME) .append(" ").append(helixInstanceTags); } return containerCommand.append(" 1>").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR).append(File.separator).append( containerProcessName).append(".").append(ApplicationConstants.STDOUT) .append(" 2>").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR).append(File.separator).append( containerProcessName).append(".").append(ApplicationConstants.STDERR).toString(); }
YarnService extends AbstractIdleService { @VisibleForTesting protected String buildContainerCommand(Container container, String helixInstanceName) { String containerProcessName = GobblinYarnTaskRunner.class.getSimpleName(); StringBuilder containerCommand = new StringBuilder() .append(ApplicationConstants.Environment.JAVA_HOME.$()).append("/bin/java") .append(" -Xmx").append((int) (container.getResource().getMemory() * this.jvmMemoryXmxRatio) - this.jvmMemoryOverheadMbs).append("M") .append(" -D").append(GobblinYarnConfigurationKeys.JVM_USER_TIMEZONE_CONFIG).append("=").append(this.containerTimezone) .append(" -D").append(GobblinYarnConfigurationKeys.GOBBLIN_YARN_CONTAINER_LOG_DIR_NAME).append("=").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR) .append(" -D").append(GobblinYarnConfigurationKeys.GOBBLIN_YARN_CONTAINER_LOG_FILE_NAME).append("=").append(containerProcessName).append(".").append(ApplicationConstants.STDOUT) .append(" ").append(JvmUtils.formatJvmArguments(this.containerJvmArgs)) .append(" ").append(GobblinYarnTaskRunner.class.getName()) .append(" --").append(GobblinClusterConfigurationKeys.APPLICATION_NAME_OPTION_NAME) .append(" ").append(this.applicationName) .append(" --").append(GobblinClusterConfigurationKeys.APPLICATION_ID_OPTION_NAME) .append(" ").append(this.applicationId) .append(" --").append(GobblinClusterConfigurationKeys.HELIX_INSTANCE_NAME_OPTION_NAME) .append(" ").append(helixInstanceName); if (!Strings.isNullOrEmpty(this.helixInstanceTags)) { containerCommand.append(" --").append(GobblinClusterConfigurationKeys.HELIX_INSTANCE_TAGS_OPTION_NAME) .append(" ").append(helixInstanceTags); } return containerCommand.append(" 1>").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR).append(File.separator).append( containerProcessName).append(".").append(ApplicationConstants.STDOUT) .append(" 2>").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR).append(File.separator).append( containerProcessName).append(".").append(ApplicationConstants.STDERR).toString(); } }
YarnService extends AbstractIdleService { @VisibleForTesting protected String buildContainerCommand(Container container, String helixInstanceName) { String containerProcessName = GobblinYarnTaskRunner.class.getSimpleName(); StringBuilder containerCommand = new StringBuilder() .append(ApplicationConstants.Environment.JAVA_HOME.$()).append("/bin/java") .append(" -Xmx").append((int) (container.getResource().getMemory() * this.jvmMemoryXmxRatio) - this.jvmMemoryOverheadMbs).append("M") .append(" -D").append(GobblinYarnConfigurationKeys.JVM_USER_TIMEZONE_CONFIG).append("=").append(this.containerTimezone) .append(" -D").append(GobblinYarnConfigurationKeys.GOBBLIN_YARN_CONTAINER_LOG_DIR_NAME).append("=").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR) .append(" -D").append(GobblinYarnConfigurationKeys.GOBBLIN_YARN_CONTAINER_LOG_FILE_NAME).append("=").append(containerProcessName).append(".").append(ApplicationConstants.STDOUT) .append(" ").append(JvmUtils.formatJvmArguments(this.containerJvmArgs)) .append(" ").append(GobblinYarnTaskRunner.class.getName()) .append(" --").append(GobblinClusterConfigurationKeys.APPLICATION_NAME_OPTION_NAME) .append(" ").append(this.applicationName) .append(" --").append(GobblinClusterConfigurationKeys.APPLICATION_ID_OPTION_NAME) .append(" ").append(this.applicationId) .append(" --").append(GobblinClusterConfigurationKeys.HELIX_INSTANCE_NAME_OPTION_NAME) .append(" ").append(helixInstanceName); if (!Strings.isNullOrEmpty(this.helixInstanceTags)) { containerCommand.append(" --").append(GobblinClusterConfigurationKeys.HELIX_INSTANCE_TAGS_OPTION_NAME) .append(" ").append(helixInstanceTags); } return containerCommand.append(" 1>").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR).append(File.separator).append( containerProcessName).append(".").append(ApplicationConstants.STDOUT) .append(" 2>").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR).append(File.separator).append( containerProcessName).append(".").append(ApplicationConstants.STDERR).toString(); } YarnService(Config config, String applicationName, String applicationId, YarnConfiguration yarnConfiguration,
      FileSystem fs, EventBus eventBus, HelixManager helixManager); }
YarnService extends AbstractIdleService { @VisibleForTesting protected String buildContainerCommand(Container container, String helixInstanceName) { String containerProcessName = GobblinYarnTaskRunner.class.getSimpleName(); StringBuilder containerCommand = new StringBuilder() .append(ApplicationConstants.Environment.JAVA_HOME.$()).append("/bin/java") .append(" -Xmx").append((int) (container.getResource().getMemory() * this.jvmMemoryXmxRatio) - this.jvmMemoryOverheadMbs).append("M") .append(" -D").append(GobblinYarnConfigurationKeys.JVM_USER_TIMEZONE_CONFIG).append("=").append(this.containerTimezone) .append(" -D").append(GobblinYarnConfigurationKeys.GOBBLIN_YARN_CONTAINER_LOG_DIR_NAME).append("=").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR) .append(" -D").append(GobblinYarnConfigurationKeys.GOBBLIN_YARN_CONTAINER_LOG_FILE_NAME).append("=").append(containerProcessName).append(".").append(ApplicationConstants.STDOUT) .append(" ").append(JvmUtils.formatJvmArguments(this.containerJvmArgs)) .append(" ").append(GobblinYarnTaskRunner.class.getName()) .append(" --").append(GobblinClusterConfigurationKeys.APPLICATION_NAME_OPTION_NAME) .append(" ").append(this.applicationName) .append(" --").append(GobblinClusterConfigurationKeys.APPLICATION_ID_OPTION_NAME) .append(" ").append(this.applicationId) .append(" --").append(GobblinClusterConfigurationKeys.HELIX_INSTANCE_NAME_OPTION_NAME) .append(" ").append(helixInstanceName); if (!Strings.isNullOrEmpty(this.helixInstanceTags)) { containerCommand.append(" --").append(GobblinClusterConfigurationKeys.HELIX_INSTANCE_TAGS_OPTION_NAME) .append(" ").append(helixInstanceTags); } return containerCommand.append(" 1>").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR).append(File.separator).append( containerProcessName).append(".").append(ApplicationConstants.STDOUT) .append(" 2>").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR).append(File.separator).append( containerProcessName).append(".").append(ApplicationConstants.STDERR).toString(); } YarnService(Config config, String applicationName, String applicationId, YarnConfiguration yarnConfiguration,
      FileSystem fs, EventBus eventBus, HelixManager helixManager); @SuppressWarnings("unused") @Subscribe void handleNewContainerRequest(NewContainerRequest newContainerRequest); @SuppressWarnings("unused") @Subscribe void handleContainerShutdownRequest(ContainerShutdownRequest containerShutdownRequest); @Subscribe void handleContainerReleaseRequest(ContainerReleaseRequest containerReleaseRequest); void updateToken(); synchronized void requestTargetNumberOfContainers(int numTargetContainers, Set<String> inUseInstances); }
YarnService extends AbstractIdleService { @VisibleForTesting protected String buildContainerCommand(Container container, String helixInstanceName) { String containerProcessName = GobblinYarnTaskRunner.class.getSimpleName(); StringBuilder containerCommand = new StringBuilder() .append(ApplicationConstants.Environment.JAVA_HOME.$()).append("/bin/java") .append(" -Xmx").append((int) (container.getResource().getMemory() * this.jvmMemoryXmxRatio) - this.jvmMemoryOverheadMbs).append("M") .append(" -D").append(GobblinYarnConfigurationKeys.JVM_USER_TIMEZONE_CONFIG).append("=").append(this.containerTimezone) .append(" -D").append(GobblinYarnConfigurationKeys.GOBBLIN_YARN_CONTAINER_LOG_DIR_NAME).append("=").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR) .append(" -D").append(GobblinYarnConfigurationKeys.GOBBLIN_YARN_CONTAINER_LOG_FILE_NAME).append("=").append(containerProcessName).append(".").append(ApplicationConstants.STDOUT) .append(" ").append(JvmUtils.formatJvmArguments(this.containerJvmArgs)) .append(" ").append(GobblinYarnTaskRunner.class.getName()) .append(" --").append(GobblinClusterConfigurationKeys.APPLICATION_NAME_OPTION_NAME) .append(" ").append(this.applicationName) .append(" --").append(GobblinClusterConfigurationKeys.APPLICATION_ID_OPTION_NAME) .append(" ").append(this.applicationId) .append(" --").append(GobblinClusterConfigurationKeys.HELIX_INSTANCE_NAME_OPTION_NAME) .append(" ").append(helixInstanceName); if (!Strings.isNullOrEmpty(this.helixInstanceTags)) { containerCommand.append(" --").append(GobblinClusterConfigurationKeys.HELIX_INSTANCE_TAGS_OPTION_NAME) .append(" ").append(helixInstanceTags); } return containerCommand.append(" 1>").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR).append(File.separator).append( containerProcessName).append(".").append(ApplicationConstants.STDOUT) .append(" 2>").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR).append(File.separator).append( containerProcessName).append(".").append(ApplicationConstants.STDERR).toString(); } YarnService(Config config, String applicationName, String applicationId, YarnConfiguration yarnConfiguration,
      FileSystem fs, EventBus eventBus, HelixManager helixManager); @SuppressWarnings("unused") @Subscribe void handleNewContainerRequest(NewContainerRequest newContainerRequest); @SuppressWarnings("unused") @Subscribe void handleContainerShutdownRequest(ContainerShutdownRequest containerShutdownRequest); @Subscribe void handleContainerReleaseRequest(ContainerReleaseRequest containerReleaseRequest); void updateToken(); synchronized void requestTargetNumberOfContainers(int numTargetContainers, Set<String> inUseInstances);  }