|
@@ -19,7 +19,6 @@
|
|
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
|
|
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
|
|
|
|
|
|
import java.io.IOException;
|
|
import java.io.IOException;
|
|
-import java.io.InputStream;
|
|
|
|
import java.util.ArrayList;
|
|
import java.util.ArrayList;
|
|
import java.util.Collection;
|
|
import java.util.Collection;
|
|
import java.util.EnumSet;
|
|
import java.util.EnumSet;
|
|
@@ -103,6 +102,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.Activi
|
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityDiagnosticConstant;
|
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityDiagnosticConstant;
|
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityState;
|
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityState;
|
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.AllocationState;
|
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.AllocationState;
|
|
|
|
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.conf.CSConfigurationProvider;
|
|
|
|
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.conf.FileBasedCSConfigurationProvider;
|
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.preemption.KillableContainer;
|
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.preemption.KillableContainer;
|
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.preemption.PreemptionManager;
|
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.preemption.PreemptionManager;
|
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.AssignmentInformation;
|
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.AssignmentInformation;
|
|
@@ -167,6 +168,8 @@ public class CapacityScheduler extends
|
|
|
|
|
|
private int maxAssignPerHeartbeat;
|
|
private int maxAssignPerHeartbeat;
|
|
|
|
|
|
|
|
+ private CSConfigurationProvider csConfProvider;
|
|
|
|
+
|
|
@Override
|
|
@Override
|
|
public void setConf(Configuration conf) {
|
|
public void setConf(Configuration conf) {
|
|
yarnConf = conf;
|
|
yarnConf = conf;
|
|
@@ -289,7 +292,18 @@ public class CapacityScheduler extends
|
|
IOException {
|
|
IOException {
|
|
try {
|
|
try {
|
|
writeLock.lock();
|
|
writeLock.lock();
|
|
- this.conf = loadCapacitySchedulerConfiguration(configuration);
|
|
|
|
|
|
+ String confProviderStr = configuration.get(
|
|
|
|
+ CapacitySchedulerConfiguration.CS_CONF_PROVIDER,
|
|
|
|
+ CapacitySchedulerConfiguration.DEFAULT_CS_CONF_PROVIDER);
|
|
|
|
+ if (confProviderStr.equals(
|
|
|
|
+ CapacitySchedulerConfiguration.FILE_CS_CONF_PROVIDER)) {
|
|
|
|
+ this.csConfProvider = new FileBasedCSConfigurationProvider(rmContext);
|
|
|
|
+ } else {
|
|
|
|
+ throw new IOException("Invalid CS configuration provider: " +
|
|
|
|
+ confProviderStr);
|
|
|
|
+ }
|
|
|
|
+ this.csConfProvider.init(configuration);
|
|
|
|
+ this.conf = this.csConfProvider.loadConfiguration(configuration);
|
|
validateConf(this.conf);
|
|
validateConf(this.conf);
|
|
this.minimumAllocation = super.getMinimumAllocation();
|
|
this.minimumAllocation = super.getMinimumAllocation();
|
|
initMaximumResourceCapability(super.getMaximumAllocation());
|
|
initMaximumResourceCapability(super.getMaximumAllocation());
|
|
@@ -399,7 +413,7 @@ public class CapacityScheduler extends
|
|
writeLock.lock();
|
|
writeLock.lock();
|
|
Configuration configuration = new Configuration(newConf);
|
|
Configuration configuration = new Configuration(newConf);
|
|
CapacitySchedulerConfiguration oldConf = this.conf;
|
|
CapacitySchedulerConfiguration oldConf = this.conf;
|
|
- this.conf = loadCapacitySchedulerConfiguration(configuration);
|
|
|
|
|
|
+ this.conf = csConfProvider.loadConfiguration(configuration);
|
|
validateConf(this.conf);
|
|
validateConf(this.conf);
|
|
try {
|
|
try {
|
|
LOG.info("Re-initializing queues...");
|
|
LOG.info("Re-initializing queues...");
|
|
@@ -1831,23 +1845,6 @@ public class CapacityScheduler extends
|
|
return true;
|
|
return true;
|
|
}
|
|
}
|
|
|
|
|
|
- private CapacitySchedulerConfiguration loadCapacitySchedulerConfiguration(
|
|
|
|
- Configuration configuration) throws IOException {
|
|
|
|
- try {
|
|
|
|
- InputStream CSInputStream =
|
|
|
|
- this.rmContext.getConfigurationProvider()
|
|
|
|
- .getConfigurationInputStream(configuration,
|
|
|
|
- YarnConfiguration.CS_CONFIGURATION_FILE);
|
|
|
|
- if (CSInputStream != null) {
|
|
|
|
- configuration.addResource(CSInputStream);
|
|
|
|
- return new CapacitySchedulerConfiguration(configuration, false);
|
|
|
|
- }
|
|
|
|
- return new CapacitySchedulerConfiguration(configuration, true);
|
|
|
|
- } catch (Exception e) {
|
|
|
|
- throw new IOException(e);
|
|
|
|
- }
|
|
|
|
- }
|
|
|
|
-
|
|
|
|
private String getDefaultReservationQueueName(String planQueueName) {
|
|
private String getDefaultReservationQueueName(String planQueueName) {
|
|
return planQueueName + ReservationConstants.DEFAULT_QUEUE_SUFFIX;
|
|
return planQueueName + ReservationConstants.DEFAULT_QUEUE_SUFFIX;
|
|
}
|
|
}
|