You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by sr...@apache.org on 2019/02/21 10:35:59 UTC
[storm] 01/03: STORM-3335 allow timing out when scheduling a
topology
This is an automated email from the ASF dual-hosted git repository.
srdo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
commit 435e57b09ec8d00d2e630f710f0f6db9f0746842
Author: Aaron Gresch <ag...@yahoo-inc.com>
AuthorDate: Fri Feb 8 15:53:53 2019 -0600
STORM-3335 allow timing out when scheduling a topology
---
.../src/jvm/org/apache/storm/task/IBolt.java | 2 +-
.../storm/scheduler/multitenant_scheduler_test.clj | 18 +-
.../main/java/org/apache/storm/DaemonConfig.java | 7 +
.../org/apache/storm/daemon/nimbus/Nimbus.java | 1 +
.../org/apache/storm/scheduler/IScheduler.java | 6 +
.../scheduler/blacklist/BlacklistScheduler.java | 6 +
.../scheduler/resource/ResourceAwareScheduler.java | 36 ++-
.../scheduling/BaseResourceAwareStrategy.java | 4 +-
.../scheduling/ConstraintSolverStrategy.java | 4 +
.../scheduling/DefaultResourceAwareStrategy.java | 8 +-
.../scheduling/GenericResourceAwareStrategy.java | 8 +-
.../resource/strategies/scheduling/IStrategy.java | 5 +-
.../blacklist/TestBlacklistScheduler.java | 75 +++---
.../resource/TestResourceAwareScheduler.java | 276 ++++++++++++++-------
.../eviction/TestDefaultEvictionStrategy.java | 61 +++--
.../TestFIFOSchedulingPriorityStrategy.java | 44 ++--
.../scheduling/TestConstraintSolverStrategy.java | 20 +-
.../TestDefaultResourceAwareStrategy.java | 23 +-
.../TestGenericResourceAwareStrategy.java | 31 ++-
19 files changed, 435 insertions(+), 200 deletions(-)
diff --git a/storm-client/src/jvm/org/apache/storm/task/IBolt.java b/storm-client/src/jvm/org/apache/storm/task/IBolt.java
index 1705e7a..21f1b8d 100644
--- a/storm-client/src/jvm/org/apache/storm/task/IBolt.java
+++ b/storm-client/src/jvm/org/apache/storm/task/IBolt.java
@@ -63,7 +63,7 @@ public interface IBolt extends Serializable {
void execute(Tuple input);
/**
- * Called when an IBolt is going to be shutdown. There is no guarentee that cleanup will be called, because the supervisor kill -9's
+ * Called when an IBolt is going to be shutdown. There is no guarantee that cleanup will be called, because the supervisor kill -9's
* worker processes on the cluster.
*
* The one context where cleanup is guaranteed to be called is when a topology is killed when running Storm in local mode.
diff --git a/storm-core/test/clj/org/apache/storm/scheduler/multitenant_scheduler_test.clj b/storm-core/test/clj/org/apache/storm/scheduler/multitenant_scheduler_test.clj
index 9e170b9..a9e843c 100644
--- a/storm-core/test/clj/org/apache/storm/scheduler/multitenant_scheduler_test.clj
+++ b/storm-core/test/clj/org/apache/storm/scheduler/multitenant_scheduler_test.clj
@@ -675,6 +675,7 @@
(.assign (.get node-map "super0") "topology1" (list (ed 1)) cluster)
(.assign (.get node-map "super1") "topology2" (list (ed 5)) cluster)
(.prepare scheduler conf)
+ (try
(.schedule scheduler topologies cluster)
(let [assignment (.getAssignmentById cluster "topology1")
assigned-slots (.getSlots assignment)
@@ -687,6 +688,7 @@
(is (= "Fully Scheduled" (.get (.getStatusMap cluster) "topology1")))
(is (= "Scheduled Isolated on 2 Nodes" (.get (.getStatusMap cluster) "topology2")))
(is (= "Scheduled Isolated on 5 Nodes" (.get (.getStatusMap cluster) "topology3")))
+ (finally (.cleanup scheduler)))
))
(deftest test-force-free-slot-in-bad-state
@@ -712,6 +714,7 @@
scheduler (MultitenantScheduler.)]
(.assign (.get node-map "super0") "topology1" (list (ed 1)) cluster)
(.prepare scheduler conf)
+ (try
(.schedule scheduler topologies cluster)
(let [assignment (.getAssignmentById cluster "topology1")
assigned-slots (.getSlots assignment)
@@ -722,6 +725,7 @@
(is (= 1 (.size (into #{} (for [slot assigned-slots] (.getNodeId slot))))))
)
(is (= "Fully Scheduled" (.get (.getStatusMap cluster) "topology1")))
+ (finally (.cleanup scheduler)))
))
(deftest test-multitenant-scheduler-bad-starting-state
@@ -752,6 +756,7 @@
conf {MULTITENANT-SCHEDULER-USER-POOLS {"userA" 2 "userB" 1}}
scheduler (MultitenantScheduler.)]
(.prepare scheduler conf)
+ (try
(.schedule scheduler topologies cluster)
(let [assignment (.getAssignmentById cluster "topology1")
assigned-slots (.getSlots assignment)
@@ -759,7 +764,8 @@
(is (= 1 (.size assigned-slots))))
(is (= "Fully Scheduled" (.get (.getStatusMap cluster) "topology1")))
(is (= "Scheduled Isolated on 2 Nodes" (.get (.getStatusMap cluster) "topology2")))
- (is (= "Scheduled Isolated on 1 Nodes" (.get (.getStatusMap cluster) "topology3"))))))
+ (is (= "Scheduled Isolated on 1 Nodes" (.get (.getStatusMap cluster) "topology3")))
+ (finally (.cleanup scheduler))))))
(deftest test-existing-assignment-slot-not-found-in-supervisor
(testing "Scheduler should handle discrepancy when a live supervisor heartbeat does not report slot,
@@ -779,12 +785,14 @@
conf {}
scheduler (MultitenantScheduler.)]
(.prepare scheduler conf)
+ (try
(.schedule scheduler topologies cluster)
(let [assignment (.getAssignmentById cluster "topology1")
assigned-slots (.getSlots assignment)
executors (.getExecutors assignment)]
(is (= 1 (.size assigned-slots))))
- (is (= "Fully Scheduled" (.get (.getStatusMap cluster) "topology1"))))))
+ (is (= "Fully Scheduled" (.get (.getStatusMap cluster) "topology1")))
+ (finally(.cleanup scheduler))))))
(deftest test-existing-assignment-slot-on-dead-supervisor
(testing "Dead supervisor could have slot with duplicate assignments or slot never reported by supervisor"
@@ -817,6 +825,7 @@
conf {}
scheduler (MultitenantScheduler.)]
(.prepare scheduler conf)
+ (try
(.schedule scheduler topologies cluster)
(let [assignment (.getAssignmentById cluster "topology1")
assigned-slots (.getSlots assignment)
@@ -829,7 +838,8 @@
executors (.getExecutors assignment)]
(is (= 2 (.size assigned-slots)))
(is (= 2 (.size executors))))
- (is (= "Fully Scheduled" (.get (.getStatusMap cluster) "topology2"))))))
+ (is (= "Fully Scheduled" (.get (.getStatusMap cluster) "topology2")))
+ (finally (.cleanup scheduler))))))
(deftest test-isolated-pool-scheduling-with-nodes-with-different-number-of-slots
@@ -855,5 +865,7 @@
conf {MULTITENANT-SCHEDULER-USER-POOLS {"userA" 2}}
scheduler (MultitenantScheduler.)]
(.prepare scheduler conf)
+ (try
(.schedule scheduler topologies cluster)
+ (finally (.cleanup scheduler)))
(is (= "Scheduled Isolated on 2 Nodes" (.get (.getStatusMap cluster) "topology1")))))
diff --git a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
index 4b51ef7..ae013de 100644
--- a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
+++ b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
@@ -96,6 +96,13 @@ public class DaemonConfig implements Validated {
public static final String STORM_SCHEDULER = "storm.scheduler";
/**
+ * Max time to attempt to schedule one topology. The default is 60 seconds
+ */
+ @isInteger
+ @isPositiveNumber
+ public static final String SCHEDULING_TIMEOUT_SECONDS_PER_TOPOLOGY = "scheduling.timeout.seconds.per.topology";
+
+ /**
* The number of seconds that the blacklist scheduler will concern of bad slots or supervisors.
*/
@isPositiveNumber
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
index a9e0484..ecad5c9 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
@@ -4613,6 +4613,7 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
blobDownloaders.cleanup();
blobUploaders.cleanup();
blobListers.cleanup();
+ scheduler.cleanup();
blobStore.shutdown();
leaderElector.close();
assignmentsDistributer.close();
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/IScheduler.java b/storm-server/src/main/java/org/apache/storm/scheduler/IScheduler.java
index 18e9afd..f2c0d26 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/IScheduler.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/IScheduler.java
@@ -38,4 +38,10 @@ public interface IScheduler {
* @return The scheduler's configuration.
*/
Map config();
+
+ /**
+ * called once when the system is shutting down, should be idempotent.
+ */
+ default void cleanup() {
+ }
}
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/blacklist/BlacklistScheduler.java b/storm-server/src/main/java/org/apache/storm/scheduler/blacklist/BlacklistScheduler.java
index 1e5b04d..109cd27 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/blacklist/BlacklistScheduler.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/blacklist/BlacklistScheduler.java
@@ -95,6 +95,12 @@ public class BlacklistScheduler implements IScheduler {
}
@Override
+ public void cleanup() {
+ LOG.info("Cleanup black list scheduler");
+ underlyingScheduler.cleanup();
+ }
+
+ @Override
public void schedule(Topologies topologies, Cluster cluster) {
LOG.debug("running Black List scheduler");
LOG.debug("AssignableSlots: {}", cluster.getAssignableSlots());
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java
index 3ee693a..5de265c 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java
@@ -17,6 +17,11 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import org.apache.storm.Config;
import org.apache.storm.DaemonConfig;
@@ -45,6 +50,8 @@ public class ResourceAwareScheduler implements IScheduler {
private ISchedulingPriorityStrategy schedulingPriorityStrategy;
private IConfigLoader configLoader;
private int maxSchedulingAttempts;
+ private int schedulingTimeoutSeconds;
+ private ExecutorService backgroundScheduling;
private static void markFailedTopology(User u, Cluster c, TopologyDetails td, String message) {
markFailedTopology(u, c, td, message, null);
@@ -78,6 +85,15 @@ public class ResourceAwareScheduler implements IScheduler {
configLoader = ConfigLoaderFactoryService.createConfigLoader(conf);
maxSchedulingAttempts = ObjectReader.getInt(
conf.get(DaemonConfig.RESOURCE_AWARE_SCHEDULER_MAX_TOPOLOGY_SCHEDULING_ATTEMPTS), 5);
+ schedulingTimeoutSeconds = ObjectReader.getInt(
+ conf.get(DaemonConfig.SCHEDULING_TIMEOUT_SECONDS_PER_TOPOLOGY), 60);
+ backgroundScheduling = Executors.newFixedThreadPool(1);
+ }
+
+ @Override
+ public void cleanup() {
+ LOG.info("Cleanup ResourceAwareScheduler scheduler");
+ backgroundScheduling.shutdown();
}
@Override
@@ -113,10 +129,10 @@ public class ResourceAwareScheduler implements IScheduler {
try {
String strategy = (String) td.getConf().get(Config.TOPOLOGY_SCHEDULER_STRATEGY);
if (strategy.startsWith("backtype.storm")) {
- // Storm supports to launch workers of older version.
+ // Storm support to launch workers of older version.
// If the config of TOPOLOGY_SCHEDULER_STRATEGY comes from the older version, replace the package name.
strategy = strategy.replace("backtype.storm", "org.apache.storm");
- LOG.debug("Replace backtype.storm with org.apache.storm for Config.TOPOLOGY_SCHEDULER_STRATEGY");
+ LOG.debug("Replaced backtype.storm with org.apache.storm for Config.TOPOLOGY_SCHEDULER_STRATEGY");
}
rasStrategy = ReflectionUtils.newSchedulerStrategyInstance(strategy, conf);
rasStrategy.prepare(conf);
@@ -136,10 +152,24 @@ public class ResourceAwareScheduler implements IScheduler {
return;
}
+ final IStrategy finalRasStrategy = rasStrategy;
for (int i = 0; i < maxSchedulingAttempts; i++) {
SingleTopologyCluster toSchedule = new SingleTopologyCluster(workingState, td.getId());
try {
- SchedulingResult result = rasStrategy.schedule(toSchedule, td);
+ SchedulingResult result = null;
+ Future<SchedulingResult> schedulingFuture = backgroundScheduling.submit(
+ () -> finalRasStrategy.schedule(toSchedule, td)
+ );
+ try {
+ result = schedulingFuture.get(schedulingTimeoutSeconds, TimeUnit.SECONDS);
+ } catch (TimeoutException te) {
+ markFailedTopology(topologySubmitter, cluster, td, "Scheduling took too long for "
+ + td.getId() + " using strategy " + rasStrategy.getClass().getName() + " timeout after "
+ + schedulingTimeoutSeconds + " seconds using config "
+ + DaemonConfig.SCHEDULING_TIMEOUT_SECONDS_PER_TOPOLOGY + ".");
+ schedulingFuture.cancel(true);
+ return;
+ }
LOG.debug("scheduling result: {}", result);
if (result == null) {
markFailedTopology(topologySubmitter, cluster, td, "Internal scheduler error");
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/BaseResourceAwareStrategy.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/BaseResourceAwareStrategy.java
index b5da8db..15ddbf6 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/BaseResourceAwareStrategy.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/BaseResourceAwareStrategy.java
@@ -631,7 +631,7 @@ public abstract class BaseResourceAwareStrategy implements IStrategy {
/**
* a class to contain individual object resources as well as cumulative stats.
*/
- static class AllResources {
+ protected static class AllResources {
List<ObjectResources> objectResources = new LinkedList<>();
final NormalizedResourceOffer availableResourcesOverall;
final NormalizedResourceOffer totalResourcesOverall;
@@ -667,7 +667,7 @@ public abstract class BaseResourceAwareStrategy implements IStrategy {
/**
* class to keep track of resources on a rack or node.
*/
- static class ObjectResources {
+ protected static class ObjectResources {
public final String id;
public NormalizedResourceOffer availableResources;
public NormalizedResourceOffer totalResources;
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
index 9c56773..64d5acb 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
@@ -331,6 +331,10 @@ public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
return new SolverResult(state, false);
}
+ if (Thread.currentThread().isInterrupted()) {
+ return new SolverResult(state, false);
+ }
+
ExecutorDetails exec = state.currentExec();
Iterable<String> sortedNodes = sortAllNodes(state.td, exec, favoredNodeIds, unFavoredNodeIds);
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
index a832557..6c3c1f7 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
@@ -47,7 +47,7 @@ public class DefaultResourceAwareStrategy extends BaseResourceAwareStrategy impl
}
Collection<ExecutorDetails> unassignedExecutors =
new HashSet<>(this.cluster.getUnassignedExecutors(td));
- LOG.debug("ExecutorsNeedScheduling: {}", unassignedExecutors);
+ LOG.debug("{} ExecutorsNeedScheduling: {}", td.getId(), unassignedExecutors);
Collection<ExecutorDetails> scheduledTasks = new ArrayList<>();
List<Component> spouts = this.getSpouts(td);
@@ -65,6 +65,9 @@ public class DefaultResourceAwareStrategy extends BaseResourceAwareStrategy impl
final Iterable<String> sortedNodes = sortAllNodes(td, null, favoredNodesIds, unFavoredNodesIds);
for (ExecutorDetails exec : orderedExecutors) {
+ if (Thread.currentThread().isInterrupted()) {
+ return null;
+ }
LOG.debug(
"Attempting to schedule: {} of component {}[ REQ {} ]",
exec,
@@ -79,6 +82,9 @@ public class DefaultResourceAwareStrategy extends BaseResourceAwareStrategy impl
LOG.debug("/* Scheduling left over task (most likely sys tasks) */");
// schedule left over system tasks
for (ExecutorDetails exec : executorsNotScheduled) {
+ if (Thread.currentThread().isInterrupted()) {
+ return null;
+ }
if (!scheduleExecutor(exec, td, scheduledTasks, sortedNodes)) {
return mkNotEnoughResources(td);
}
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/GenericResourceAwareStrategy.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/GenericResourceAwareStrategy.java
index 0c85e1c..b2555f2 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/GenericResourceAwareStrategy.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/GenericResourceAwareStrategy.java
@@ -86,7 +86,7 @@ public class GenericResourceAwareStrategy extends BaseResourceAwareStrategy impl
}
Collection<ExecutorDetails> unassignedExecutors =
new HashSet<>(this.cluster.getUnassignedExecutors(td));
- LOG.debug("Num ExecutorsNeedScheduling: {}", unassignedExecutors.size());
+ LOG.debug("{} Num ExecutorsNeedScheduling: {}", td.getId(), unassignedExecutors.size());
Collection<ExecutorDetails> scheduledTasks = new ArrayList<>();
List<Component> spouts = this.getSpouts(td);
@@ -103,6 +103,9 @@ public class GenericResourceAwareStrategy extends BaseResourceAwareStrategy impl
List<String> unFavoredNodeIds = makeHostToNodeIds((List<String>) td.getConf().get(Config.TOPOLOGY_SCHEDULER_UNFAVORED_NODES));
for (ExecutorDetails exec : orderedExecutors) {
+ if (Thread.currentThread().isInterrupted()) {
+ return null;
+ }
LOG.debug(
"Attempting to schedule: {} of component {}[ REQ {} ]",
exec,
@@ -120,6 +123,9 @@ public class GenericResourceAwareStrategy extends BaseResourceAwareStrategy impl
LOG.warn("Scheduling {} left over task (most likely sys tasks)", executorsNotScheduled);
// schedule left over system tasks
for (ExecutorDetails exec : executorsNotScheduled) {
+ if (Thread.currentThread().isInterrupted()) {
+ return null;
+ }
final Iterable<String> sortedNodes = sortAllNodes(td, exec, favoredNodeIds, unFavoredNodeIds);
if (!scheduleExecutor(exec, td, scheduledTasks, sortedNodes)) {
return mkNotEnoughResources(td);
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/IStrategy.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/IStrategy.java
index 2cdbb32..d246b30 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/IStrategy.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/IStrategy.java
@@ -19,7 +19,6 @@ import org.apache.storm.scheduler.resource.SchedulingResult;
/**
* An interface to for implementing different scheduling strategies for the resource aware scheduling.
- * In the future strategies will be pluggable
*/
public interface IStrategy {
@@ -32,6 +31,10 @@ public interface IStrategy {
/**
* This method is invoked to calculate a scheduling for topology td. Cluster will reject any changes that are
* not for the given topology. Any changes made to the cluster will be committed if the scheduling is successful.
+ * <P></P>
+ * NOTE: scheduling occurs as a runnable in an interruptible thread. Scheduling should consider being interrupted if
+ * long running.
+ *
* @param schedulingState the current state of the cluster
* @param td the topology to schedule for
* @return returns a SchedulingResult object containing SchedulingStatus object to indicate whether scheduling is
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/blacklist/TestBlacklistScheduler.java b/storm-server/src/test/java/org/apache/storm/scheduler/blacklist/TestBlacklistScheduler.java
index 7815154..76ad01f 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/blacklist/TestBlacklistScheduler.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/blacklist/TestBlacklistScheduler.java
@@ -25,11 +25,13 @@ import org.apache.storm.DaemonConfig;
import org.apache.storm.scheduler.Cluster;
import org.apache.storm.scheduler.DefaultScheduler;
import org.apache.storm.scheduler.INimbus;
+import org.apache.storm.scheduler.IScheduler;
import org.apache.storm.scheduler.SchedulerAssignmentImpl;
import org.apache.storm.scheduler.SupervisorDetails;
import org.apache.storm.scheduler.Topologies;
import org.apache.storm.scheduler.TopologyDetails;
import org.apache.storm.utils.Utils;
+import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
@@ -50,6 +52,15 @@ public class TestBlacklistScheduler {
private static final Logger LOG = LoggerFactory.getLogger(TestBlacklistScheduler.class);
private static int currentTime = 1468216504;
+ private static IScheduler scheduler = null;
+
+ @After
+ public void cleanup() {
+ if (scheduler != null) {
+ scheduler.cleanup();
+ scheduler = null;
+ }
+ }
@Test
public void TestBadSupervisor() {
@@ -72,15 +83,15 @@ public class TestBlacklistScheduler {
StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
ResourceMetrics resourceMetrics = new ResourceMetrics(metricsRegistry);
Cluster cluster = new Cluster(iNimbus, resourceMetrics, supMap, new HashMap<>(), topologies, config);
- BlacklistScheduler bs = new BlacklistScheduler(new DefaultScheduler(), metricsRegistry);
- bs.prepare(config);
- bs.schedule(topologies, cluster);
+ scheduler = new BlacklistScheduler(new DefaultScheduler(), metricsRegistry);
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
cluster = new Cluster(iNimbus, resourceMetrics, TestUtilsForBlacklistScheduler.removeSupervisorFromSupervisors(supMap, "sup-0"), TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
cluster = new Cluster(iNimbus, resourceMetrics, TestUtilsForBlacklistScheduler.removeSupervisorFromSupervisors(supMap, "sup-0"), TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
cluster = new Cluster(iNimbus, resourceMetrics, supMap, TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
Assert.assertEquals("blacklist", Collections.singleton("host-0"), cluster.getBlacklistedHosts());
}
@@ -105,16 +116,16 @@ public class TestBlacklistScheduler {
StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
ResourceMetrics resourceMetrics = new ResourceMetrics(metricsRegistry);
Cluster cluster = new Cluster(iNimbus, resourceMetrics, supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
- BlacklistScheduler bs = new BlacklistScheduler(new DefaultScheduler(), metricsRegistry);
- bs.prepare(config);
- bs.schedule(topologies, cluster);
+ scheduler = new BlacklistScheduler(new DefaultScheduler(), metricsRegistry);
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
cluster = new Cluster(iNimbus, resourceMetrics, TestUtilsForBlacklistScheduler.removePortFromSupervisors(supMap, "sup-0", 0), TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
cluster = new Cluster(iNimbus, resourceMetrics, TestUtilsForBlacklistScheduler.removePortFromSupervisors(supMap, "sup-0", 0), TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
cluster = new Cluster(iNimbus, resourceMetrics, supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
Assert.assertEquals("blacklist", Collections.singleton("host-0"), cluster.getBlacklistedHosts());
}
@@ -139,21 +150,21 @@ public class TestBlacklistScheduler {
StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
ResourceMetrics resourceMetrics = new ResourceMetrics(metricsRegistry);
Cluster cluster = new Cluster(iNimbus, resourceMetrics, supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
- BlacklistScheduler bs = new BlacklistScheduler(new DefaultScheduler(), metricsRegistry);
- bs.prepare(config);
- bs.schedule(topologies, cluster);
+ scheduler = new BlacklistScheduler(new DefaultScheduler(), metricsRegistry);
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
cluster = new Cluster(iNimbus, resourceMetrics, TestUtilsForBlacklistScheduler.removeSupervisorFromSupervisors(supMap, "sup-0"), TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
cluster = new Cluster(iNimbus, resourceMetrics, TestUtilsForBlacklistScheduler.removeSupervisorFromSupervisors(supMap, "sup-0"), TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
cluster = new Cluster(iNimbus, resourceMetrics, supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
Assert.assertEquals("blacklist", Collections.singleton("host-0"), cluster.getBlacklistedHosts());
for (int i = 0; i < 300 / 10 - 2; i++) {
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
}
Assert.assertEquals("blacklist", Collections.singleton("host-0"), cluster.getBlacklistedHosts());
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
Assert.assertEquals("blacklist", Collections.emptySet(), cluster.getBlacklistedHosts());
}
@@ -181,22 +192,22 @@ public class TestBlacklistScheduler {
StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
ResourceMetrics resourceMetrics = new ResourceMetrics(metricsRegistry);
Cluster cluster = new Cluster(iNimbus, resourceMetrics, supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
- BlacklistScheduler bs = new BlacklistScheduler(new DefaultScheduler(), metricsRegistry);
- bs.prepare(config);
- bs.schedule(topologies, cluster);
+ scheduler = new BlacklistScheduler(new DefaultScheduler(), metricsRegistry);
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
cluster = new Cluster(iNimbus, resourceMetrics, TestUtilsForBlacklistScheduler.removeSupervisorFromSupervisors(supMap, "sup-0"), TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
cluster = new Cluster(iNimbus, resourceMetrics, TestUtilsForBlacklistScheduler.removeSupervisorFromSupervisors(supMap, "sup-0"), TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
cluster = new Cluster(iNimbus, resourceMetrics, supMap, TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
Assert.assertEquals("blacklist", Collections.singleton("host-0"), cluster.getBlacklistedHosts());
topoMap.put(topo2.getId(), topo2);
topoMap.put(topo3.getId(), topo3);
topoMap.put(topo4.getId(), topo4);
topologies = new Topologies(topoMap);
cluster = new Cluster(iNimbus, resourceMetrics, supMap, TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
Assert.assertEquals("blacklist", Collections.emptySet(), cluster.getBlacklistedHosts());
}
@@ -215,8 +226,8 @@ public class TestBlacklistScheduler {
topoMap.put(topo1.getId(), topo1);
topoMap.put(topo2.getId(), topo2);
Topologies topologies = new Topologies(topoMap);
- BlacklistScheduler bs = new BlacklistScheduler(new DefaultScheduler(), new StormMetricsRegistry());
- bs.prepare(config);
+ scheduler = new BlacklistScheduler(new DefaultScheduler(), new StormMetricsRegistry());
+ scheduler.prepare(config);
List<Map<Integer, List<Integer>>> faultList = new ArrayList<>();
@@ -242,7 +253,7 @@ public class TestBlacklistScheduler {
int count = 0;
for (Map<String, SupervisorDetails> supervisors : supervisorsList) {
cluster = FaultGenerateUtils.nextCluster(cluster, supervisors, iNimbus, config, topologies);
- bs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
if (count == 0) {
Set<String> hosts = new HashSet<>();
Assert.assertEquals("blacklist", hosts, cluster.getBlacklistedHosts());
@@ -282,7 +293,6 @@ public class TestBlacklistScheduler {
}
count++;
}
-
}
@Test
@@ -306,6 +316,7 @@ public class TestBlacklistScheduler {
ResourceMetrics resourceMetrics = new ResourceMetrics(metricsRegistry);
Cluster cluster = new Cluster(iNimbus, resourceMetrics, supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
BlacklistScheduler bs = new BlacklistScheduler(new DefaultScheduler(), metricsRegistry);
+ scheduler = bs;
bs.prepare(config);
bs.schedule(topologies,cluster);
cluster = new Cluster(iNimbus, resourceMetrics, TestUtilsForBlacklistScheduler.removeSupervisorFromSupervisors(supMap,"sup-0"),TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
@@ -315,7 +326,7 @@ public class TestBlacklistScheduler {
Set<String> cached = new HashSet<>();
cached.add("sup-1");
cached.add("sup-2");
- Assert.assertEquals(cached,bs.cachedSupervisors.keySet());
+ Assert.assertEquals(cached, bs.cachedSupervisors.keySet());
cluster = new Cluster(iNimbus, resourceMetrics, supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
bs.schedule(topologies,cluster);
cluster = new Cluster(iNimbus, resourceMetrics, TestUtilsForBlacklistScheduler.removePortFromSupervisors(supMap,"sup-0",0),TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/resource/TestResourceAwareScheduler.java b/storm-server/src/test/java/org/apache/storm/scheduler/resource/TestResourceAwareScheduler.java
index 798a4a3..1791a11 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/resource/TestResourceAwareScheduler.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/resource/TestResourceAwareScheduler.java
@@ -24,13 +24,17 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
+import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicLong;
+
import org.apache.storm.Config;
+import org.apache.storm.DaemonConfig;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.generated.WorkerResources;
import org.apache.storm.scheduler.Cluster;
import org.apache.storm.scheduler.ExecutorDetails;
import org.apache.storm.scheduler.INimbus;
+import org.apache.storm.scheduler.IScheduler;
import org.apache.storm.scheduler.SchedulerAssignment;
import org.apache.storm.scheduler.SchedulerAssignmentImpl;
import org.apache.storm.scheduler.SupervisorDetails;
@@ -38,8 +42,10 @@ import org.apache.storm.scheduler.Topologies;
import org.apache.storm.scheduler.TopologyDetails;
import org.apache.storm.scheduler.WorkerSlot;
import org.apache.storm.scheduler.resource.normalization.NormalizedResources;
+import org.apache.storm.scheduler.resource.strategies.scheduling.BaseResourceAwareStrategy;
import org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy;
import org.apache.storm.scheduler.resource.strategies.scheduling.GenericResourceAwareStrategy;
+import org.apache.storm.scheduler.resource.strategies.scheduling.IStrategy;
import org.apache.storm.testing.PerformanceTest;
import org.apache.storm.testing.TestWordCounter;
import org.apache.storm.testing.TestWordSpout;
@@ -50,6 +56,7 @@ import org.apache.storm.utils.ReflectionUtils;
import org.apache.storm.utils.Time;
import org.apache.storm.utils.Utils;
import org.apache.storm.validation.ConfigValidation;
+import org.junit.After;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -68,6 +75,7 @@ public class TestResourceAwareScheduler {
private static final Logger LOG = LoggerFactory.getLogger(TestResourceAwareScheduler.class);
private static final Config defaultTopologyConf = createClusterConfig(10, 128, 0, null);
private static int currentTime = 1450418597;
+ private static IScheduler scheduler = null;
@BeforeAll
public static void initConf() {
@@ -75,6 +83,14 @@ public class TestResourceAwareScheduler {
defaultTopologyConf.put(Config.TOPOLOGY_PRIORITY, 0);
}
+ @After
+ public void cleanup() {
+ if (scheduler != null) {
+ scheduler.cleanup();
+ scheduler = null;
+ }
+ }
+
@Test
public void testRASNodeSlotAssign() {
Config config = new Config();
@@ -150,15 +166,15 @@ public class TestResourceAwareScheduler {
Config config = new Config();
config.putAll(defaultTopologyConf);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ scheduler = new ResourceAwareScheduler();
TopologyDetails topology1 = genTopology("topology1", config, 1, 1, 1, 1, 0, 0, "user");
Topologies topologies = new Topologies(topology1);
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- rs.prepare(config);
- rs.schedule(topologies, cluster);
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
SchedulerAssignment assignment = cluster.getAssignmentById(topology1.getId());
Set<WorkerSlot> assignedSlots = assignment.getSlots();
@@ -201,13 +217,13 @@ public class TestResourceAwareScheduler {
Map<ExecutorDetails, String> executorMap2 = genExecsAndComps(stormTopology2);
TopologyDetails topology2 = new TopologyDetails("topology2", config, stormTopology2, 0, executorMap2, 0, "user");
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ scheduler = new ResourceAwareScheduler();
Topologies topologies = new Topologies(topology1, topology2);
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- rs.prepare(config);
- rs.schedule(topologies, cluster);
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
SchedulerAssignment assignment1 = cluster.getAssignmentById(topology1.getId());
Set<WorkerSlot> assignedSlots1 = assignment1.getSlots();
@@ -252,6 +268,7 @@ public class TestResourceAwareScheduler {
TopologyDetails topology1 = new TopologyDetails("topology1", config, stormTopology1, 0, executorMap1, 0, "user");
ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ scheduler = rs;
Topologies topologies = new Topologies(topology1);
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
@@ -296,6 +313,7 @@ public class TestResourceAwareScheduler {
TopologyDetails topology1 = new TopologyDetails("topology1", config, stormTopology1, 2, executorMap1, 0, "user");
ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ scheduler = rs;
Topologies topologies = new Topologies(topology1);
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
@@ -391,12 +409,12 @@ public class TestResourceAwareScheduler {
TopologyDetails topology2 = new TopologyDetails("topology2", config2, stormTopology2, 2, executorMap2, 0, "user");
// Test1: When a worker fails, RAS does not alter existing assignments on healthy workers
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ scheduler = new ResourceAwareScheduler();
Topologies topologies = new Topologies(topology2);
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);
- rs.prepare(config1);
- rs.schedule(topologies, cluster);
+ scheduler.prepare(config1);
+ scheduler.schedule(topologies, cluster);
SchedulerAssignment assignment = cluster.getAssignmentById(topology2.getId());
// pick a worker to mock as failed
@@ -414,7 +432,7 @@ public class TestResourceAwareScheduler {
Map<ExecutorDetails, WorkerSlot> copyOfOldMapping = new HashMap<>(executorToSlot);
Set<ExecutorDetails> healthyExecutors = copyOfOldMapping.keySet();
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
SchedulerAssignment newAssignment = cluster.getAssignmentById(topology2.getId());
Map<ExecutorDetails, WorkerSlot> newExecutorToSlot = newAssignment.getExecutorToSlot();
@@ -439,7 +457,7 @@ public class TestResourceAwareScheduler {
topologies = new Topologies(topology1);
Cluster cluster1 = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap1, existingAssignments, topologies, config1);
- rs.schedule(topologies, cluster1);
+ scheduler.schedule(topologies, cluster1);
newAssignment = cluster1.getAssignmentById(topology1.getId());
newExecutorToSlot = newAssignment.getExecutorToSlot();
@@ -468,7 +486,7 @@ public class TestResourceAwareScheduler {
topologies = new Topologies(topology1);
cluster1 = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap1, existingAssignments, topologies, config1);
- rs.schedule(topologies, cluster1);
+ scheduler.schedule(topologies, cluster1);
newAssignment = cluster1.getAssignmentById(topology1.getId());
newExecutorToSlot = newAssignment.getExecutorToSlot();
@@ -482,14 +500,14 @@ public class TestResourceAwareScheduler {
// Test4: Scheduling a new topology does not disturb other assignments unnecessarily
topologies = new Topologies(topology1);
cluster1 = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);
- rs.schedule(topologies, cluster1);
+ scheduler.schedule(topologies, cluster1);
assignment = cluster1.getAssignmentById(topology1.getId());
executorToSlot = assignment.getExecutorToSlot();
copyOfOldMapping = new HashMap<>(executorToSlot);
topologies = addTopologies(topologies, topology2);
cluster1 = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);
- rs.schedule(topologies, cluster1);
+ scheduler.schedule(topologies, cluster1);
newAssignment = cluster1.getAssignmentById(topology1.getId());
newExecutorToSlot = newAssignment.getExecutorToSlot();
@@ -576,25 +594,34 @@ public class TestResourceAwareScheduler {
Topologies topologies = new Topologies(topology1, topology2, topology3);
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);
rs.prepare(config1);
- rs.schedule(topologies, cluster);
- assertEquals("Running - Fully Scheduled by " + strategyName, cluster.getStatusMap().get(topology1.getId()));
- assertEquals("Running - Fully Scheduled by " + strategyName, cluster.getStatusMap().get(topology2.getId()));
- assertEquals("Running - Fully Scheduled by " + strategyName, cluster.getStatusMap().get(topology3.getId()));
+ Map<SupervisorDetails, Double> superToCpu = null;
+ Map<SupervisorDetails, Double> superToMem = null;
+
+ try {
+ rs.schedule(topologies, cluster);
+
+ assertEquals("Running - Fully Scheduled by " + strategyName, cluster.getStatusMap().get(topology1.getId()));
+ assertEquals("Running - Fully Scheduled by " + strategyName, cluster.getStatusMap().get(topology2.getId()));
+ assertEquals("Running - Fully Scheduled by " + strategyName, cluster.getStatusMap().get(topology3.getId()));
- Map<SupervisorDetails, Double> superToCpu = getSupervisorToCpuUsage(cluster, topologies);
- Map<SupervisorDetails, Double> superToMem = getSupervisorToMemoryUsage(cluster, topologies);
+ superToCpu = getSupervisorToCpuUsage(cluster, topologies);
+ superToMem = getSupervisorToMemoryUsage(cluster, topologies);
- final Double EPSILON = 0.0001;
- for (SupervisorDetails supervisor : supMap.values()) {
- Double cpuAvailable = supervisor.getTotalCpu();
- Double memAvailable = supervisor.getTotalMemory();
- Double cpuUsed = superToCpu.get(supervisor);
- Double memUsed = superToMem.get(supervisor);
+ final Double EPSILON = 0.0001;
+ for (SupervisorDetails supervisor : supMap.values()) {
+ Double cpuAvailable = supervisor.getTotalCpu();
+ Double memAvailable = supervisor.getTotalMemory();
+ Double cpuUsed = superToCpu.get(supervisor);
+ Double memUsed = superToMem.get(supervisor);
- assertTrue(supervisor.getId() + " MEM: " + memAvailable + " == " + memUsed + " OR CPU: " + cpuAvailable + " == " + cpuUsed,
- (Math.abs(memAvailable - memUsed) < EPSILON) || (Math.abs(cpuAvailable - cpuUsed) < EPSILON));
+ assertTrue(supervisor.getId() + " MEM: " + memAvailable + " == " + memUsed + " OR CPU: " + cpuAvailable + " == " + cpuUsed,
+ (Math.abs(memAvailable - memUsed) < EPSILON) || (Math.abs(cpuAvailable - cpuUsed) < EPSILON));
+ }
+ } finally {
+ rs.cleanup();
}
+
// end of Test1
LOG.warn("\n\n\t\tSwitching to topologies 1, 2 and 4");
@@ -603,21 +630,25 @@ public class TestResourceAwareScheduler {
topologies = new Topologies(topology1, topology2, topology4);
cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);
rs.prepare(config1);
- rs.schedule(topologies, cluster);
- int numTopologiesAssigned = 0;
- if (cluster.getStatusMap().get(topology1.getId()).equals("Running - Fully Scheduled by " + strategyName)) {
- LOG.info("TOPO 1 scheduled");
- numTopologiesAssigned++;
- }
- if (cluster.getStatusMap().get(topology2.getId()).equals("Running - Fully Scheduled by " + strategyName)) {
- LOG.info("TOPO 2 scheduled");
- numTopologiesAssigned++;
- }
- if (cluster.getStatusMap().get(topology4.getId()).equals("Running - Fully Scheduled by " + strategyName)) {
- LOG.info("TOPO 3 scheduled");
- numTopologiesAssigned++;
+ try {
+ rs.schedule(topologies, cluster);
+ int numTopologiesAssigned = 0;
+ if (cluster.getStatusMap().get(topology1.getId()).equals("Running - Fully Scheduled by " + strategyName)) {
+ LOG.info("TOPO 1 scheduled");
+ numTopologiesAssigned++;
+ }
+ if (cluster.getStatusMap().get(topology2.getId()).equals("Running - Fully Scheduled by " + strategyName)) {
+ LOG.info("TOPO 2 scheduled");
+ numTopologiesAssigned++;
+ }
+ if (cluster.getStatusMap().get(topology4.getId()).equals("Running - Fully Scheduled by " + strategyName)) {
+ LOG.info("TOPO 3 scheduled");
+ numTopologiesAssigned++;
+ }
+ assertEquals(2, numTopologiesAssigned);
+ } finally {
+ rs.cleanup();
}
- assertEquals(2, numTopologiesAssigned);
//end of Test2
LOG.info("\n\n\t\tScheduling just topo 5");
@@ -625,16 +656,20 @@ public class TestResourceAwareScheduler {
topologies = new Topologies(topology5);
cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);
rs.prepare(config1);
- rs.schedule(topologies, cluster);
- superToCpu = getSupervisorToCpuUsage(cluster, topologies);
- superToMem = getSupervisorToMemoryUsage(cluster, topologies);
- for (SupervisorDetails supervisor : supMap.values()) {
- Double cpuAvailable = supervisor.getTotalCpu();
- Double memAvailable = supervisor.getTotalMemory();
- Double cpuUsed = superToCpu.get(supervisor);
- Double memUsed = superToMem.get(supervisor);
- assertEquals(cpuAvailable, cpuUsed, 0.0001);
- assertEquals(memAvailable, memUsed, 0.0001);
+ try {
+ rs.schedule(topologies, cluster);
+ superToCpu = getSupervisorToCpuUsage(cluster, topologies);
+ superToMem = getSupervisorToMemoryUsage(cluster, topologies);
+ for (SupervisorDetails supervisor : supMap.values()) {
+ Double cpuAvailable = supervisor.getTotalCpu();
+ Double memAvailable = supervisor.getTotalMemory();
+ Double cpuUsed = superToCpu.get(supervisor);
+ Double memUsed = superToMem.get(supervisor);
+ assertEquals(cpuAvailable, cpuUsed, 0.0001);
+ assertEquals(memAvailable, memUsed, 0.0001);
+ }
+ } finally {
+ rs.cleanup();
}
//end of Test3
}
@@ -669,9 +704,13 @@ public class TestResourceAwareScheduler {
Topologies topologies = new Topologies(topology1);
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);
rs.prepare(config1);
- rs.schedule(topologies, cluster);
- assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster.getStatusMap().get(topology1.getId()));
- assertEquals(4, cluster.getAssignedNumWorkers(topology1));
+ try {
+ rs.schedule(topologies, cluster);
+ assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster.getStatusMap().get(topology1.getId()));
+ assertEquals(4, cluster.getAssignedNumWorkers(topology1));
+ } finally {
+ rs.cleanup();
+ }
// Test2: test when no more workers are available due to topology worker max heap size limit but there is memory is still available
// wordSpout2 is going to contain 5 executors that needs scheduling. Each of those executors has a memory requirement of 128.0 MB
@@ -689,11 +728,15 @@ public class TestResourceAwareScheduler {
topologies = new Topologies(topology2);
cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config2);
rs.prepare(config2);
- rs.schedule(topologies, cluster);
- String status = cluster.getStatusMap().get(topology2.getId());
- assert status.startsWith("Not enough resources to schedule") : status;
- assert status.endsWith("5 executors not scheduled") : status;
- assertEquals(5, cluster.getUnassignedExecutors(topology2).size());
+ try {
+ rs.schedule(topologies, cluster);
+ String status = cluster.getStatusMap().get(topology2.getId());
+ assert status.startsWith("Not enough resources to schedule") : status;
+ assert status.endsWith("5 executors not scheduled") : status;
+ assertEquals(5, cluster.getUnassignedExecutors(topology2).size());
+ } finally {
+ rs.cleanup();
+ }
}
@Test
@@ -720,10 +763,9 @@ public class TestResourceAwareScheduler {
genTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 20, "bobby"));
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
- rs.prepare(config);
- rs.schedule(topologies, cluster);
-
+ scheduler = new ResourceAwareScheduler();
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
assertTopologiesFullyScheduled(cluster, "topo-1", "topo-2", "topo-3", "topo-4");
assertTopologiesNotScheduled(cluster, "topo-5");
}
@@ -756,10 +798,9 @@ public class TestResourceAwareScheduler {
genTopology("topo-15", config, 5, 15, 1, 1, currentTime - 24, 29, "derek"));
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
-
- rs.prepare(config);
- rs.schedule(topologies, cluster);
+ scheduler = new ResourceAwareScheduler();
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
for (TopologyDetails td : topologies) {
assertTopologiesFullyScheduled(cluster, td.getName());
}
@@ -779,10 +820,9 @@ public class TestResourceAwareScheduler {
genTopology("topo-1", config, 5, 15, 1, 1, currentTime - 2, 20, "jerry"),
genTopology("topo-2", config, 5, 15, 1, 1, currentTime - 8, 29, "jerry"));
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
-
- rs.prepare(config);
- rs.schedule(topologies, cluster);
+ scheduler = new ResourceAwareScheduler();
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
assertTopologiesFullyScheduled(cluster, "topo-1");
assertTopologiesNotScheduled(cluster, "topo-2");
@@ -810,10 +850,10 @@ public class TestResourceAwareScheduler {
genTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 29, "derek"),
genTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 10, "derek"));
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ scheduler = new ResourceAwareScheduler();
- rs.prepare(config);
- rs.schedule(topologies, cluster);
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
assertTopologiesFullyScheduled(cluster, "topo-1", "topo-2", "topo-3", "topo-4", "topo-5", "topo-6");
@@ -839,7 +879,7 @@ public class TestResourceAwareScheduler {
LOG.warn("Rescheduling with removed Supervisor....");
cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, newAssignments, topologies, config);
cluster.setStatusMap(statusMap);
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
assertTopologiesFullyScheduled(cluster, "topo-2", "topo-3", "topo-4", "topo-5", "topo-6");
assertTopologiesNotScheduled(cluster, "topo-1");
@@ -858,10 +898,10 @@ public class TestResourceAwareScheduler {
genTopology("topo-1", config, 1, 0, 2, 0, currentTime - 2, 29, "user"),
genTopology("topo-2", config, 1, 0, 2, 0, currentTime - 2, 10, "user"));
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ scheduler = new ResourceAwareScheduler();
- rs.prepare(config);
- rs.schedule(topologies, cluster);
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
Map<String, RAS_Node> nodes = RAS_Nodes.getAllNodesFrom(cluster);
@@ -900,10 +940,10 @@ public class TestResourceAwareScheduler {
Topologies topologies = new Topologies(topo1, topo2, topo3);
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ scheduler = new ResourceAwareScheduler();
- rs.prepare(config);
- rs.schedule(topologies, cluster);
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
assertTrue("Topo-2 scheduled?", cluster.getAssignmentById(topo2.getId()) != null);
assertEquals("Topo-2 all executors scheduled?", 4, cluster.getAssignmentById(topo2.getId()).getExecutorToSlot().size());
@@ -941,10 +981,9 @@ public class TestResourceAwareScheduler {
Topologies topologies = new Topologies(topo);
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
-
- rs.prepare(config);
- rs.schedule(topologies, cluster);
+ scheduler = new ResourceAwareScheduler();
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
assertTrue("Topo scheduled?", cluster.getAssignmentById(topo.getId()) != null);
assertEquals("Topo all executors scheduled?", 25, cluster.getAssignmentById(topo.getId()).getExecutorToSlot().size());
@@ -994,6 +1033,65 @@ public class TestResourceAwareScheduler {
() -> testLargeTopologiesCommon(GenericResourceAwareStrategy.class.getName(), true, 1));
}
+ public static class NeverEndingSchedulingStrategy extends BaseResourceAwareStrategy implements IStrategy {
+
+ @Override
+ public void prepare(Map<String, Object> config) {
+ }
+
+ @Override
+ protected TreeSet<ObjectResources> sortObjectResources(
+ BaseResourceAwareStrategy.AllResources allResources, ExecutorDetails exec, TopologyDetails topologyDetails, ExistingScheduleFunc existingScheduleFunc) {
+ // NO-OP
+ return null;
+ }
+
+ @Override
+ public SchedulingResult schedule(Cluster schedulingState, TopologyDetails td) {
+ this.cluster = schedulingState;
+ while (true) {
+ if (Thread.currentThread().isInterrupted()) {
+ LOG.info("scheduling interrupted");
+ return null;
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testStrategyTakingTooLong() {
+ INimbus iNimbus = new INimbusTest();
+ Map<String, SupervisorDetails> supMap = genSupervisors(8, 4, 100, 1000);
+ Config config = createClusterConfig(100, 500, 500, null);
+ List<String> allowedSchedulerStrategies = new ArrayList();
+ allowedSchedulerStrategies.add(DefaultResourceAwareStrategy.class.getName());
+ allowedSchedulerStrategies.add(NeverEndingSchedulingStrategy.class.getName());
+ config.put(Config.NIMBUS_SCHEDULER_STRATEGY_CLASS_WHITELIST, allowedSchedulerStrategies);
+ config.put(DaemonConfig.SCHEDULING_TIMEOUT_SECONDS_PER_TOPOLOGY, 30);
+
+ TopologyDetails topo1 = genTopology("topo-1", config, 1, 0, 2, 0, currentTime - 2, 10, "jerry");
+ TopologyDetails topo3 = genTopology("topo-3", config, 1, 2, 1, 1, currentTime - 2, 20, "jerry");
+
+ config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, NeverEndingSchedulingStrategy.class.getName());
+ TopologyDetails topo2 = genTopology("topo-2", config, 2, 0, 2, 0, currentTime - 2, 20, "jerry");
+
+ Topologies topologies = new Topologies(topo1, topo2, topo3);
+ Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
+ scheduler = new ResourceAwareScheduler();
+
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
+
+ assertTrue("Topo-1 scheduled?", cluster.getAssignmentById(topo1.getId()) != null);
+ assertEquals("Topo-1 all executors scheduled?", 2, cluster.getAssignmentById(topo1.getId()).getExecutorToSlot().size());
+ assertTrue("Topo-2 not scheduled", cluster.getAssignmentById(topo2.getId()) == null);
+ assertEquals("Scheduling took too long for " + topo2.getId() + " using strategy "
+ + NeverEndingSchedulingStrategy.class.getName()
+ + " timeout after 30 seconds using config scheduling.timeout.seconds.per.topology.", cluster.getStatusMap().get(topo2.getId()));
+ assertTrue("Topo-3 scheduled?", cluster.getAssignmentById(topo3.getId()) != null);
+ assertEquals("Topo-3 all executors scheduled?", 3, cluster.getAssignmentById(topo3.getId()).getExecutorToSlot().size());
+ }
+
public void testLargeTopologiesCommon(final String strategy, final boolean includeGpu, final int multiplier) {
INimbus iNimbus = new INimbusTest();
Map<String, SupervisorDetails> supMap = genSupervisorsWithRacks(25 * multiplier, 40, 66, 3 * multiplier, 0, 4700, 226200, new HashMap<>());
@@ -1007,7 +1105,7 @@ public class TestResourceAwareScheduler {
config.putAll(createClusterConfig(88, 775, 25, null));
config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, strategy);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ scheduler = new ResourceAwareScheduler();
Map<String, TopologyDetails> topologyDetailsMap = new HashMap<>();
for (int i = 0; i < 11 * multiplier; i++) {
@@ -1030,8 +1128,8 @@ public class TestResourceAwareScheduler {
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
long startTime = Time.currentTimeMillis();
- rs.prepare(config);
- rs.schedule(topologies, cluster);
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
long schedulingDuration = Time.currentTimeMillis() - startTime;
LOG.info("Scheduling took " + schedulingDuration + " ms");
LOG.info("HAS {} SLOTS USED", cluster.getUsedSlots().size());
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java
index f8c2aeb..7baca51 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java
@@ -21,9 +21,11 @@ package org.apache.storm.scheduler.resource.strategies.eviction;
import org.apache.storm.Config;
import org.apache.storm.scheduler.Cluster;
import org.apache.storm.scheduler.INimbus;
+import org.apache.storm.scheduler.IScheduler;
import org.apache.storm.scheduler.SupervisorDetails;
import org.apache.storm.scheduler.Topologies;
import org.apache.storm.scheduler.resource.ResourceAwareScheduler;
+import org.junit.After;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,7 +41,16 @@ import org.apache.storm.scheduler.resource.normalization.ResourceMetrics;
public class TestDefaultEvictionStrategy {
private static final Logger LOG = LoggerFactory.getLogger(TestDefaultEvictionStrategy.class);
private static int currentTime = 1450418597;
-
+ private static IScheduler scheduler = null;
+
+ @After
+ public void cleanup() {
+ if (scheduler != null) {
+ scheduler.cleanup();
+ scheduler = null;
+ }
+ }
+
/**
* The resources in the cluster are limited. In the first round of scheduling, all resources in the cluster is used.
* User jerry submits another topology. Since user jerry has his resource guarantees satisfied, and user bobby
@@ -61,9 +72,9 @@ public class TestDefaultEvictionStrategy {
genTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 29, "derek"));
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
- rs.prepare(config);
- rs.schedule(topologies, cluster);
+ scheduler = new ResourceAwareScheduler();
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
assertTopologiesFullyScheduled(cluster, "topo-1", "topo-2", "topo-3", "topo-4");
@@ -71,7 +82,7 @@ public class TestDefaultEvictionStrategy {
topologies = addTopologies(topologies,
genTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 20, "jerry"));
cluster = new Cluster(cluster, topologies);
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
//topo-3 evicted (lowest priority)
assertTopologiesFullyScheduled(cluster, "topo-1", "topo-2", "topo-4", "topo-6");
@@ -93,10 +104,10 @@ public class TestDefaultEvictionStrategy {
genTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 29, "derek"),
genTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 29, "derek"));
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
- rs.prepare(config);
+ scheduler = new ResourceAwareScheduler();
+ scheduler.prepare(config);
LOG.info("\n\n\t\tScheduling topos 2 to 5...");
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
LOG.info("\n\n\t\tDone scheduling...");
assertTopologiesFullyScheduled(cluster, "topo-2", "topo-3", "topo-4", "topo-5");
@@ -105,7 +116,7 @@ public class TestDefaultEvictionStrategy {
genTopology("topo-1", config, 2, 0, 1, 0, currentTime - 2, 10, "jerry"));
cluster = new Cluster(cluster, topologies);
LOG.info("\n\n\t\tScheduling topos 1 to 5");
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
LOG.info("\n\n\t\tDone scheduling...");
//bobby has no guarantee so topo-2 and topo-3 evicted
assertTopologiesFullyScheduled(cluster, "topo-1", "topo-4", "topo-5");
@@ -127,9 +138,9 @@ public class TestDefaultEvictionStrategy {
genTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 29, "derek"),
genTopology("topo-5", config, 1, 0, 1, 0, currentTime - 15, 29, "derek"));
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
- rs.prepare(config);
- rs.schedule(topologies, cluster);
+ scheduler = new ResourceAwareScheduler();
+ scheduler.prepare(config);
+ scheduler.schedule(topologies, cluster);
assertTopologiesFullyScheduled(cluster, "topo-2", "topo-3", "topo-4", "topo-5");
@@ -137,7 +148,7 @@ public class TestDefaultEvictionStrategy {
topologies = addTopologies(topologies,
genTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 10, "jerry"));
cluster = new Cluster(cluster, topologies);
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
//topo-3 evicted since user bobby don't have any resource guarantees and topo-3 is the lowest priority for user bobby
assertTopologiesFullyScheduled(cluster, "topo-1", "topo-2", "topo-4", "topo-5");
@@ -146,7 +157,7 @@ public class TestDefaultEvictionStrategy {
topologies = addTopologies(topologies,
genTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 10, "jerry"));
cluster = new Cluster(cluster, topologies);
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
//topo-2 evicted since user bobby don't have any resource guarantees and topo-2 is the next lowest priority for user bobby
assertTopologiesFullyScheduled(cluster, "topo-1", "topo-4", "topo-5");
@@ -155,7 +166,7 @@ public class TestDefaultEvictionStrategy {
topologies = addTopologies(topologies,
genTopology("topo-7", config, 1, 0, 1, 0, currentTime - 2, 10, "jerry"));
cluster = new Cluster(cluster, topologies);
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
// since user derek has exceeded his resource guarantee while user jerry has not topo-5 or topo-4 could be evicted because they have the same priority
// but topo-4 was submitted earlier thus we choose that one to evict (somewhat arbitrary)
@@ -183,10 +194,10 @@ public class TestDefaultEvictionStrategy {
genTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 10, "bobby"),
genTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 29, "derek"));
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
- rs.prepare(config);
+ scheduler = new ResourceAwareScheduler();
+ scheduler.prepare(config);
LOG.info("\n\n\t\tScheduling topos 1,2,5,6");
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
LOG.info("\n\n\t\tDone Scheduling...");
assertTopologiesFullyScheduled(cluster, "topo-1", "topo-2", "topo-5", "topo-6");
@@ -197,7 +208,7 @@ public class TestDefaultEvictionStrategy {
genTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 29, "jerry"));
cluster = new Cluster(cluster, topologies);
LOG.info("\n\n\t\tScheduling topos 1,2,3,5,6");
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
LOG.info("\n\n\t\tDone Scheduling...");
assertTopologiesFullyScheduled(cluster, "topo-1", "topo-2", "topo-5", "topo-6");
@@ -209,7 +220,7 @@ public class TestDefaultEvictionStrategy {
genTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 10, "jerry"));
cluster = new Cluster(cluster, topologies);
LOG.info("\n\n\t\tScheduling topos 1-6");
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
LOG.info("\n\n\t\tDone Scheduling...");
assertTopologiesFullyScheduled(cluster, "topo-1", "topo-4", "topo-5", "topo-6");
@@ -235,10 +246,10 @@ public class TestDefaultEvictionStrategy {
genTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 10, "bobby"),
genTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 29, "derek"));
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
- rs.prepare(config);
+ scheduler = new ResourceAwareScheduler();
+ scheduler.prepare(config);
LOG.info("\n\n\t\tScheduling topos 1,3,4,5");
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
LOG.info("\n\n\t\tDone scheduling...");
assertTopologiesFullyScheduled(cluster, "topo-1", "topo-3", "topo-4", "topo-5");
@@ -250,7 +261,7 @@ public class TestDefaultEvictionStrategy {
genTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 10, "derek"));
cluster = new Cluster(cluster, topologies);
LOG.info("\n\n\t\tScheduling topos 1,3,4,5,6");
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
LOG.info("\n\n\t\tDone scheduling...");
assertTopologiesFullyScheduled(cluster, "topo-1", "topo-3", "topo-4", "topo-6");
@@ -261,7 +272,7 @@ public class TestDefaultEvictionStrategy {
genTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 20, "jerry"));
cluster = new Cluster(cluster, topologies);
LOG.info("\n\n\t\tScheduling topos 1-6");
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
LOG.info("\n\n\t\tDone scheduling...");
assertTopologiesFullyScheduled(cluster, "topo-1", "topo-3", "topo-4", "topo-6");
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/priority/TestFIFOSchedulingPriorityStrategy.java b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/priority/TestFIFOSchedulingPriorityStrategy.java
index 472ea5e..e6d93db 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/priority/TestFIFOSchedulingPriorityStrategy.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/priority/TestFIFOSchedulingPriorityStrategy.java
@@ -60,33 +60,37 @@ public class TestFIFOSchedulingPriorityStrategy {
ResourceAwareScheduler rs = new ResourceAwareScheduler();
rs.prepare(config);
- rs.schedule(topologies, cluster);
+ try {
+ rs.schedule(topologies, cluster);
- assertTopologiesFullyScheduled(cluster, "topo-1-jerry", "topo-2-bobby", "topo-3-bobby", "topo-4-derek");
+ assertTopologiesFullyScheduled(cluster, "topo-1-jerry", "topo-2-bobby", "topo-3-bobby", "topo-4-derek");
- LOG.info("\n\n\t\tINSERTING topo-5");
- //new topology needs to be scheduled
- //topo-3 should be evicted since its been up the longest
- topologies = addTopologies(topologies,
- genTopology("topo-5-derek", config, 1, 0, 1, 0,Time.currentTimeSecs() - 15,29, "derek"));
+ LOG.info("\n\n\t\tINSERTING topo-5");
+ //new topology needs to be scheduled
+ //topo-3 should be evicted since its been up the longest
+ topologies = addTopologies(topologies,
+ genTopology("topo-5-derek", config, 1, 0, 1, 0, Time.currentTimeSecs() - 15, 29, "derek"));
- cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- rs.schedule(topologies, cluster);
+ cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
+ rs.schedule(topologies, cluster);
- assertTopologiesFullyScheduled(cluster, "topo-1-jerry", "topo-2-bobby", "topo-4-derek", "topo-5-derek");
- assertTopologiesNotScheduled(cluster, "topo-3-bobby");
+ assertTopologiesFullyScheduled(cluster, "topo-1-jerry", "topo-2-bobby", "topo-4-derek", "topo-5-derek");
+ assertTopologiesNotScheduled(cluster, "topo-3-bobby");
- LOG.info("\n\n\t\tINSERTING topo-6");
- //new topology needs to be scheduled. topo-4 should be evicted. Even though topo-1 from user jerry is older, topo-1 will not be evicted
- //since user jerry has enough resource guarantee
- topologies = addTopologies(topologies,
- genTopology("topo-6-bobby", config, 1, 0, 1, 0,Time.currentTimeSecs() - 10,29, "bobby"));
+ LOG.info("\n\n\t\tINSERTING topo-6");
+ //new topology needs to be scheduled. topo-4 should be evicted. Even though topo-1 from user jerry is older, topo-1 will not be evicted
+ //since user jerry has enough resource guarantee
+ topologies = addTopologies(topologies,
+ genTopology("topo-6-bobby", config, 1, 0, 1, 0, Time.currentTimeSecs() - 10, 29, "bobby"));
- cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- rs.schedule(topologies, cluster);
+ cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
+ rs.schedule(topologies, cluster);
- assertTopologiesFullyScheduled(cluster, "topo-1-jerry", "topo-2-bobby", "topo-5-derek", "topo-6-bobby");
- assertTopologiesNotScheduled(cluster, "topo-3-bobby", "topo-4-derek");
+ assertTopologiesFullyScheduled(cluster, "topo-1-jerry", "topo-2-bobby", "topo-5-derek", "topo-6-bobby");
+ assertTopologiesNotScheduled(cluster, "topo-3-bobby", "topo-4-derek");
+ } finally {
+ rs.cleanup();
+ }
}
}
}
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
index df3afba..d609bb6 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
@@ -223,10 +223,14 @@ public class TestConstraintSolverStrategy {
Cluster cluster = new Cluster(new INimbusTest(), new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
ResourceAwareScheduler rs = new ResourceAwareScheduler();
rs.prepare(config);
- rs.schedule(topologies, cluster);
+ try {
+ rs.schedule(topologies, cluster);
- assertStatusSuccess(cluster, topo.getId());
- Assert.assertEquals("topo all executors scheduled?", 0, cluster.getUnassignedExecutors(topo).size());
+ assertStatusSuccess(cluster, topo.getId());
+ Assert.assertEquals("topo all executors scheduled?", 0, cluster.getUnassignedExecutors(topo).size());
+ } finally {
+ rs.cleanup();
+ }
//simulate worker loss
Map<ExecutorDetails, WorkerSlot> newExecToSlot = new HashMap<>();
@@ -242,10 +246,14 @@ public class TestConstraintSolverStrategy {
cluster.setAssignments(newAssignments, false);
rs.prepare(config);
- rs.schedule(topologies, cluster);
+ try {
+ rs.schedule(topologies, cluster);
- assertStatusSuccess(cluster, topo.getId());
- Assert.assertEquals("topo all executors scheduled?", 0, cluster.getUnassignedExecutors(topo).size());
+ assertStatusSuccess(cluster, topo.getId());
+ Assert.assertEquals("topo all executors scheduled?", 0, cluster.getUnassignedExecutors(topo).size());
+ } finally {
+ rs.cleanup();
+ }
}
public static void addContraints(String comp1, String comp2, List<List<String>> constraints) {
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java
index d552798..d72d362 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java
@@ -18,6 +18,7 @@
package org.apache.storm.scheduler.resource.strategies.scheduling;
+import org.apache.storm.scheduler.IScheduler;
import org.apache.storm.scheduler.resource.normalization.NormalizedResourcesRule;
import java.util.Collections;
import org.apache.storm.Config;
@@ -41,6 +42,7 @@ import org.apache.storm.topology.SharedOffHeapWithinNode;
import org.apache.storm.topology.SharedOffHeapWithinWorker;
import org.apache.storm.topology.SharedOnHeap;
import org.apache.storm.topology.TopologyBuilder;
+import org.junit.After;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
@@ -69,6 +71,7 @@ public class TestDefaultResourceAwareStrategy {
private static final Logger LOG = LoggerFactory.getLogger(TestDefaultResourceAwareStrategy.class);
private static final int CURRENT_TIME = 1450418597;
+ private static IScheduler scheduler = null;
private static class TestDNSToSwitchMapping implements DNSToSwitchMapping {
private final Map<String, String> result;
@@ -93,6 +96,14 @@ public class TestDefaultResourceAwareStrategy {
@Rule
public NormalizedResourcesRule nrRule = new NormalizedResourcesRule();
+ @After
+ public void cleanup() {
+ if (scheduler != null) {
+ scheduler.cleanup();
+ scheduler = null;
+ }
+ }
+
/**
* test if the scheduling logic for the DefaultResourceAwareStrategy is correct
*/
@@ -132,10 +143,10 @@ public class TestDefaultResourceAwareStrategy {
Topologies topologies = new Topologies(topo);
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, conf);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ scheduler = new ResourceAwareScheduler();
- rs.prepare(conf);
- rs.schedule(topologies, cluster);
+ scheduler.prepare(conf);
+ scheduler.schedule(topologies, cluster);
for (Entry<String, SupervisorResources> entry: cluster.getSupervisorsResourcesMap().entrySet()) {
String supervisorId = entry.getKey();
@@ -199,10 +210,10 @@ public class TestDefaultResourceAwareStrategy {
Topologies topologies = new Topologies(topo);
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, conf);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ scheduler = new ResourceAwareScheduler();
- rs.prepare(conf);
- rs.schedule(topologies, cluster);
+ scheduler.prepare(conf);
+ scheduler.schedule(topologies, cluster);
HashSet<HashSet<ExecutorDetails>> expectedScheduling = new HashSet<>();
expectedScheduling.add(new HashSet<>(Arrays.asList(new ExecutorDetails(0, 0)))); //Spout
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
index 124247e..037d226 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
@@ -34,6 +34,7 @@ import org.apache.storm.generated.WorkerResources;
import org.apache.storm.scheduler.Cluster;
import org.apache.storm.scheduler.ExecutorDetails;
import org.apache.storm.scheduler.INimbus;
+import org.apache.storm.scheduler.IScheduler;
import org.apache.storm.scheduler.SchedulerAssignment;
import org.apache.storm.scheduler.SupervisorDetails;
import org.apache.storm.scheduler.SupervisorResources;
@@ -45,6 +46,7 @@ import org.apache.storm.topology.SharedOffHeapWithinNode;
import org.apache.storm.topology.SharedOffHeapWithinWorker;
import org.apache.storm.topology.SharedOnHeap;
import org.apache.storm.topology.TopologyBuilder;
+import org.junit.After;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -59,6 +61,15 @@ public class TestGenericResourceAwareStrategy {
private static final Logger LOG = LoggerFactory.getLogger(TestGenericResourceAwareStrategy.class);
private static int currentTime = 1450418597;
+ private static IScheduler scheduler = null;
+
+ @After
+ public void cleanup() {
+ if (scheduler != null) {
+ scheduler.cleanup();
+ scheduler = null;
+ }
+ }
/**
* test if the scheduling logic for the GenericResourceAwareStrategy is correct.
@@ -104,10 +115,10 @@ public class TestGenericResourceAwareStrategy {
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, conf);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ scheduler = new ResourceAwareScheduler();
- rs.prepare(conf);
- rs.schedule(topologies, cluster);
+ scheduler.prepare(conf);
+ scheduler.schedule(topologies, cluster);
for (Entry<String, SupervisorResources> entry: cluster.getSupervisorsResourcesMap().entrySet()) {
String supervisorId = entry.getKey();
@@ -192,10 +203,10 @@ public class TestGenericResourceAwareStrategy {
Topologies topologies = new Topologies(topo);
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, conf);
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ scheduler = new ResourceAwareScheduler();
- rs.prepare(conf);
- rs.schedule(topologies, cluster);
+ scheduler.prepare(conf);
+ scheduler.schedule(topologies, cluster);
//We need to have 3 slots on 3 separate hosts. The topology needs 6 GPUs 3500 MB memory and 350% CPU
// The bolt-3 instances must be on separate nodes because they each need 2 GPUs.
@@ -236,8 +247,8 @@ public class TestGenericResourceAwareStrategy {
Config config = new Config();
config.putAll(createGrasClusterConfig(88, 775, 25, null, null));
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
- rs.prepare(config);
+ scheduler = new ResourceAwareScheduler();
+ scheduler.prepare(config);
TopologyDetails tdSimple = genTopology("topology-simple", config, 1,
5, 100, 300, 0, 0, "user", 8192);
@@ -245,7 +256,7 @@ public class TestGenericResourceAwareStrategy {
//Schedule the simple topology first
Topologies topologies = new Topologies(tdSimple);
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
TopologyBuilder builder = topologyBuilder(1, 5, 100, 300);
builder.setBolt("gpu-bolt", new TestBolt(), 40)
@@ -256,7 +267,7 @@ public class TestGenericResourceAwareStrategy {
//Now schedule GPU but with the simple topology in place.
topologies = new Topologies(tdSimple, tdGpu);
cluster = new Cluster(cluster, topologies);
- rs.schedule(topologies, cluster);
+ scheduler.schedule(topologies, cluster);
Map<String, SchedulerAssignment> assignments = new TreeMap<>(cluster.getAssignments());
assertEquals(2, assignments.size());