FairSchedulerConfiguration extends Configuration { public Resource getIncrementAllocation() { Long memory = null; Integer vCores = null; Map<String, Long> others = new HashMap<>(); ResourceInformation[] resourceTypes = ResourceUtils.getResourceTypesArray(); for (int i=0; i < resourceTypes.length; ++i) { String name = resourceTypes[i].getName(); String propertyKey = getAllocationIncrementPropKey(name); String propValue = get(propertyKey); if (propValue != null) { Matcher matcher = RESOURCE_REQUEST_VALUE_PATTERN.matcher(propValue); if (matcher.matches()) { long value = Long.parseLong(matcher.group(1)); String unit = matcher.group(2); long valueInDefaultUnits = getValueInDefaultUnits(value, unit, name); others.put(name, valueInDefaultUnits); } else { throw new IllegalArgumentException("Property " + propertyKey + " is not in \"value [unit]\" format: " + propValue); } } } if (others.containsKey(ResourceInformation.MEMORY_MB.getName())) { memory = others.get(ResourceInformation.MEMORY_MB.getName()); if (get(RM_SCHEDULER_INCREMENT_ALLOCATION_MB) != null) { String overridingKey = getAllocationIncrementPropKey( ResourceInformation.MEMORY_MB.getName()); LOG.warn("Configuration " + overridingKey + "=" + get(overridingKey) + " is overriding the " + RM_SCHEDULER_INCREMENT_ALLOCATION_MB + "=" + get(RM_SCHEDULER_INCREMENT_ALLOCATION_MB) + " property"); } others.remove(ResourceInformation.MEMORY_MB.getName()); } else { memory = getLong( RM_SCHEDULER_INCREMENT_ALLOCATION_MB, DEFAULT_RM_SCHEDULER_INCREMENT_ALLOCATION_MB); } if (others.containsKey(ResourceInformation.VCORES.getName())) { vCores = others.get(ResourceInformation.VCORES.getName()).intValue(); if (get(RM_SCHEDULER_INCREMENT_ALLOCATION_VCORES) != null) { String overridingKey = getAllocationIncrementPropKey( ResourceInformation.VCORES.getName()); LOG.warn("Configuration " + overridingKey + "=" + get(overridingKey) + " is overriding the " + RM_SCHEDULER_INCREMENT_ALLOCATION_VCORES + "=" + get(RM_SCHEDULER_INCREMENT_ALLOCATION_VCORES) + " property"); } others.remove(ResourceInformation.VCORES.getName()); } else { vCores = getInt( RM_SCHEDULER_INCREMENT_ALLOCATION_VCORES, DEFAULT_RM_SCHEDULER_INCREMENT_ALLOCATION_VCORES); } return Resource.newInstance(memory, vCores, others); } FairSchedulerConfiguration();  FairSchedulerConfiguration(Configuration conf); Resource getMinimumAllocation(); Resource getMaximumAllocation(); Resource getIncrementAllocation(); float getReservationThresholdIncrementMultiple(); float getLocalityThresholdNode(); float getLocalityThresholdRack(); @Deprecated boolean isContinuousSchedulingEnabled(); @Deprecated int getContinuousSchedulingSleepMs(); @Deprecated long getLocalityDelayNodeMs(); @Deprecated long getLocalityDelayRackMs(); boolean getPreemptionEnabled(); boolean getAMPreemptionEnabled(String queueName); float getPreemptionUtilizationThreshold(); boolean getAssignMultiple(); boolean isMaxAssignDynamic(); int getMaxAssign(); boolean getSizeBasedWeight(); long getWaitTimeBeforeNextStarvationCheck(); int getWaitTimeBeforeKill(); boolean getUsePortForNodeName(); float getReservableNodes(); static ConfigurableResource parseResourceConfigValue(String value); static ConfigurableResource parseResourceConfigValue(String value,
      long missing); long getUpdateInterval(); static final Logger LOG; @Deprecated
  static final String RM_SCHEDULER_INCREMENT_ALLOCATION_MB; @Deprecated
  static final int DEFAULT_RM_SCHEDULER_INCREMENT_ALLOCATION_MB; @Deprecated
  static final String RM_SCHEDULER_INCREMENT_ALLOCATION_VCORES; @Deprecated
  static final int DEFAULT_RM_SCHEDULER_INCREMENT_ALLOCATION_VCORES; static final String RM_SCHEDULER_RESERVATION_THRESHOLD_INCREMENT_MULTIPLE; static final float DEFAULT_RM_SCHEDULER_RESERVATION_THRESHOLD_INCREMENT_MULTIPLE; static final String MIGRATION_MODE; static final String NO_TERMINAL_RULE_CHECK; static final String ALLOCATION_FILE; static final String ALLOW_UNDECLARED_POOLS; static final boolean DEFAULT_ALLOW_UNDECLARED_POOLS; static final String USER_AS_DEFAULT_QUEUE; static final boolean DEFAULT_USER_AS_DEFAULT_QUEUE; static final String LOCALITY_THRESHOLD_NODE; static final float DEFAULT_LOCALITY_THRESHOLD_NODE; static final String LOCALITY_THRESHOLD_RACK; static final float DEFAULT_LOCALITY_THRESHOLD_RACK; @Deprecated
  static final String CONTINUOUS_SCHEDULING_ENABLED; @Deprecated
  static final boolean DEFAULT_CONTINUOUS_SCHEDULING_ENABLED; @Deprecated
  static final String CONTINUOUS_SCHEDULING_SLEEP_MS; @Deprecated
  static final int DEFAULT_CONTINUOUS_SCHEDULING_SLEEP_MS; static final String PREEMPTION; static final boolean DEFAULT_PREEMPTION; static final String WAIT_TIME_BEFORE_KILL; static final int DEFAULT_WAIT_TIME_BEFORE_KILL; static final String WAIT_TIME_BEFORE_NEXT_STARVATION_CHECK_MS; static final long DEFAULT_WAIT_TIME_BEFORE_NEXT_STARVATION_CHECK_MS; static final String ASSIGN_MULTIPLE; static final boolean DEFAULT_ASSIGN_MULTIPLE; static final String SIZE_BASED_WEIGHT; static final boolean DEFAULT_SIZE_BASED_WEIGHT; static final String DYNAMIC_MAX_ASSIGN; static final String MAX_ASSIGN; static final int DEFAULT_MAX_ASSIGN; static final String UPDATE_INTERVAL_MS; static final int DEFAULT_UPDATE_INTERVAL_MS; static final String RESERVABLE_NODES; static final float RESERVABLE_NODES_DEFAULT; }
