Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

YARN-10917. Investigate and simplify CapacitySchedulerConfigValidator#validateQueueHierarchy (#3403) 

* YARN-10917. Investigate and simplify CapacitySchedulerConfigValidator#validateQueueHierarchy.

Co-authored-by: Tamas Domok <tdomok@cloudera.com>
This commit is contained in:
Tamas Domok 2021-09-14 17:54:25 +02:00 committed by GitHub
parent 63c892278f
commit 783d94f5cd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 91 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

158
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfigValidator.java
View File

@ -106,12 +106,8 @@ public final class CapacitySchedulerConfigValidator {
} }
} }
private static boolean isDynamicQueue(CSQueue csQueue) {
return ((AbstractCSQueue)csQueue).isDynamicQueue();
}
/** /**
* Ensure all existing queues are present. Queues cannot be deleted if its not * Ensure all existing queues are present. Queues cannot be deleted if it's not
* in Stopped state, Queue's cannot be moved from one hierarchy to other also. * in Stopped state, Queue's cannot be moved from one hierarchy to other also.
* Previous child queue could be converted into parent queue if it is in * Previous child queue could be converted into parent queue if it is in
* STOPPED state. * STOPPED state.
@ -125,78 +121,106 @@ public final class CapacitySchedulerConfigValidator {
CapacitySchedulerConfiguration newConf) throws IOException { CapacitySchedulerConfiguration newConf) throws IOException {
// check that all static queues are included in the newQueues list // check that all static queues are included in the newQueues list
for (CSQueue oldQueue : queues.getQueues()) { for (CSQueue oldQueue : queues.getQueues()) {
if (!(AbstractAutoCreatedLeafQueue.class.isAssignableFrom( if (AbstractAutoCreatedLeafQueue.class.isAssignableFrom(oldQueue.getClass())) {
oldQueue.getClass()))) { continue;
String queuePath = oldQueue.getQueuePath(); }
CSQueue newQueue = newQueues.get(queuePath);
String configPrefix = newConf.getQueuePrefix( final String queuePath = oldQueue.getQueuePath();
oldQueue.getQueuePath()); final String configPrefix = CapacitySchedulerConfiguration.getQueuePrefix(
String state = newConf.get(configPrefix + "state"); oldQueue.getQueuePath());
QueueState newQueueState = null; final QueueState newQueueState = createQueueState(newConf.get(configPrefix + "state"),
if (state != null) { queuePath);
try { final CSQueue newQueue = newQueues.get(queuePath);
newQueueState = QueueState.valueOf(state);
} catch (Exception ex) { if (null == newQueue) {
LOG.warn("Not a valid queue state for queue " // old queue doesn't exist in the new XML
+ oldQueue.getQueuePath()); if (isEitherQueueStopped(oldQueue.getState(), newQueueState)) {
LOG.info("Deleting Queue {}, as it is not present in the modified capacity " +
"configuration xml", queuePath);
} else {
if (!isDynamicQueue(oldQueue)) {
throw new IOException(oldQueue.getQueuePath() + " cannot be"
+ " deleted from the capacity scheduler configuration, as the"
+ " queue is not yet in stopped state. Current State : "
+ oldQueue.getState());
} }
} }
if (null == newQueue) { } else {
// old queue doesn't exist in the new XML validateSameQueuePath(oldQueue, newQueue);
if (oldQueue.getState() == QueueState.STOPPED || validateParentQueueConversion(oldQueue, newQueue);
newQueueState == QueueState.STOPPED) { validateLeafQueueConversion(oldQueue, newQueue);
LOG.info("Deleting Queue " + queuePath + ", as it is not" }
+ " present in the modified capacity configuration xml"); }
} else { }
if (!isDynamicQueue(oldQueue)) {
throw new IOException(oldQueue.getQueuePath() + " cannot be" private static void validateSameQueuePath(CSQueue oldQueue, CSQueue newQueue) throws IOException {
+ " deleted from the capacity scheduler configuration, as the" if (!oldQueue.getQueuePath().equals(newQueue.getQueuePath())) {
+ " queue is not yet in stopped state. Current State : " // Queues cannot be moved from one hierarchy to another
+ oldQueue.getState()); throw new IOException(
} oldQueue.getQueuePath() + " is moved from:" + oldQueue.getQueuePath() + " to:"
}
} else if (!oldQueue.getQueuePath().equals(newQueue.getQueuePath())) {
//Queue's cannot be moved from one hierarchy to other
throw new IOException(
queuePath + " is moved from:" + oldQueue.getQueuePath() + " to:"
+ newQueue.getQueuePath() + newQueue.getQueuePath()
+ " after refresh, which is not allowed."); + " after refresh, which is not allowed.");
} else if (oldQueue instanceof ParentQueue }
&& !(oldQueue instanceof ManagedParentQueue) }
&& newQueue instanceof ManagedParentQueue) {
throw new IOException( private static void validateParentQueueConversion(CSQueue oldQueue,
CSQueue newQueue) throws IOException {
if (oldQueue instanceof ParentQueue) {
if (!(oldQueue instanceof ManagedParentQueue) && newQueue instanceof ManagedParentQueue) {
throw new IOException(
"Can not convert parent queue: " + oldQueue.getQueuePath() "Can not convert parent queue: " + oldQueue.getQueuePath()
+ " to auto create enabled parent queue since " + " to auto create enabled parent queue since "
+ "it could have other pre-configured queues which is not " + "it could have other pre-configured queues which is not "
+ "supported"); + "supported");
} else if (oldQueue instanceof ManagedParentQueue }
&& !(newQueue instanceof ManagedParentQueue)) {
throw new IOException( if (oldQueue instanceof ManagedParentQueue
&& !(newQueue instanceof ManagedParentQueue)) {
throw new IOException(
"Cannot convert auto create enabled parent queue: " "Cannot convert auto create enabled parent queue: "
+ oldQueue.getQueuePath() + " to leaf queue. Please check " + oldQueue.getQueuePath() + " to leaf queue. Please check "
+ " parent queue's configuration " + " parent queue's configuration "
+ CapacitySchedulerConfiguration + CapacitySchedulerConfiguration.AUTO_CREATE_CHILD_QUEUE_ENABLED
.AUTO_CREATE_CHILD_QUEUE_ENABLED + " is set to true");
+ " is set to true"); }
} else if (oldQueue instanceof LeafQueue
&& newQueue instanceof ParentQueue) { if (newQueue instanceof LeafQueue) {
if (oldQueue.getState() == QueueState.STOPPED || LOG.info("Converting the parent queue: {} to leaf queue.", oldQueue.getQueuePath());
newQueueState == QueueState.STOPPED) {
LOG.info("Converting the leaf queue: " + oldQueue.getQueuePath()
+ " to parent queue.");
} else{
throw new IOException(
"Can not convert the leaf queue: " + oldQueue.getQueuePath()
+ " to parent queue since "
+ "it is not yet in stopped state. Current State : "
+ oldQueue.getState());
}
} else if (oldQueue instanceof ParentQueue
&& newQueue instanceof LeafQueue) {
LOG.info("Converting the parent queue: " + oldQueue.getQueuePath()
+ " to leaf queue.");
}
} }
} }
} }
private static void validateLeafQueueConversion(CSQueue oldQueue,
CSQueue newQueue) throws IOException {
if (oldQueue instanceof LeafQueue && newQueue instanceof ParentQueue) {
if (isEitherQueueStopped(oldQueue.getState(), newQueue.getState())) {
LOG.info("Converting the leaf queue: {} to parent queue.", oldQueue.getQueuePath());
} else {
throw new IOException(
"Can not convert the leaf queue: " + oldQueue.getQueuePath()
+ " to parent queue since "
+ "it is not yet in stopped state. Current State : "
+ oldQueue.getState());
}
}
}
private static QueueState createQueueState(String state, String queuePath) {
if (state != null) {
try {
return QueueState.valueOf(state);
} catch (Exception ex) {
LOG.warn("Not a valid queue state for queue: {}, state: {}", queuePath, state);
}
}
return null;
}
private static boolean isDynamicQueue(CSQueue csQueue) {
return ((AbstractCSQueue)csQueue).isDynamicQueue();
}
private static boolean isEitherQueueStopped(QueueState a, QueueState b) {
return a == QueueState.STOPPED || b == QueueState.STOPPED;
}
} }