YARN-7370: Preemption properties should be refreshable. Contrubted by Gergely Novák.

(cherry picked from commit e6ec02001fc4eed9eb51c8653d8f931135e49eda)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/SchedulingMonitor.java

@@ -75,9 +75,13 @@ public class SchedulingMonitor extends AbstractService {
         return t;
       }
     });
+    schedulePreemptionChecker();
+    super.serviceStart();
+  }
+
+  private void schedulePreemptionChecker() {
     handler = ses.scheduleAtFixedRate(new PreemptionChecker(),
         0, monitorInterval, TimeUnit.MILLISECONDS);
-    super.serviceStart();
   }
 
   @Override
@@ -100,8 +104,13 @@ public class SchedulingMonitor extends AbstractService {
     @Override
     public void run() {
       try {
-        //invoke the preemption policy
-        invokePolicy();
+        if (monitorInterval != scheduleEditPolicy.getMonitoringInterval()) {
+          handler.cancel(true);
+          monitorInterval = scheduleEditPolicy.getMonitoringInterval();
+          schedulePreemptionChecker();
+        } else {
+          invokePolicy();
+        }
       } catch (Throwable t) {
         // The preemption monitor does not alter structures nor do structures
         // persist across invocations. Therefore, log, skip, and retry.

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/ProportionalCapacityPreemptionPolicy.java

@@ -107,6 +107,9 @@ public class ProportionalCapacityPreemptionPolicy
   private float minimumThresholdForIntraQueuePreemption;
   private IntraQueuePreemptionOrderPolicy intraQueuePreemptionOrderPolicy;
 
+  // Current configuration
+  private CapacitySchedulerConfiguration csConfig;
+
   // Pointer to other RM components
   private RMContext rmContext;
   private ResourceCalculator rc;
@@ -120,8 +123,7 @@ public class ProportionalCapacityPreemptionPolicy
       new HashMap<>();
   private Map<String, LinkedHashSet<String>> partitionToUnderServedQueues =
       new HashMap<String, LinkedHashSet<String>>();
-  private List<PreemptionCandidatesSelector>
-      candidatesSelectionPolicies = new ArrayList<>();
+  private List<PreemptionCandidatesSelector> candidatesSelectionPolicies;
   private Set<String> allPartitions;
   private Set<String> leafQueueNames;
 
@@ -159,60 +161,68 @@ public class ProportionalCapacityPreemptionPolicy
     }
     rmContext = context;
     scheduler = (CapacityScheduler) sched;
-    CapacitySchedulerConfiguration csConfig = scheduler.getConfiguration();
+    rc = scheduler.getResourceCalculator();
+    nlm = scheduler.getRMContext().getNodeLabelManager();
+    updateConfigIfNeeded();
+  }
 
-    maxIgnoredOverCapacity = csConfig.getDouble(
+  private void updateConfigIfNeeded() {
+    CapacitySchedulerConfiguration config = scheduler.getConfiguration();
+    if (config == csConfig) {
+      return;
+    }
+
+    maxIgnoredOverCapacity = config.getDouble(
         CapacitySchedulerConfiguration.PREEMPTION_MAX_IGNORED_OVER_CAPACITY,
         CapacitySchedulerConfiguration.DEFAULT_PREEMPTION_MAX_IGNORED_OVER_CAPACITY);
 
-    naturalTerminationFactor = csConfig.getDouble(
+    naturalTerminationFactor = config.getDouble(
         CapacitySchedulerConfiguration.PREEMPTION_NATURAL_TERMINATION_FACTOR,
         CapacitySchedulerConfiguration.DEFAULT_PREEMPTION_NATURAL_TERMINATION_FACTOR);
 
-    maxWaitTime = csConfig.getLong(
+    maxWaitTime = config.getLong(
         CapacitySchedulerConfiguration.PREEMPTION_WAIT_TIME_BEFORE_KILL,
         CapacitySchedulerConfiguration.DEFAULT_PREEMPTION_WAIT_TIME_BEFORE_KILL);
 
-    monitoringInterval = csConfig.getLong(
+    monitoringInterval = config.getLong(
         CapacitySchedulerConfiguration.PREEMPTION_MONITORING_INTERVAL,
         CapacitySchedulerConfiguration.DEFAULT_PREEMPTION_MONITORING_INTERVAL);
 
-    percentageClusterPreemptionAllowed = csConfig.getFloat(
+    percentageClusterPreemptionAllowed = config.getFloat(
         CapacitySchedulerConfiguration.TOTAL_PREEMPTION_PER_ROUND,
         CapacitySchedulerConfiguration.DEFAULT_TOTAL_PREEMPTION_PER_ROUND);
 
-    observeOnly = csConfig.getBoolean(
+    observeOnly = config.getBoolean(
         CapacitySchedulerConfiguration.PREEMPTION_OBSERVE_ONLY,
         CapacitySchedulerConfiguration.DEFAULT_PREEMPTION_OBSERVE_ONLY);
 
-    lazyPreempionEnabled = csConfig.getBoolean(
-        CapacitySchedulerConfiguration.LAZY_PREEMPTION_ENALBED,
+    lazyPreempionEnabled = config.getBoolean(
+        CapacitySchedulerConfiguration.LAZY_PREEMPTION_ENABLED,
         CapacitySchedulerConfiguration.DEFAULT_LAZY_PREEMPTION_ENABLED);
 
-    maxAllowableLimitForIntraQueuePreemption = csConfig.getFloat(
+    maxAllowableLimitForIntraQueuePreemption = config.getFloat(
         CapacitySchedulerConfiguration.
         INTRAQUEUE_PREEMPTION_MAX_ALLOWABLE_LIMIT,
         CapacitySchedulerConfiguration.
         DEFAULT_INTRAQUEUE_PREEMPTION_MAX_ALLOWABLE_LIMIT);
 
-    minimumThresholdForIntraQueuePreemption = csConfig.getFloat(
+    minimumThresholdForIntraQueuePreemption = config.getFloat(
         CapacitySchedulerConfiguration.
         INTRAQUEUE_PREEMPTION_MINIMUM_THRESHOLD,
         CapacitySchedulerConfiguration.
         DEFAULT_INTRAQUEUE_PREEMPTION_MINIMUM_THRESHOLD);
 
     intraQueuePreemptionOrderPolicy = IntraQueuePreemptionOrderPolicy
-        .valueOf(csConfig
+        .valueOf(config
             .get(
                 CapacitySchedulerConfiguration.INTRAQUEUE_PREEMPTION_ORDER_POLICY,
                 CapacitySchedulerConfiguration.DEFAULT_INTRAQUEUE_PREEMPTION_ORDER_POLICY)
             .toUpperCase());
 
-    rc = scheduler.getResourceCalculator();
-    nlm = scheduler.getRMContext().getNodeLabelManager();
+    candidatesSelectionPolicies = new ArrayList<>();
 
     // Do we need to specially consider reserved containers?
-    boolean selectCandidatesForResevedContainers = csConfig.getBoolean(
+    boolean selectCandidatesForResevedContainers = config.getBoolean(
         CapacitySchedulerConfiguration.
         PREEMPTION_SELECT_CANDIDATES_FOR_RESERVED_CONTAINERS,
         CapacitySchedulerConfiguration.
@@ -222,7 +232,7 @@ public class ProportionalCapacityPreemptionPolicy
           .add(new ReservedContainerCandidatesSelector(this));
     }
 
-    boolean additionalPreemptionBasedOnReservedResource = csConfig.getBoolean(
+    boolean additionalPreemptionBasedOnReservedResource = config.getBoolean(
         CapacitySchedulerConfiguration.ADDITIONAL_RESOURCE_BALANCE_BASED_ON_RESERVED_CONTAINERS,
         CapacitySchedulerConfiguration.DEFAULT_ADDITIONAL_RESOURCE_BALANCE_BASED_ON_RESERVED_CONTAINERS);
 
@@ -231,12 +241,37 @@ public class ProportionalCapacityPreemptionPolicy
         additionalPreemptionBasedOnReservedResource));
 
     // Do we need to specially consider intra queue
-    boolean isIntraQueuePreemptionEnabled = csConfig.getBoolean(
+    boolean isIntraQueuePreemptionEnabled = config.getBoolean(
         CapacitySchedulerConfiguration.INTRAQUEUE_PREEMPTION_ENABLED,
         CapacitySchedulerConfiguration.DEFAULT_INTRAQUEUE_PREEMPTION_ENABLED);
     if (isIntraQueuePreemptionEnabled) {
       candidatesSelectionPolicies.add(new IntraQueueCandidatesSelector(this));
     }
+
+    LOG.info("Capacity Scheduler configuration changed, updated preemption " +
+        "properties to:\n" +
+        "max_ignored_over_capacity = " + maxIgnoredOverCapacity + "\n" +
+        "natural_termination_factor = " + naturalTerminationFactor + "\n" +
+        "max_wait_before_kill = " + maxWaitTime + "\n" +
+        "monitoring_interval = " + monitoringInterval + "\n" +
+        "total_preemption_per_round = " + percentageClusterPreemptionAllowed +
+          "\n" +
+        "observe_only = " + observeOnly + "\n" +
+        "lazy-preemption-enabled = " + lazyPreempionEnabled + "\n" +
+        "intra-queue-preemption.enabled = " + isIntraQueuePreemptionEnabled +
+          "\n" +
+        "intra-queue-preemption.max-allowable-limit = " +
+          maxAllowableLimitForIntraQueuePreemption + "\n" +
+        "intra-queue-preemption.minimum-threshold = " +
+          minimumThresholdForIntraQueuePreemption + "\n" +
+        "intra-queue-preemption.preemption-order-policy = " +
+          intraQueuePreemptionOrderPolicy + "\n" +
+        "select_based_on_reserved_containers = " +
+          selectCandidatesForResevedContainers + "\n" +
+        "additional_res_balance_based_on_reserved_containers = " +
+          additionalPreemptionBasedOnReservedResource);
+
+    csConfig = config;
   }
 
   @Override
@@ -246,6 +281,8 @@ public class ProportionalCapacityPreemptionPolicy
 
   @Override
   public void editSchedule() {
+    updateConfigIfNeeded();
+
     long startTs = clock.getTime();
 
     CSQueue root = scheduler.getRootQueue();

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java

@@ -274,7 +274,8 @@ public class CapacitySchedulerConfiguration extends ReservationSchedulerConfigur
       "reservation-enforcement-window";
 
   @Private
-  public static final String LAZY_PREEMPTION_ENALBED = PREFIX + "lazy-preemption-enabled";
+  public static final String LAZY_PREEMPTION_ENABLED =
+      PREFIX + "lazy-preemption-enabled";
 
   @Private
   public static final boolean DEFAULT_LAZY_PREEMPTION_ENABLED = false;
@@ -1048,7 +1049,7 @@ public class CapacitySchedulerConfiguration extends ReservationSchedulerConfigur
   }
 
   public boolean getLazyPreemptionEnabled() {
-    return getBoolean(LAZY_PREEMPTION_ENALBED, DEFAULT_LAZY_PREEMPTION_ENABLED);
+    return getBoolean(LAZY_PREEMPTION_ENABLED, DEFAULT_LAZY_PREEMPTION_ENABLED);
   }
 
   private static final String PREEMPTION_CONFIG_PREFIX =
@@ -1089,7 +1090,7 @@ public class CapacitySchedulerConfiguration extends ReservationSchedulerConfigur
    * completions) it might prevent convergence to guaranteed capacity. */
   public static final String PREEMPTION_MAX_IGNORED_OVER_CAPACITY =
       PREEMPTION_CONFIG_PREFIX + "max_ignored_over_capacity";
-  public static final float DEFAULT_PREEMPTION_MAX_IGNORED_OVER_CAPACITY = 0.1f;
+  public static final double DEFAULT_PREEMPTION_MAX_IGNORED_OVER_CAPACITY = 0.1;
   /**
    * Given a computed preemption target, account for containers naturally
    * expiring and preempt only this percentage of the delta. This determines
@@ -1099,8 +1100,8 @@ public class CapacitySchedulerConfiguration extends ReservationSchedulerConfigur
    * #PREEMPTION_WAIT_TIME_BEFORE_KILL}, even absent natural termination. */
   public static final String PREEMPTION_NATURAL_TERMINATION_FACTOR =
       PREEMPTION_CONFIG_PREFIX + "natural_termination_factor";
-  public static final float DEFAULT_PREEMPTION_NATURAL_TERMINATION_FACTOR =
-      0.2f;
+  public static final double DEFAULT_PREEMPTION_NATURAL_TERMINATION_FACTOR =
+      0.2;
 
   /**
    * By default, reserved resource will be excluded while balancing capacities

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicy.java

@@ -183,9 +183,7 @@ public class TestProportionalCapacityPreemptionPolicy {
     appAlloc = 0;
   }
 
-  @Test
-  public void testIgnore() {
-    int[][] qData = new int[][]{
+  private static final int[][] Q_DATA_FOR_IGNORE = new int[][]{
       //  /   A   B   C
       { 100, 40, 40, 20 },  // abs
       { 100, 100, 100, 100 },  // maxCap
@@ -195,8 +193,12 @@ public class TestProportionalCapacityPreemptionPolicy {
       {   3,  1,  1,  1 },  // apps
       {  -1,  1,  1,  1 },  // req granularity
       {   3,  0,  0,  0 },  // subqueues
-    };
-    ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData);
+  };
+
+  @Test
+  public void testIgnore() {
+    ProportionalCapacityPreemptionPolicy policy =
+        buildPolicy(Q_DATA_FOR_IGNORE);
     policy.editSchedule();
     // don't correct imbalances without demand
     verify(mDisp, never()).handle(isA(ContainerPreemptEvent.class));
@@ -1000,6 +1002,36 @@ public class TestProportionalCapacityPreemptionPolicy {
     assertEquals(extraForQueueA,tempQueueAPartition.preemptableExtra.getMemorySize());
   }
 
+
+  @Test
+  public void testRefreshPreemptionProperties() throws Exception {
+    ProportionalCapacityPreemptionPolicy policy =
+        buildPolicy(Q_DATA_FOR_IGNORE);
+
+    assertEquals(
+        CapacitySchedulerConfiguration.DEFAULT_PREEMPTION_MONITORING_INTERVAL,
+        policy.getMonitoringInterval());
+    assertEquals(
+        CapacitySchedulerConfiguration.DEFAULT_PREEMPTION_OBSERVE_ONLY,
+        policy.isObserveOnly());
+
+    CapacitySchedulerConfiguration newConf =
+        new CapacitySchedulerConfiguration(conf);
+    long newMonitoringInterval = 5000;
+    boolean newObserveOnly = true;
+    newConf.setLong(
+        CapacitySchedulerConfiguration.PREEMPTION_MONITORING_INTERVAL,
+        newMonitoringInterval);
+    newConf.setBoolean(CapacitySchedulerConfiguration.PREEMPTION_OBSERVE_ONLY,
+        newObserveOnly);
+    when(mCS.getConfiguration()).thenReturn(newConf);
+
+    policy.editSchedule();
+
+    assertEquals(newMonitoringInterval, policy.getMonitoringInterval());
+    assertEquals(newObserveOnly, policy.isObserveOnly());
+  }
+
   static class IsPreemptionRequestFor
       extends ArgumentMatcher<ContainerPreemptEvent> {
     private final ApplicationAttemptId appAttId;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacitySchedulerLazyPreemption.java

@@ -65,7 +65,7 @@ public class TestCapacitySchedulerLazyPreemption
   @Before
   public void setUp() throws Exception {
     super.setUp();
-    conf.setBoolean(CapacitySchedulerConfiguration.LAZY_PREEMPTION_ENALBED,
+    conf.setBoolean(CapacitySchedulerConfiguration.LAZY_PREEMPTION_ENABLED,
         true);
   }