You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2017/07/31 02:07:01 UTC
[02/20] storm git commit: STORM-2497: Let Supervisor enforce memory
and add in support for shared memory regions
http://git-wip-us.apache.org/repos/asf/storm/blob/b4d33955/storm-server/src/test/java/org/apache/storm/scheduler/resource/TestUtilsForResourceAwareScheduler.java
----------------------------------------------------------------------
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/resource/TestUtilsForResourceAwareScheduler.java b/storm-server/src/test/java/org/apache/storm/scheduler/resource/TestUtilsForResourceAwareScheduler.java
index beca679..cae2e20 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/resource/TestUtilsForResourceAwareScheduler.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/resource/TestUtilsForResourceAwareScheduler.java
@@ -19,6 +19,7 @@
package org.apache.storm.scheduler.resource;
import org.apache.storm.Config;
+import org.apache.storm.DaemonConfig;
import org.apache.storm.generated.Bolt;
import org.apache.storm.generated.SpoutSpec;
import org.apache.storm.generated.StormTopology;
@@ -26,11 +27,15 @@ 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.ISchedulingState;
import org.apache.storm.scheduler.SchedulerAssignment;
import org.apache.storm.scheduler.SupervisorDetails;
import org.apache.storm.scheduler.Topologies;
import org.apache.storm.scheduler.TopologyDetails;
import org.apache.storm.scheduler.WorkerSlot;
+import org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy;
+import org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy;
+import org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
@@ -53,60 +58,83 @@ import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Random;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
public class TestUtilsForResourceAwareScheduler {
- private static int currentTime = 1450418597;
-
private static final Logger LOG = LoggerFactory.getLogger(TestUtilsForResourceAwareScheduler.class);
-
- public static List<TopologyDetails> getListOfTopologies(Config config, String user) {
-
- List<TopologyDetails> topos = new LinkedList<>();
-
- topos.add(TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 5, 15, 1, 1, currentTime - 2, 20, user));
- topos.add(TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 5, 15, 1, 1, currentTime - 8, 30, user));
- topos.add(TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 5, 15, 1, 1, currentTime - 16, 30, user));
- topos.add(TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 5, 15, 1, 1, currentTime - 16, 20, user));
- topos.add(TestUtilsForResourceAwareScheduler.getTopology("topo-5", config, 5, 15, 1, 1, currentTime - 24, 30, user));
- topos.add(TestUtilsForResourceAwareScheduler.getTopology("topo-6", config, 5, 15, 1, 1, currentTime - 2, 0, user));
- topos.add(TestUtilsForResourceAwareScheduler.getTopology("topo-7", config, 5, 15, 1, 1, currentTime - 8, 0, user ));
- topos.add(TestUtilsForResourceAwareScheduler.getTopology("topo-8", config, 5, 15, 1, 1, currentTime - 16, 15, user));
- topos.add(TestUtilsForResourceAwareScheduler.getTopology("topo-9", config, 5, 15, 1, 1, currentTime - 16, 8, user));
- topos.add(TestUtilsForResourceAwareScheduler.getTopology("topo-10", config, 5, 15, 1, 1, currentTime - 24, 9, user));
- return topos;
+ public static final int currentTime = 1450418597;
+
+ public static class TestUserResources {
+ private final String name;
+ private final Map<String, Number> resources = new HashMap<>();
+
+ public TestUserResources(String name, double cpu, double mem) {
+ this.name = name;
+ resources.put("cpu", cpu);
+ resources.put("memory", mem);
+ }
+
+ public void addSelfTo(Map<String, Map<String, Number>> fullPool) {
+ if (fullPool.put(name, resources) != null) {
+ throw new IllegalStateException("Cannot have 2 copies of " + name + " in a pool");
+ }
+ }
}
-
- public static List<String> getListOfTopologiesCorrectOrder() {
- List<String> topos = new LinkedList<String>();
- topos.add("topo-7");
- topos.add("topo-6");
- topos.add("topo-9");
- topos.add("topo-10");
- topos.add("topo-8");
- topos.add("topo-4");
- topos.add("topo-1");
- topos.add("topo-5");
- topos.add("topo-3");
- topos.add("topo-2");
- return topos;
+
+ public static TestUserResources userRes(String name, double cpu, double mem) {
+ return new TestUserResources(name, cpu, mem);
}
-
- public static Map<String, SupervisorDetails> genSupervisors(int numSup, int numPorts, Map resourceMap) {
- return genSupervisors(numSup, numPorts, 0, resourceMap);
+
+ public static Map<String, Double> toDouble(Map<String, Number> resources) {
+ Map<String, Double> ret = new HashMap<>();
+ for (Entry<String, Number> entry: resources.entrySet()) {
+ ret.put(entry.getKey(), entry.getValue().doubleValue());
+ }
+ return ret;
+ }
+
+ public static Map<String, Map<String, Number>> userResourcePool(TestUserResources... resources) {
+ Map<String, Map<String, Number>> ret = new HashMap<>();
+ for (TestUserResources res: resources) {
+ res.addSelfTo(ret);
+ }
+ return ret;
+ }
+
+ public static Config createClusterConfig(double compPcore, double compOnHeap, double compOffHeap, Map<String, Map<String, Number>> pools) {
+ Config config = new Config();
+ config.putAll(Utils.readDefaultConfig());
+ config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, DefaultEvictionStrategy.class.getName());
+ config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, DefaultSchedulingPriorityStrategy.class.getName());
+ config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, DefaultResourceAwareStrategy.class.getName());
+ config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, compPcore);
+ config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, compOffHeap);
+ config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, compOnHeap);
+ if (pools != null) {
+ config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_USER_POOLS, pools);
+ }
+ return config;
}
- public static Map<String, SupervisorDetails> genSupervisors(int numSup, int numPorts, int start, Map resourceMap) {
+ public static Map<String, SupervisorDetails> genSupervisors(int numSup, int numPorts, double cpu, double mem) {
+ return genSupervisors(numSup, numPorts, 0, cpu, mem);
+ }
+
+ public static Map<String, SupervisorDetails> genSupervisors(int numSup, int numPorts, int start, double cpu, double mem) {
+ Map<String, Double> resourceMap = new HashMap<>();
+ resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, cpu);
+ resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, mem);
Map<String, SupervisorDetails> retList = new HashMap<String, SupervisorDetails>();
for (int i = start; i < numSup + start; i++) {
List<Number> ports = new LinkedList<Number>();
for (int j = 0; j < numPorts; j++) {
ports.add(j);
}
- SupervisorDetails sup = new SupervisorDetails("sup-" + i, "host-" + i, null, ports, new HashMap<String, Double>(resourceMap));
+ SupervisorDetails sup = new SupervisorDetails("sup-" + i, "host-" + i, null, ports, resourceMap);
retList.put(sup.getId(), sup);
}
return retList;
@@ -140,18 +168,33 @@ public class TestUtilsForResourceAwareScheduler {
return retMap;
}
- public static TopologyDetails getTopology(String name, Map<String, Object> config, int numSpout, int numBolt,
- int spoutParallelism, int boltParallelism, int launchTime, int priority, String owner) {
+ public static Topologies addTopologies(Topologies topos, TopologyDetails ... details) {
+ Map<String, TopologyDetails> topoMap = new HashMap<>();
+ for (TopologyDetails td: topos.getTopologies()) {
+ topoMap.put(td.getId(), td);
+ }
+ for (TopologyDetails td: details) {
+ if (topoMap.put(td.getId(), td) != null) {
+ throw new IllegalArgumentException("Cannot have multiple topologies with id " + td.getId());
+ }
+ }
+ return new Topologies(topoMap);
+ }
+
+ public static TopologyDetails genTopology(String name, Map<String, Object> config, int numSpout, int numBolt,
+ int spoutParallelism, int boltParallelism, int launchTime, int priority,
+ String user) {
Config conf = new Config();
conf.putAll(config);
conf.put(Config.TOPOLOGY_PRIORITY, priority);
conf.put(Config.TOPOLOGY_NAME, name);
+ conf.put(Config.TOPOLOGY_SUBMITTER_USER, user);
conf.put(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB, Double.MAX_VALUE);
StormTopology topology = buildTopology(numSpout, numBolt, spoutParallelism, boltParallelism);
TopologyDetails topo = new TopologyDetails(name + "-" + launchTime, conf, topology,
0,
- genExecsAndComps(topology), launchTime, owner);
+ genExecsAndComps(topology), launchTime, user);
return topo;
}
@@ -285,7 +328,37 @@ public class TestUtilsForResourceAwareScheduler {
return m.find();
}
- public static boolean assertStatusSuccess(String status) {
+ public static void assertTopologiesNotScheduled(Cluster cluster, String ... topoNames) {
+ Topologies topologies = cluster.getTopologies();
+ for (String topoName: topoNames) {
+ TopologyDetails td = topologies.getByName(topoName);
+ assert(td != null) : topoName;
+ String topoId = td.getId();
+ String status = cluster.getStatus(topoId);
+ assert(status != null) : topoName;
+ assert(!isStatusSuccess(status)) : topoName;
+ assert(cluster.getAssignmentById(topoId) == null) : topoName;
+ assert(cluster.needsSchedulingRas(td)) : topoName;
+ }
+ }
+
+ public static void assertTopologiesFullyScheduled(Cluster cluster, String ... topoNames) {
+ Topologies topologies = cluster.getTopologies();
+ for (String topoName: topoNames) {
+ TopologyDetails td = topologies.getByName(topoName);
+ assert(td != null) : topoName;
+ String topoId = td.getId();
+ assertStatusSuccess(cluster, topoId);
+ assert(cluster.getAssignmentById(topoId) != null) : topoName;
+ assert(cluster.needsSchedulingRas(td) == false) : topoName;
+ }
+ }
+
+ public static void assertStatusSuccess(Cluster cluster, String topoId) {
+ assert(isStatusSuccess(cluster.getStatus(topoId))) : "topology status " + topoId;
+ }
+
+ public static boolean isStatusSuccess(String status) {
return isContain(status, "fully") && isContain(status, "scheduled") && !isContain(status, "unsuccessful");
}
@@ -299,7 +372,7 @@ public class TestUtilsForResourceAwareScheduler {
return ret;
}
- public static Map<SupervisorDetails, Double> getSupervisorToMemoryUsage(Cluster cluster, Topologies topologies) {
+ public static Map<SupervisorDetails, Double> getSupervisorToMemoryUsage(ISchedulingState cluster, Topologies topologies) {
Map<SupervisorDetails, Double> superToMem = new HashMap<>();
Collection<SchedulerAssignment> assignments = cluster.getAssignments().values();
Collection<SupervisorDetails> supervisors = cluster.getSupervisors().values();
@@ -333,7 +406,7 @@ public class TestUtilsForResourceAwareScheduler {
return superToMem;
}
- public static Map<SupervisorDetails, Double> getSupervisorToCpuUsage(Cluster cluster, Topologies topologies) {
+ public static Map<SupervisorDetails, Double> getSupervisorToCpuUsage(ISchedulingState cluster, Topologies topologies) {
Map<SupervisorDetails, Double> superToCpu = new HashMap<>();
Collection<SchedulerAssignment> assignments = cluster.getAssignments().values();
Collection<SupervisorDetails> supervisors = cluster.getSupervisors().values();
http://git-wip-us.apache.org/repos/asf/storm/blob/b4d33955/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java
----------------------------------------------------------------------
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 020b187..666f90f 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
@@ -19,425 +19,143 @@
package org.apache.storm.scheduler.resource.strategies.eviction;
import org.apache.storm.Config;
-import org.apache.storm.DaemonConfig;
import org.apache.storm.scheduler.Cluster;
import org.apache.storm.scheduler.INimbus;
-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.scheduler.resource.ResourceAwareScheduler;
-import org.apache.storm.scheduler.resource.TestUtilsForResourceAwareScheduler;
-import org.apache.storm.utils.Utils;
-import org.junit.Assert;
import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
-public class TestDefaultEvictionStrategy {
+import static org.apache.storm.scheduler.resource.TestUtilsForResourceAwareScheduler.*;
+public class TestDefaultEvictionStrategy {
+ private static final Logger LOG = LoggerFactory.getLogger(TestDefaultEvictionStrategy.class);
private static int currentTime = 1450418597;
-
+
/**
* The resources in the cluster are limited. In the first round of scheduling, all resources in the cluster is used.
- * User jerry submits another toploogy. Since user jerry has his resource guarantees satisfied, and user bobby
+ * User jerry submits another topology. Since user jerry has his resource guarantees satisfied, and user bobby
* has exceeded his resource guarantee, topo-3 from user bobby should be evicted.
*/
@Test
public void testEviction() {
- INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
- Map<String, Number> resourceMap = new HashMap<String, Number>();
- resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
- resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
- Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
- Config config = new Config();
- config.putAll(Utils.readDefaultConfig());
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
- config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
- config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
- Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
- resourceUserPool.put("jerry", new HashMap<String, Number>());
- resourceUserPool.get("jerry").put("cpu", 200.0);
- resourceUserPool.get("jerry").put("memory", 2000.0);
-
- resourceUserPool.put("bobby", new HashMap<String, Number>());
- resourceUserPool.get("bobby").put("cpu", 100.0);
- resourceUserPool.get("bobby").put("memory", 1000.0);
-
- resourceUserPool.put("derek", new HashMap<String, Number>());
- resourceUserPool.get("derek").put("cpu", 200.0);
- resourceUserPool.get("derek").put("memory", 2000.0);
-
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
- Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
-
- TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 10,
- "jerry");
- TopologyDetails topo6 = TestUtilsForResourceAwareScheduler.getTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 20,
- "jerry");
-
- TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 10,
- "bobby");
- TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 20,
- "bobby");
-
- TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 29,
- "derek");
-
- Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
- topoMap.put(topo1.getId(), topo1);
- topoMap.put(topo2.getId(), topo2);
- topoMap.put(topo3.getId(), topo3);
- topoMap.put(topo4.getId(), topo4);
-
- Topologies topologies = new Topologies(topoMap);
-
+ INimbus iNimbus = new INimbusTest();
+ Map<String, SupervisorDetails> supMap = genSupervisors(4, 4, 100, 1000);
+ Map<String, Map<String, Number>> resourceUserPool = userResourcePool(
+ userRes("jerry", 200, 2000),
+ userRes("bobby", 100, 1000),
+ userRes("derek", 200, 2000));
+ Config config = createClusterConfig(100, 500, 500, resourceUserPool);
+ Topologies topologies = new Topologies(
+ genTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 10, "jerry"),
+ genTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 10, "bobby"),
+ genTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 20, "bobby"),
+ genTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 29, "derek"));
+
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<>(), topologies, config);
ResourceAwareScheduler rs = new ResourceAwareScheduler();
-
rs.prepare(config);
rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+
+ assertTopologiesFullyScheduled(cluster, "topo-1", "topo-2", "topo-3", "topo-4");
//user jerry submits another topology
- topoMap.put(topo6.getId(), topo6);
- topologies = new Topologies(topoMap);
-
+ topologies = addTopologies(topologies,
+ genTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 20, "jerry"));
+ cluster = new Cluster(cluster, topologies);
rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
- Assert.assertFalse("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of attempted topologies", 1, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("correct topology to evict", "topo-3", rs.getUser("bobby").getTopologiesAttempted().iterator().next().getName());
+
+ //topo-3 evicted (lowest priority)
+ assertTopologiesFullyScheduled(cluster, "topo-1", "topo-2", "topo-4", "topo-6");
+ assertTopologiesNotScheduled(cluster, "topo-3");
}
@Test
- public void TestEvictMultipleTopologies() {
- INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
- Map<String, Number> resourceMap = new HashMap<String, Number>();
- resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
- resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
- Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
- Config config = new Config();
- config.putAll(Utils.readDefaultConfig());
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
- config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
- config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
- Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
- resourceUserPool.put("jerry", new HashMap<String, Number>());
- resourceUserPool.get("jerry").put("cpu", 200.0);
- resourceUserPool.get("jerry").put("memory", 2000.0);
-
- resourceUserPool.put("derek", new HashMap<String, Number>());
- resourceUserPool.get("derek").put("cpu", 100.0);
- resourceUserPool.get("derek").put("memory", 1000.0);
-
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
- Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
-
- TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 2, 0, 1, 0, currentTime - 2, 10,
- "jerry");
-
- TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 10,
- "bobby");
- TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 20,
- "bobby");
-
- TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 29,
- "derek");
- TopologyDetails topo5 = TestUtilsForResourceAwareScheduler.getTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 29,
- "derek");
-
- Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
- topoMap.put(topo2.getId(), topo2);
- topoMap.put(topo3.getId(), topo3);
- topoMap.put(topo4.getId(), topo4);
- topoMap.put(topo5.getId(), topo5);
-
- Topologies topologies = new Topologies(topoMap);
-
+ public void testEvictMultipleTopologies() {
+ INimbus iNimbus = new INimbusTest();
+ Map<String, SupervisorDetails> supMap = genSupervisors(4, 4, 100, 1000);
+ Map<String, Map<String, Number>> resourceUserPool = userResourcePool(
+ userRes("jerry", 200, 2000),
+ userRes("derek", 100, 1000));
+ Config config = createClusterConfig(100, 500, 500, resourceUserPool);
+
+ Topologies topologies = new Topologies(
+ genTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 10, "bobby"),
+ genTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 20, "bobby"),
+ 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, supMap, new HashMap<>(), topologies, config);
ResourceAwareScheduler rs = new ResourceAwareScheduler();
-
rs.prepare(config);
rs.schedule(topologies, cluster);
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
+ assertTopologiesFullyScheduled(cluster, "topo-2", "topo-3", "topo-4", "topo-5");
//user jerry submits another topology
- topoMap.put(topo1.getId(), topo1);
- topologies = new Topologies(topoMap);
+ topologies = addTopologies(topologies,
+ genTopology("topo-1", config, 2, 0, 1, 0, currentTime - 2, 10, "jerry"));
+ cluster = new Cluster(cluster, topologies);
rs.schedule(topologies, cluster);
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
- Assert.assertFalse("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of attempted topologies", 2, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of running topologies", 0, rs.getUser("bobby").getTopologiesRunning().size());
+ //bobby has no guarantee so topo-2 and topo-3 evicted
+ assertTopologiesFullyScheduled(cluster, "topo-1", "topo-4", "topo-5");
+ assertTopologiesNotScheduled(cluster, "topo-2", "topo-3");
}
- /**
- * Eviction order:
- * topo-3: since user bobby don't have any resource guarantees and topo-3 is the lowest priority for user bobby
- * topo-2: since user bobby don't have any resource guarantees and topo-2 is the next lowest priority for user bobby
- * topo-5: since user derek has exceeded his resource guarantee while user jerry has not. topo-5 and topo-4 has the same priority
- * but topo-4 was submitted earlier thus we choose that one to evict
- */
@Test
- public void TestEvictMultipleTopologiesFromMultipleUsersInCorrectOrder() {
- INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
- Map<String, Number> resourceMap = new HashMap<String, Number>();
- resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
- resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
- Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
- Config config = new Config();
- config.putAll(Utils.readDefaultConfig());
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
- config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
- config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
- Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
- resourceUserPool.put("jerry", new HashMap<String, Number>());
- resourceUserPool.get("jerry").put("cpu", 300.0);
- resourceUserPool.get("jerry").put("memory", 3000.0);
-
- resourceUserPool.put("derek", new HashMap<String, Number>());
- resourceUserPool.get("derek").put("cpu", 100.0);
- resourceUserPool.get("derek").put("memory", 1000.0);
-
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
- Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
-
- TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 10,
- "jerry");
- TopologyDetails topo6 = TestUtilsForResourceAwareScheduler.getTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 10,
- "jerry");
- TopologyDetails topo7 = TestUtilsForResourceAwareScheduler.getTopology("topo-7", config, 1, 0, 1, 0, currentTime - 2, 10,
- "jerry");
-
- TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 10,
- "bobby");
- TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 20,
- "bobby");
-
- TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 29,
- "derek");
- TopologyDetails topo5 = TestUtilsForResourceAwareScheduler.getTopology("topo-5", config, 1, 0, 1, 0, currentTime - 15, 29,
- "derek");
-
- Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
- topoMap.put(topo2.getId(), topo2);
- topoMap.put(topo3.getId(), topo3);
- topoMap.put(topo4.getId(), topo4);
- topoMap.put(topo5.getId(), topo5);
-
- Topologies topologies = new Topologies(topoMap);
-
+ public void testEvictMultipleTopologiesFromMultipleUsersInCorrectOrder() {
+ INimbus iNimbus = new INimbusTest();
+ Map<String, SupervisorDetails> supMap = genSupervisors(4, 4, 100, 1000);
+ Map<String, Map<String, Number>> resourceUserPool = userResourcePool(
+ userRes("jerry", 300, 3000),
+ userRes("derek", 100, 1000));
+ Config config = createClusterConfig(100, 500, 500, resourceUserPool);
+
+ Topologies topologies = new Topologies(
+ genTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 10, "bobby"),
+ genTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 20, "bobby"),
+ 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, supMap, new HashMap<>(), topologies, config);
ResourceAwareScheduler rs = new ResourceAwareScheduler();
-
rs.prepare(config);
rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
+
+ assertTopologiesFullyScheduled(cluster, "topo-2", "topo-3", "topo-4", "topo-5");
//user jerry submits another topology
- topoMap.put(topo1.getId(), topo1);
- topologies = new Topologies(topoMap);
+ topologies = addTopologies(topologies,
+ genTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 10, "jerry"));
+ cluster = new Cluster(cluster, topologies);
rs.schedule(topologies, cluster);
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
- Assert.assertFalse("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of attempted topologies", 1, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("correct topology to evict", rs.getUser("bobby").getTopologiesAttempted().iterator().next().getName(), "topo-3");
-
- topoMap.put(topo6.getId(), topo6);
- topologies = new Topologies(topoMap);
+ //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");
+ assertTopologiesNotScheduled(cluster, "topo-3");
+
+ topologies = addTopologies(topologies,
+ genTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 10, "jerry"));
+ cluster = new Cluster(cluster, topologies);
rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
- Assert.assertFalse("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of attempted topologies", 2, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of running topologies", 0, rs.getUser("bobby").getTopologiesRunning().size());
-
- Assert.assertTrue("correct topology to evict", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-2", rs.getUser("bobby").getTopologiesAttempted()) != null);
- Assert.assertTrue("correct topology to evict", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-3", rs.getUser("bobby").getTopologiesAttempted()) != null);
-
- topoMap.put(topo7.getId(), topo7);
- topologies = new Topologies(topoMap);
+
+ //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");
+ assertTopologiesNotScheduled(cluster, "topo-2", "topo-3");
+
+ topologies = addTopologies(topologies,
+ genTopology("topo-7", config, 1, 0, 1, 0, currentTime - 2, 10, "jerry"));
+ cluster = new Cluster(cluster, topologies);
rs.schedule(topologies, cluster);
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 3, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesAttempted()) {
- Assert.assertFalse("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 1, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
- Assert.assertEquals("correct topology to evict", rs.getUser("derek").getTopologiesAttempted().iterator().next().getName(), "topo-4");
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
- Assert.assertFalse("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of attempted topologies", 2, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of running topologies", 0, rs.getUser("bobby").getTopologiesRunning().size());
-
- Assert.assertTrue("correct topology to evict", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-2", rs.getUser("bobby").getTopologiesAttempted()) != null);
- Assert.assertTrue("correct topology to evict", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-3", rs.getUser("bobby").getTopologiesAttempted()) != null);
+ // 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)
+ assertTopologiesFullyScheduled(cluster, "topo-1", "topo-5", "topo-7");
+ assertTopologiesNotScheduled(cluster, "topo-2", "topo-3", "topo-4");
}
/**
@@ -445,306 +163,91 @@ public class TestDefaultEvictionStrategy {
* check if there is a topology with lower priority that can be evicted from the current user
*/
@Test
- public void TestEvictTopologyFromItself() {
- INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
- Map<String, Number> resourceMap = new HashMap<String, Number>();
- resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
- resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
- Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
- Config config = new Config();
- config.putAll(Utils.readDefaultConfig());
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
- config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
- config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
- Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
- resourceUserPool.put("jerry", new HashMap<String, Number>());
- resourceUserPool.get("jerry").put("cpu", 200.0);
- resourceUserPool.get("jerry").put("memory", 2000.0);
-
- resourceUserPool.put("bobby", new HashMap<String, Number>());
- resourceUserPool.get("bobby").put("cpu", 100.0);
- resourceUserPool.get("bobby").put("memory", 1000.0);
-
- resourceUserPool.put("derek", new HashMap<String, Number>());
- resourceUserPool.get("derek").put("cpu", 100.0);
- resourceUserPool.get("derek").put("memory", 1000.0);
-
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
- Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
-
- TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 20,
- "jerry");
- TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 20,
- "jerry");
- TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 29,
- "jerry");
- TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 10,
- "jerry");
-
- TopologyDetails topo5 = TestUtilsForResourceAwareScheduler.getTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 10,
- "bobby");
-
- TopologyDetails topo6 = TestUtilsForResourceAwareScheduler.getTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 29,
- "derek");
-
- Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
- topoMap.put(topo1.getId(), topo1);
- topoMap.put(topo2.getId(), topo2);
- topoMap.put(topo5.getId(), topo5);
- topoMap.put(topo6.getId(), topo6);
-
- Topologies topologies = new Topologies(topoMap);
-
+ public void testEvictTopologyFromItself() {
+ INimbus iNimbus = new INimbusTest();
+ Map<String, SupervisorDetails> supMap = genSupervisors(4, 4, 100, 1000);
+ Map<String, Map<String, Number>> resourceUserPool = userResourcePool(
+ userRes("jerry", 200, 2000),
+ userRes("bobby", 100, 1000),
+ userRes("derek", 100, 1000));
+ Config config = createClusterConfig(100, 500, 500, resourceUserPool);
+
+ Topologies topologies = new Topologies(
+ genTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 20, "jerry"),
+ genTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 20, "jerry"),
+ 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, supMap, new HashMap<>(), topologies, config);
ResourceAwareScheduler rs = new ResourceAwareScheduler();
-
rs.prepare(config);
rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
+
+ assertTopologiesFullyScheduled(cluster, "topo-1", "topo-2", "topo-5", "topo-6");
//user jerry submits another topology into a full cluster
// topo3 should not be able to scheduled
- topoMap.put(topo3.getId(), topo3);
- topologies = new Topologies(topoMap);
+ topologies = addTopologies(topologies,
+ genTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 29, "jerry"));
+ cluster = new Cluster(cluster, topologies);
rs.schedule(topologies, cluster);
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 1, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
- //make sure that topo-3 didn't get scheduled.
- Assert.assertEquals("correct topology in attempted queue", rs.getUser("jerry").getTopologiesAttempted().iterator().next().getName(), "topo-3");
-
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+ assertTopologiesFullyScheduled(cluster, "topo-1", "topo-2", "topo-5", "topo-6");
+ assertTopologiesNotScheduled(cluster, "topo-3");
//user jerry submits another topology but this one should be scheduled since it has higher priority than than the
//rest of jerry's running topologies
- topoMap.put(topo4.getId(), topo4);
- topologies = new Topologies(topoMap);
+ topologies = addTopologies(topologies,
+ genTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 10, "jerry"));
+ cluster = new Cluster(cluster, topologies);
rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 2, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
- Assert.assertTrue("correct topology in attempted queue", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-3", rs.getUser("jerry").getTopologiesAttempted()) != null);
- //Either topo-1 or topo-2 should have gotten evicted
- Assert.assertTrue("correct topology in attempted queue", ((TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-1", rs.getUser("jerry").getTopologiesAttempted())) != null)
- || (TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-2", rs.getUser("jerry").getTopologiesAttempted()) != null));
- //assert that topo-4 got scheduled
- Assert.assertTrue("correct topology in running queue", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-4", rs.getUser("jerry").getTopologiesRunning()) != null);
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+
+ assertTopologiesFullyScheduled(cluster, "topo-2", "topo-4", "topo-5", "topo-6");
+ assertTopologiesNotScheduled(cluster, "topo-1", "topo-3");
}
/**
* If users are above his or her guarantee, check if topology eviction works correct
*/
@Test
- public void TestOverGuaranteeEviction() {
- INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
- Map<String, Number> resourceMap = new HashMap<String, Number>();
- resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
- resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
- Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
- Config config = new Config();
- config.putAll(Utils.readDefaultConfig());
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
- config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
- config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
- Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
- resourceUserPool.put("jerry", new HashMap<String, Number>());
- resourceUserPool.get("jerry").put("cpu", 70.0);
- resourceUserPool.get("jerry").put("memory", 700.0);
-
- resourceUserPool.put("bobby", new HashMap<String, Number>());
- resourceUserPool.get("bobby").put("cpu", 100.0);
- resourceUserPool.get("bobby").put("memory", 1000.0);
-
- resourceUserPool.put("derek", new HashMap<String, Number>());
- resourceUserPool.get("derek").put("cpu", 25.0);
- resourceUserPool.get("derek").put("memory", 250.0);
-
- config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
- Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
-
- TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 20,
- "jerry");
- TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 20,
- "jerry");
-
- TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 10,
- "bobby");
- TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 10,
- "bobby");
-
- TopologyDetails topo5 = TestUtilsForResourceAwareScheduler.getTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 29,
- "derek");
- TopologyDetails topo6 = TestUtilsForResourceAwareScheduler.getTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 10,
- "derek");
-
- Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
- topoMap.put(topo1.getId(), topo1);
- topoMap.put(topo3.getId(), topo3);
- topoMap.put(topo4.getId(), topo4);
- topoMap.put(topo5.getId(), topo5);
-
- Topologies topologies = new Topologies(topoMap);
-
+ public void testOverGuaranteeEviction() {
+ INimbus iNimbus = new INimbusTest();
+ Map<String, SupervisorDetails> supMap = genSupervisors(4, 4, 100, 1000);
+ Map<String, Map<String, Number>> resourceUserPool = userResourcePool(
+ userRes("jerry", 70, 700),
+ userRes("bobby", 100, 1000),
+ userRes("derek", 25, 250));
+ Config config = createClusterConfig(100, 500, 500, resourceUserPool);
+
+ Topologies topologies = new Topologies(
+ genTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 20, "jerry"),
+ genTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 10, "bobby"),
+ 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, supMap, new HashMap<>(), topologies, config);
ResourceAwareScheduler rs = new ResourceAwareScheduler();
-
rs.prepare(config);
rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
+
+ assertTopologiesFullyScheduled(cluster, "topo-1", "topo-3", "topo-4", "topo-5");
//user derek submits another topology into a full cluster
- //topo6 should not be able to scheduled intially, but since topo6 has higher priority than topo5
+ //topo6 should not be able to scheduled initially, but since topo6 has higher priority than topo5
//topo5 will be evicted so that topo6 can be scheduled
- topoMap.put(topo6.getId(), topo6);
- topologies = new Topologies(topoMap);
+ topologies = addTopologies(topologies,
+ genTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 10, "derek"));
+ cluster = new Cluster(cluster, topologies);
rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 1, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
- //topo5 will be evicted since topo6 has higher priority
- Assert.assertEquals("correct topology in attempted queue", "topo-5", rs.getUser("derek").getTopologiesAttempted().iterator().next().getName());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+
+ assertTopologiesFullyScheduled(cluster, "topo-1", "topo-3", "topo-4", "topo-6");
+ assertTopologiesNotScheduled(cluster, "topo-5");
//user jerry submits topo2
- topoMap.put(topo2.getId(), topo2);
- topologies = new Topologies(topoMap);
+ topologies = addTopologies(topologies,
+ genTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 20, "jerry"));
+ cluster = new Cluster(cluster, topologies);
rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 0, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 2, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
- Assert.assertEquals("correct topology in attempted queue", "topo-6", rs.getUser("derek").getTopologiesAttempted().iterator().next().getName());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+
+ assertTopologiesFullyScheduled(cluster, "topo-1", "topo-2", "topo-3", "topo-4");
+ assertTopologiesNotScheduled(cluster, "topo-5", "topo-6");
}
}