diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java index 6c0a8541223..57428444be3 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java @@ -402,12 +402,25 @@ public RefreshQueuesResponse refreshQueues(RefreshQueuesRequest request) @Private public void refreshQueues() throws IOException, YarnException { - rm.getRMContext().getScheduler().reinitialize(getConfig(), + // We use getConfig() before which gets a capacity-scheduler.xml reference + // when parsing it into CapacityScheduler#reinitialize, but we need to get + // properties from yarn-site.xml when we want to enable/disable preemption + Configuration conf = getConfiguration(new Configuration(false), + YarnConfiguration.YARN_SITE_CONFIGURATION_FILE); + // The reason we call Configuration#size() is because when getConfiguration + // been called, it invokes Configuration#addResouce, which invokes + // Configuration#reloadConfiguration which triggers the reload process in a + // lazy way, the properties will only be reload when it's needed rather than + // reload it right after getConfiguration been called. So here we call + // Configuration#size() to force the Configuration#getProps been called to + // reload all the properties. + conf.size(); + rm.getRMContext().getScheduler().reinitialize(conf, this.rm.getRMContext()); // refresh the reservation system ReservationSystem rSystem = rm.getRMContext().getReservationSystem(); if (rSystem != null) { - rSystem.reinitialize(getConfig(), rm.getRMContext()); + rSystem.reinitialize(conf, rm.getRMContext()); } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java index 727bc521740..da233cd3055 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java @@ -18,6 +18,7 @@ package org.apache.hadoop.yarn.server.resourcemanager; +import com.google.common.annotations.VisibleForTesting; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.curator.framework.AuthInfo; @@ -67,8 +68,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.metrics.SystemMetricsPublisher; import org.apache.hadoop.yarn.server.resourcemanager.metrics.TimelineServiceV1Publisher; import org.apache.hadoop.yarn.server.resourcemanager.metrics.TimelineServiceV2Publisher; -import org.apache.hadoop.yarn.server.resourcemanager.monitor.SchedulingEditPolicy; -import org.apache.hadoop.yarn.server.resourcemanager.monitor.SchedulingMonitor; import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMDelegatedNodeLabelsUpdater; import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; import org.apache.hadoop.yarn.server.resourcemanager.recovery.NullRMStateStore; @@ -113,8 +112,6 @@ import org.apache.zookeeper.server.auth.DigestAuthenticationProvider; import org.eclipse.jetty.webapp.WebAppContext; -import com.google.common.annotations.VisibleForTesting; - import java.io.IOException; import java.io.InputStream; import java.io.PrintStream; @@ -709,8 +706,6 @@ protected void serviceInit(Configuration configuration) throws Exception { } } - createSchedulerMonitors(); - masterService = createApplicationMasterService(); addService(masterService) ; rmContext.setApplicationMasterService(masterService); @@ -809,30 +804,6 @@ protected void serviceStop() throws Exception { } } - - protected void createSchedulerMonitors() { - if (conf.getBoolean(YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS, - YarnConfiguration.DEFAULT_RM_SCHEDULER_ENABLE_MONITORS)) { - LOG.info("Loading policy monitors"); - List policies = conf.getInstances( - YarnConfiguration.RM_SCHEDULER_MONITOR_POLICIES, - SchedulingEditPolicy.class); - if (policies.size() > 0) { - for (SchedulingEditPolicy policy : policies) { - LOG.info("LOADING SchedulingEditPolicy:" + policy.getPolicyName()); - // periodically check whether we need to take action to guarantee - // constraints - SchedulingMonitor mon = new SchedulingMonitor(rmContext, policy); - addService(mon); - } - } else { - LOG.warn("Policy monitors configured (" + - YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS + - ") but none specified (" + - YarnConfiguration.RM_SCHEDULER_MONITOR_POLICIES + ")"); - } - } - } } @Private diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/SchedulingMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/SchedulingMonitor.java index 2a741ed83cb..472cab40183 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/SchedulingMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/SchedulingMonitor.java @@ -58,6 +58,7 @@ public synchronized SchedulingEditPolicy getSchedulingEditPolicy() { } public void serviceInit(Configuration conf) throws Exception { + LOG.info("Initializing SchedulingMonitor=" + getName()); scheduleEditPolicy.init(conf, rmContext, rmContext.getScheduler()); this.monitorInterval = scheduleEditPolicy.getMonitoringInterval(); super.serviceInit(conf); @@ -65,6 +66,7 @@ public void serviceInit(Configuration conf) throws Exception { @Override public void serviceStart() throws Exception { + LOG.info("Starting SchedulingMonitor=" + getName()); assert !stopped : "starting when already stopped"; ses = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() { public Thread newThread(Runnable r) { diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/SchedulingMonitorManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/SchedulingMonitorManager.java new file mode 100644 index 00000000000..b603b298e49 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/SchedulingMonitorManager.java @@ -0,0 +1,171 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.yarn.server.resourcemanager.monitor; + +import com.google.common.collect.Sets; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.util.ReflectionUtils; +import org.apache.hadoop.yarn.conf.YarnConfiguration; +import org.apache.hadoop.yarn.exceptions.YarnException; +import org.apache.hadoop.yarn.server.resourcemanager.RMContext; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +/** + * Manages scheduling monitors + */ +public class SchedulingMonitorManager { + private static final Log LOG = LogFactory.getLog( + SchedulingMonitorManager.class); + + private Map runningSchedulingMonitors = new HashMap<>(); + private RMContext rmContext; + + private void updateSchedulingMonitors(Configuration conf, + boolean startImmediately) throws YarnException { + boolean monitorsEnabled = conf.getBoolean( + YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS, + YarnConfiguration.DEFAULT_RM_SCHEDULER_ENABLE_MONITORS); + + if (!monitorsEnabled) { + if (!runningSchedulingMonitors.isEmpty()) { + // If monitors disabled while we have some running monitors, we should + // stop them. + LOG.info("Scheduling Monitor disabled, stopping all services"); + stopAndRemoveAll(); + } + + return; + } + + // When monitor is enabled, loading policies + String[] configuredPolicies = conf.getStrings( + YarnConfiguration.RM_SCHEDULER_MONITOR_POLICIES); + if (configuredPolicies == null || configuredPolicies.length == 0) { + return; + } + + Set configurePoliciesSet = new HashSet<>(); + for (String s : configuredPolicies) { + configurePoliciesSet.add(s); + } + + // Add new monitor when needed + for (String s : configurePoliciesSet) { + if (!runningSchedulingMonitors.containsKey(s)) { + Class policyClass; + try { + policyClass = Class.forName(s); + } catch (ClassNotFoundException e) { + String message = "Failed to find class of specified policy=" + s; + LOG.warn(message); + throw new YarnException(message); + } + + if (SchedulingEditPolicy.class.isAssignableFrom(policyClass)) { + SchedulingEditPolicy policyInstance = + (SchedulingEditPolicy) ReflectionUtils.newInstance(policyClass, + null); + SchedulingMonitor mon = new SchedulingMonitor(rmContext, + policyInstance); + mon.init(conf); + if (startImmediately) { + mon.start(); + } + runningSchedulingMonitors.put(s, mon); + } else { + String message = + "Specified policy=" + s + " is not a SchedulingEditPolicy class."; + LOG.warn(message); + throw new YarnException(message); + } + } + } + + // Stop monitor when needed. + Set disabledPolicies = Sets.difference( + runningSchedulingMonitors.keySet(), configurePoliciesSet); + for (String disabledPolicy : disabledPolicies) { + LOG.info("SchedulingEditPolicy=" + disabledPolicy + + " removed, stopping it now ..."); + silentlyStopSchedulingMonitor(disabledPolicy); + runningSchedulingMonitors.remove(disabledPolicy); + } + } + + public synchronized void initialize(RMContext rmContext, + Configuration configuration) throws YarnException { + this.rmContext = rmContext; + stopAndRemoveAll(); + + updateSchedulingMonitors(configuration, false); + } + + public synchronized void reinitialize(RMContext rmContext, + Configuration configuration) throws YarnException { + this.rmContext = rmContext; + + updateSchedulingMonitors(configuration, true); + } + + public synchronized void startAll() { + for (SchedulingMonitor schedulingMonitor : runningSchedulingMonitors + .values()) { + schedulingMonitor.start(); + } + } + + private void silentlyStopSchedulingMonitor(String name) { + SchedulingMonitor mon = runningSchedulingMonitors.get(name); + try { + mon.stop(); + LOG.info("Sucessfully stopped monitor=" + mon.getName()); + } catch (Exception e) { + LOG.warn("Exception while stopping monitor=" + mon.getName(), e); + } + } + + private void stopAndRemoveAll() { + if (!runningSchedulingMonitors.isEmpty()) { + for (String schedulingMonitorName : runningSchedulingMonitors + .keySet()) { + silentlyStopSchedulingMonitor(schedulingMonitorName); + } + runningSchedulingMonitors.clear(); + } + } + + public boolean isRSMEmpty() { + return runningSchedulingMonitors.isEmpty(); + } + + public boolean isSameConfiguredPolicies(Set configurePoliciesSet) { + return configurePoliciesSet.equals(runningSchedulingMonitors.keySet()); + } + + public synchronized void stop() throws YarnException { + stopAndRemoveAll(); + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java index 7308fd8b0ff..88a0d9834e8 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java @@ -68,6 +68,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.RMCriticalThreadUncaughtExceptionHandler; import org.apache.hadoop.yarn.server.resourcemanager.RMServerUtils; import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager; +import org.apache.hadoop.yarn.server.resourcemanager.monitor.SchedulingMonitorManager; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType; @@ -168,6 +169,8 @@ // the NM in the next heartbeat. private boolean autoUpdateContainers = false; + protected SchedulingMonitorManager schedulingMonitorManager; + /** * Construct the service. * @@ -216,6 +219,8 @@ protected void serviceStart() throws Exception { if (updateThread != null) { updateThread.start(); } + + schedulingMonitorManager.startAll(); super.serviceStart(); } @@ -225,6 +230,8 @@ protected void serviceStop() throws Exception { updateThread.interrupt(); updateThread.join(THREAD_JOIN_TIMEOUT_MS); } + + schedulingMonitorManager.stop(); super.serviceStop(); } @@ -233,6 +240,9 @@ public ClusterNodeTracker getNodeTracker() { return nodeTracker; } + @VisibleForTesting + public SchedulingMonitorManager getSchedulingMonitorManager() {return schedulingMonitorManager;} + /* * YARN-3136 removed synchronized lock for this method for performance * purposes @@ -1415,4 +1425,15 @@ protected void triggerUpdate() { updateThreadMonitor.notify(); } } + + @Override + public void reinitialize(Configuration conf, RMContext rmContext) + throws IOException { + try { + LOG.info("Reinitializing SchedulingMonitorManager ..."); + schedulingMonitorManager.reinitialize(rmContext, conf); + } catch (YarnException e) { + throw new IOException(e); + } + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java index 5e172b8e743..d106c459273 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java @@ -63,6 +63,7 @@ import org.apache.hadoop.yarn.exceptions.YarnRuntimeException; import org.apache.hadoop.yarn.proto.YarnServiceProtos.SchedulerResourceTypes; import org.apache.hadoop.yarn.server.resourcemanager.RMContext; +import org.apache.hadoop.yarn.server.resourcemanager.monitor.SchedulingMonitorManager; import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; import org.apache.hadoop.yarn.server.resourcemanager.placement.PlacementFactory; import org.apache.hadoop.yarn.server.resourcemanager.placement.PlacementRule; @@ -389,6 +390,10 @@ public void serviceInit(Configuration conf) throws Exception { Configuration configuration = new Configuration(conf); super.serviceInit(conf); initScheduler(configuration); + + // Initialize SchedulingMonitorManager + schedulingMonitorManager = new SchedulingMonitorManager(); + schedulingMonitorManager.initialize(rmContext, conf); } @Override @@ -443,6 +448,8 @@ public void reinitialize(Configuration newConf, RMContext rmContext) // Setup how many containers we can allocate for each round offswitchPerHeartbeatLimit = this.conf.getOffSwitchPerHeartbeatLimit(); + + super.reinitialize(newConf, rmContext); } finally { writeLock.unlock(); } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java index 7f1b91e32af..a9c5fbb8082 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java @@ -52,6 +52,7 @@ import org.apache.hadoop.yarn.server.api.protocolrecords.NMContainerStatus; import org.apache.hadoop.yarn.server.resourcemanager.RMContext; import org.apache.hadoop.yarn.server.resourcemanager.RMCriticalThreadUncaughtExceptionHandler; +import org.apache.hadoop.yarn.server.resourcemanager.monitor.SchedulingMonitorManager; import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState; import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationConstants; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; @@ -1372,6 +1373,10 @@ private void startSchedulerThreads() { public void serviceInit(Configuration conf) throws Exception { initScheduler(conf); super.serviceInit(conf); + + // Initialize SchedulingMonitorManager + schedulingMonitorManager = new SchedulingMonitorManager(); + schedulingMonitorManager.initialize(rmContext, conf); } @Override @@ -1409,6 +1414,7 @@ public void reinitialize(Configuration conf, RMContext rmContext) throws IOException { try { allocsLoader.reloadAllocations(); + super.reinitialize(conf, rmContext); } catch (Exception e) { LOG.error("Failed to reload allocations file", e); } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java index 32889128369..0e1f308cc9c 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java @@ -312,6 +312,7 @@ public synchronized void setRMContext(RMContext rmContext) { reinitialize(Configuration conf, RMContext rmContext) throws IOException { setConf(conf); + super.reinitialize(conf, rmContext); } @Override diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/TestSchedulingMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/TestSchedulingMonitor.java index c38236d0abf..84126c72877 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/TestSchedulingMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/TestSchedulingMonitor.java @@ -23,8 +23,15 @@ import org.apache.hadoop.yarn.server.resourcemanager.MockRM; import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager; import org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity.ProportionalCapacityPreemptionPolicy; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration; import org.junit.Test; +import java.util.HashSet; +import java.util.Set; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.verify; @@ -51,4 +58,38 @@ public void testRMStarts() throws Exception { monitor.close(); rm.close(); } + + @Test(timeout = 10000) + public void testRMUpdateSchedulingEditPolicy() throws Exception { + CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration(); + conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class, + ResourceScheduler.class); + conf.setBoolean(YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS, true); + MockRM rm = new MockRM(conf); + rm.start(); + CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler(); + SchedulingMonitorManager smm = cs.getSchedulingMonitorManager(); + + // runningSchedulingMonitors should not be empty when initialize RM + // scheduler monitor + cs.reinitialize(conf, rm.getRMContext()); + assertFalse(smm.isRSMEmpty()); + + // make sure runningSchedulingPolicies contains all the configured policy + // in YARNConfiguration + String[] configuredPolicies = conf.getStrings( + YarnConfiguration.RM_SCHEDULER_MONITOR_POLICIES); + Set configurePoliciesSet = new HashSet<>(); + for (String s : configuredPolicies) { + configurePoliciesSet.add(s); + } + assertTrue(smm.isSameConfiguredPolicies(configurePoliciesSet)); + + // disable RM scheduler monitor + conf.setBoolean( + YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS, + YarnConfiguration.DEFAULT_RM_SCHEDULER_ENABLE_MONITORS); + cs.reinitialize(conf, rm.getRMContext()); + assertTrue(smm.isRSMEmpty()); + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java index 1dea4eea75f..334708877d9 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java @@ -4333,6 +4333,7 @@ public void testCSQueueBlocked() throws Exception { ResourceScheduler.class); MockRM rm = new MockRM(conf); rm.start(); + rm.getAdminService().refreshQueues(); CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler(); LeafQueue q = (LeafQueue) cs.getQueue("a");