don't scale workers down when the available slots is less than the as…

…signed slots
erikdw · Jul 9, 2016 · 3f901c7 · 3f901c7
1 parent d653da9
commit 3f901c7
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/storm/src/main/storm/mesos/schedulers/DefaultScheduler.java b/storm/src/main/storm/mesos/schedulers/DefaultScheduler.java
@@ -205,7 +205,7 @@ List<List<ExecutorDetails>> executorsPerWorkerList(Cluster cluster, TopologyDeta
     }
 
     // If there are not any unassigned executors, we need to re-distribute all currently existing executors across workers
-    if (executors.isEmpty()) {
+    if (executors.isEmpty() && slotsAvailable > slotsAssigned) {
       log.info("There are currently no unassigned executors. Using all currently existing executors instead.");
       SchedulerAssignment schedulerAssignment = cluster.getAssignmentById(topologyDetails.getId());
       // Un-assign them
@@ -214,6 +214,8 @@ List<List<ExecutorDetails>> executorsPerWorkerList(Cluster cluster, TopologyDeta
       log.info("executorsPerWorkerList - slotsAvailable: {}, slotsAssigned: {}, slotsFreed: {}", slotsAvailable, slotsAssigned, slotsFreed);
       executors = cluster.getUnassignedExecutors(topologyDetails);
       slotsAvailable += slotsAssigned;
+    } else if (executors.isEmpty()) {
+      log.info("There are currently no unassigned executors. Not going to redistribute work because slotsAvailable is {} and slotsAssigned is {}", slotsAvailable, slotsAssigned);
     }
 
     for (ExecutorDetails exec : executors) {