@Test public void testAllocationIncrementMemoryDefaultUnit() { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RESOURCE_TYPES + "." + ResourceInformation.MEMORY_MB.getName() + FairSchedulerConfiguration.INCREMENT_ALLOCATION, "256"); FairSchedulerConfiguration fsc = new FairSchedulerConfiguration(conf); Resource minimum = Resources.createResource(0L, 0); Resource maximum = Resources.createResource(Long.MAX_VALUE, Integer.MAX_VALUE); Resource increment = fsc.getIncrementAllocation(); DominantResourceCalculator resourceCalculator = new DominantResourceCalculator(); assertEquals(1024L, resourceCalculator.normalize( Resources.createResource(769L), minimum, maximum, increment) .getMemorySize()); assertEquals(1024L, resourceCalculator.normalize( Resources.createResource(1023L), minimum, maximum, increment) .getMemorySize()); assertEquals(1024L, resourceCalculator.normalize( Resources.createResource(1024L), minimum, maximum, increment) .getMemorySize()); assertEquals(1280L, resourceCalculator.normalize( Resources.createResource(1025L), minimum, maximum, increment) .getMemorySize()); }
@Test public void testAllocationIncrementMemoryNonDefaultUnit() throws Exception { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RESOURCE_TYPES + "." + ResourceInformation.MEMORY_MB.getName() + FairSchedulerConfiguration.INCREMENT_ALLOCATION, "1 Gi"); FairSchedulerConfiguration fsc = new FairSchedulerConfiguration(conf); Resource minimum = Resources.createResource(0L, 0); Resource maximum = Resources.createResource(Long.MAX_VALUE, Integer.MAX_VALUE); Resource increment = fsc.getIncrementAllocation(); DominantResourceCalculator resourceCalculator = new DominantResourceCalculator(); assertEquals(1024L, resourceCalculator.normalize( Resources.createResource(1023L), minimum, maximum, increment) .getMemorySize()); assertEquals(1024L, resourceCalculator.normalize( Resources.createResource(1024L), minimum, maximum, increment) .getMemorySize()); assertEquals(2048L, resourceCalculator.normalize( Resources.createResource(1025L), minimum, maximum, increment) .getMemorySize()); }
@Test(expected=IllegalArgumentException.class) public void testAllocationIncrementInvalidUnit() throws Exception { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RESOURCE_TYPES + "." + ResourceInformation.MEMORY_MB.getName() + FairSchedulerConfiguration.INCREMENT_ALLOCATION, "1 Xi"); new FairSchedulerConfiguration(conf).getIncrementAllocation(); }
@Test public void testAllocationIncrementVCoreNoUnit() throws Exception { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RESOURCE_TYPES + "." + ResourceInformation.VCORES.getName() + FairSchedulerConfiguration.INCREMENT_ALLOCATION, "10"); FairSchedulerConfiguration fsc = new FairSchedulerConfiguration(conf); Resource min = Resources.createResource(0L, 0); Resource max = Resources.createResource(Long.MAX_VALUE, Integer.MAX_VALUE); Resource increment = fsc.getIncrementAllocation(); DominantResourceCalculator resourceCalculator = new DominantResourceCalculator(); assertEquals(10, resourceCalculator.normalize( Resources.createResource(0L, 9), min, max, increment) .getVirtualCores()); assertEquals(10, resourceCalculator.normalize( Resources.createResource(0L, 10), min, max, increment) .getVirtualCores()); assertEquals(20, resourceCalculator.normalize( Resources.createResource(0L, 11), min, max, increment) .getVirtualCores()); }
@Test public void testAllocationIncrementVCoreWithUnit() throws Exception { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RESOURCE_TYPES + "." + ResourceInformation.VCORES.getName() + FairSchedulerConfiguration.INCREMENT_ALLOCATION, "1k"); FairSchedulerConfiguration fsc = new FairSchedulerConfiguration(conf); Resource min = Resources.createResource(0L, 0); Resource max = Resources.createResource(Long.MAX_VALUE, Integer.MAX_VALUE); Resource increment = fsc.getIncrementAllocation(); DominantResourceCalculator resourceCalculator = new DominantResourceCalculator(); assertEquals(1000, resourceCalculator.normalize( Resources.createResource(0L, 999), min, max, increment) .getVirtualCores()); assertEquals(1000, resourceCalculator.normalize( Resources.createResource(0L, 1000), min, max, increment) .getVirtualCores()); assertEquals(2000, resourceCalculator.normalize( Resources.createResource(0L, 1001), min, max, increment) .getVirtualCores()); }
@Test public void testAllocationIncrementCustomResource() { try { initResourceTypes(); Configuration conf = new Configuration(); conf.set(YarnConfiguration.RESOURCE_TYPES + ".a-custom-resource" + FairSchedulerConfiguration.INCREMENT_ALLOCATION, "10"); FairSchedulerConfiguration fsc = new FairSchedulerConfiguration(conf); Resource increment = fsc.getIncrementAllocation(); DominantResourceCalculator calculator = new DominantResourceCalculator(); Resource min = Resources.createResource(0L, 0); Resource max = Resource.newInstance(Long.MAX_VALUE, Integer.MAX_VALUE, Collections.singletonMap(A_CUSTOM_RESOURCE, Long.MAX_VALUE / UnitsConversionUtil.convert("k", "", 1L))); assertEquals(customResourceInformation(10000L, ""), calculator.normalize(customResource(9999L, ""), min, max, increment) .getResourceInformation(A_CUSTOM_RESOURCE)); assertEquals(customResourceInformation(10000L, ""), calculator.normalize(customResource(10000L, ""), min, max, increment) .getResourceInformation(A_CUSTOM_RESOURCE)); assertEquals(customResourceInformation(20000L, ""), calculator.normalize(customResource(19999L, ""), min, max, increment) .getResourceInformation(A_CUSTOM_RESOURCE)); assertEquals(customResourceInformation(10L, "k"), calculator.normalize(customResource(9L, "k"), min, max, increment) .getResourceInformation(A_CUSTOM_RESOURCE)); assertEquals(customResourceInformation(10L, "k"), calculator.normalize(customResource(10L, "k"), min, max, increment) .getResourceInformation(A_CUSTOM_RESOURCE)); assertEquals(customResourceInformation(20L, "k"), calculator.normalize(customResource(11L, "k"), min, max, increment) .getResourceInformation(A_CUSTOM_RESOURCE)); } finally { ResourceUtils.resetResourceTypes(new Configuration()); } }
@Test public void testMemoryIncrementConfiguredViaMultipleProperties() { TestAppender testAppender = new TestAppender(); Logger logger = LogManager.getRootLogger(); logger.addAppender(testAppender); try { Configuration conf = new Configuration(); conf.set("yarn.scheduler.increment-allocation-mb", "7"); conf.set(YarnConfiguration.RESOURCE_TYPES + "." + ResourceInformation.MEMORY_MB.getName() + FairSchedulerConfiguration.INCREMENT_ALLOCATION, "13"); FairSchedulerConfiguration fsc = new FairSchedulerConfiguration(conf); Resource increment = fsc.getIncrementAllocation(); Assert.assertEquals(13L, increment.getMemorySize()); assertTrue("Warning message is not logged when specifying memory " + "increment via multiple properties", testAppender.getLogEvents().stream().anyMatch( e -> e.getLevel() == Level.WARN && ("Configuration " + "yarn.resource-types.memory-mb.increment-allocation=13 is " + "overriding the yarn.scheduler.increment-allocation-mb=7 " + "property").equals(e.getMessage()))); } finally { logger.removeAppender(testAppender); } }
@Test public void testCpuIncrementConfiguredViaMultipleProperties() { TestAppender testAppender = new TestAppender(); Logger logger = LogManager.getRootLogger(); logger.addAppender(testAppender); try { Configuration conf = new Configuration(); conf.set("yarn.scheduler.increment-allocation-vcores", "7"); conf.set(YarnConfiguration.RESOURCE_TYPES + "." + ResourceInformation.VCORES.getName() + FairSchedulerConfiguration.INCREMENT_ALLOCATION, "13"); FairSchedulerConfiguration fsc = new FairSchedulerConfiguration(conf); Resource increment = fsc.getIncrementAllocation(); Assert.assertEquals(13, increment.getVirtualCores()); assertTrue("Warning message is not logged when specifying CPU vCores " + "increment via multiple properties", testAppender.getLogEvents().stream().anyMatch( e -> e.getLevel() == Level.WARN && ("Configuration " + "yarn.resource-types.vcores.increment-allocation=13 is " + "overriding the yarn.scheduler.increment-allocation-vcores=7 " + "property").equals(e.getMessage()))); } finally { logger.removeAppender(testAppender); } }