FSYarnSiteConverter { @SuppressWarnings({"deprecation", "checkstyle:linelength"}) public void convertSiteProperties(Configuration conf, Configuration yarnSiteConfig, boolean drfUsed, boolean enableAsyncScheduler) { yarnSiteConfig.set(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class.getCanonicalName()); if (conf.getBoolean( FairSchedulerConfiguration.CONTINUOUS_SCHEDULING_ENABLED, FairSchedulerConfiguration.DEFAULT_CONTINUOUS_SCHEDULING_ENABLED)) { yarnSiteConfig.setBoolean( CapacitySchedulerConfiguration.SCHEDULE_ASYNCHRONOUSLY_ENABLE, true); int interval = conf.getInt( FairSchedulerConfiguration.CONTINUOUS_SCHEDULING_SLEEP_MS, FairSchedulerConfiguration.DEFAULT_CONTINUOUS_SCHEDULING_SLEEP_MS); yarnSiteConfig.setInt(PREFIX + "schedule-asynchronously.scheduling-interval-ms", interval); } if (conf.getBoolean(FairSchedulerConfiguration.PREEMPTION, FairSchedulerConfiguration.DEFAULT_PREEMPTION)) { yarnSiteConfig.setBoolean( YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS, true); preemptionEnabled = true; int waitTimeBeforeKill = conf.getInt( FairSchedulerConfiguration.WAIT_TIME_BEFORE_KILL, FairSchedulerConfiguration.DEFAULT_WAIT_TIME_BEFORE_KILL); yarnSiteConfig.setInt( CapacitySchedulerConfiguration.PREEMPTION_WAIT_TIME_BEFORE_KILL, waitTimeBeforeKill); long waitBeforeNextStarvationCheck = conf.getLong( FairSchedulerConfiguration.WAIT_TIME_BEFORE_NEXT_STARVATION_CHECK_MS, FairSchedulerConfiguration.DEFAULT_WAIT_TIME_BEFORE_NEXT_STARVATION_CHECK_MS); yarnSiteConfig.setLong( CapacitySchedulerConfiguration.PREEMPTION_MONITORING_INTERVAL, waitBeforeNextStarvationCheck); } if (conf.getBoolean(FairSchedulerConfiguration.ASSIGN_MULTIPLE, FairSchedulerConfiguration.DEFAULT_ASSIGN_MULTIPLE)) { yarnSiteConfig.setBoolean( CapacitySchedulerConfiguration.ASSIGN_MULTIPLE_ENABLED, true); } else { yarnSiteConfig.setBoolean( CapacitySchedulerConfiguration.ASSIGN_MULTIPLE_ENABLED, false); } int maxAssign = conf.getInt(FairSchedulerConfiguration.MAX_ASSIGN, FairSchedulerConfiguration.DEFAULT_MAX_ASSIGN); if (maxAssign != FairSchedulerConfiguration.DEFAULT_MAX_ASSIGN) { yarnSiteConfig.setInt( CapacitySchedulerConfiguration.MAX_ASSIGN_PER_HEARTBEAT, maxAssign); } float localityThresholdNode = conf.getFloat( FairSchedulerConfiguration.LOCALITY_THRESHOLD_NODE, FairSchedulerConfiguration.DEFAULT_LOCALITY_THRESHOLD_NODE); if (localityThresholdNode != FairSchedulerConfiguration.DEFAULT_LOCALITY_THRESHOLD_NODE) { yarnSiteConfig.setFloat(CapacitySchedulerConfiguration.NODE_LOCALITY_DELAY, localityThresholdNode); } float localityThresholdRack = conf.getFloat( FairSchedulerConfiguration.LOCALITY_THRESHOLD_RACK, FairSchedulerConfiguration.DEFAULT_LOCALITY_THRESHOLD_RACK); if (localityThresholdRack != FairSchedulerConfiguration.DEFAULT_LOCALITY_THRESHOLD_RACK) { yarnSiteConfig.setFloat( CapacitySchedulerConfiguration.RACK_LOCALITY_ADDITIONAL_DELAY, localityThresholdRack); } if (conf.getBoolean(FairSchedulerConfiguration.SIZE_BASED_WEIGHT, FairSchedulerConfiguration.DEFAULT_SIZE_BASED_WEIGHT)) { sizeBasedWeight = true; } if (drfUsed) { yarnSiteConfig.set( CapacitySchedulerConfiguration.RESOURCE_CALCULATOR_CLASS, DominantResourceCalculator.class.getCanonicalName()); } if (enableAsyncScheduler) { yarnSiteConfig.setBoolean(CapacitySchedulerConfiguration.SCHEDULE_ASYNCHRONOUSLY_ENABLE, true); } }  @SuppressWarnings({"deprecation", "checkstyle:linelength"}) void convertSiteProperties(Configuration conf,
      Configuration yarnSiteConfig, boolean drfUsed, boolean enableAsyncScheduler); boolean isPreemptionEnabled(); boolean isSizeBasedWeight();  }
@SuppressWarnings("deprecation") @Test public void testSiteContinuousSchedulingConversion() { yarnConfig.setBoolean( FairSchedulerConfiguration.CONTINUOUS_SCHEDULING_ENABLED, true); yarnConfig.setInt( FairSchedulerConfiguration.CONTINUOUS_SCHEDULING_SLEEP_MS, 666); converter.convertSiteProperties(yarnConfig, yarnConvertedConfig, false, false); assertTrue("Cont. scheduling", yarnConvertedConfig.getBoolean( CapacitySchedulerConfiguration.SCHEDULE_ASYNCHRONOUSLY_ENABLE, false)); assertEquals("Scheduling interval", 666, yarnConvertedConfig.getInt( "yarn.scheduler.capacity.schedule-asynchronously" + ".scheduling-interval-ms", -1)); }
@Test public void testSitePreemptionConversion() { yarnConfig.setBoolean(FairSchedulerConfiguration.PREEMPTION, true); yarnConfig.setInt(FairSchedulerConfiguration.WAIT_TIME_BEFORE_KILL, 123); yarnConfig.setInt( FairSchedulerConfiguration.WAIT_TIME_BEFORE_NEXT_STARVATION_CHECK_MS, 321); converter.convertSiteProperties(yarnConfig, yarnConvertedConfig, false, false); assertTrue("Preemption enabled", yarnConvertedConfig.getBoolean( YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS, false)); assertEquals("Wait time before kill", 123, yarnConvertedConfig.getInt( CapacitySchedulerConfiguration.PREEMPTION_WAIT_TIME_BEFORE_KILL, -1)); assertEquals("Starvation check wait time", 321, yarnConvertedConfig.getInt( CapacitySchedulerConfiguration.PREEMPTION_MONITORING_INTERVAL, -1)); }
@Test public void testSiteAssignMultipleConversion() { yarnConfig.setBoolean(FairSchedulerConfiguration.ASSIGN_MULTIPLE, true); converter.convertSiteProperties(yarnConfig, yarnConvertedConfig, false, false); assertTrue("Assign multiple", yarnConvertedConfig.getBoolean( CapacitySchedulerConfiguration.ASSIGN_MULTIPLE_ENABLED, false)); }
@Test public void testSiteMaxAssignConversion() { yarnConfig.setInt(FairSchedulerConfiguration.MAX_ASSIGN, 111); converter.convertSiteProperties(yarnConfig, yarnConvertedConfig, false, false); assertEquals("Max assign", 111, yarnConvertedConfig.getInt( CapacitySchedulerConfiguration.MAX_ASSIGN_PER_HEARTBEAT, -1)); }
@Test public void testSiteLocalityThresholdConversion() { yarnConfig.set(FairSchedulerConfiguration.LOCALITY_THRESHOLD_NODE, "123.123"); yarnConfig.set(FairSchedulerConfiguration.LOCALITY_THRESHOLD_RACK, "321.321"); converter.convertSiteProperties(yarnConfig, yarnConvertedConfig, false, false); assertEquals("Locality threshold node", "123.123", yarnConvertedConfig.get( CapacitySchedulerConfiguration.NODE_LOCALITY_DELAY)); assertEquals("Locality threshold rack", "321.321", yarnConvertedConfig.get( CapacitySchedulerConfiguration.RACK_LOCALITY_ADDITIONAL_DELAY)); }
@Test public void testSiteDrfEnabledConversion() { converter.convertSiteProperties(yarnConfig, yarnConvertedConfig, true, false); assertEquals("Resource calculator type", DominantResourceCalculator.class, yarnConvertedConfig.getClass( CapacitySchedulerConfiguration.RESOURCE_CALCULATOR_CLASS, null)); }
@Test public void testSiteDrfDisabledConversion() { converter.convertSiteProperties(yarnConfig, yarnConvertedConfig, false, false); assertEquals("Resource calculator type", DefaultResourceCalculator.class, yarnConvertedConfig.getClass( CapacitySchedulerConfiguration.RESOURCE_CALCULATOR_CLASS, CapacitySchedulerConfiguration.DEFAULT_RESOURCE_CALCULATOR_CLASS)); }
@Test public void testAsyncSchedulingEnabledConversion() { converter.convertSiteProperties(yarnConfig, yarnConvertedConfig, true, true); assertTrue("Asynchronous scheduling", yarnConvertedConfig.getBoolean( CapacitySchedulerConfiguration.SCHEDULE_ASYNCHRONOUSLY_ENABLE, CapacitySchedulerConfiguration.DEFAULT_SCHEDULE_ASYNCHRONOUSLY_ENABLE)); }
@Test public void testAsyncSchedulingDisabledConversion() { converter.convertSiteProperties(yarnConfig, yarnConvertedConfig, false, false); assertFalse("Asynchronous scheduling", yarnConvertedConfig.getBoolean( CapacitySchedulerConfiguration.SCHEDULE_ASYNCHRONOUSLY_ENABLE, CapacitySchedulerConfiguration.DEFAULT_SCHEDULE_ASYNCHRONOUSLY_ENABLE)); }