make isolationscheduler kill workers for all non-isolated topologies …

…in case of squeezed cluster to make room for remaining isolated topology workers

src/clj/backtype/storm/scheduler/IsolationScheduler.clj

@@ -202,7 +202,13 @@
              allocated-topologies
              (leftover-topologies topologies <>)
              (DefaultScheduler/default-schedule <> cluster))
-        (log-warn "Unable to isolate topologies " (pr-str failed-iso-topologies) ". Will wait for enough resources for isolated topologies before allocating any other resources.")
+        (do
+          (log-warn "Unable to isolate topologies " (pr-str failed-iso-topologies) ". No machine had enough worker slots to run the remaining workers for these topologies. Clearing all other resources and will wait for enough resources for isolated topologies before allocating any other resources.")
+          ;; clear workers off all hosts that are not blacklisted
+          (doseq [[host slots] (host->used-slots cluster)]
+            (if-not (.isBlacklistedHost cluster host)
+              (.freeSlots cluster slots)
+              )))
         ))
     (.setBlacklistedHosts cluster orig-blacklist)
     ))

src/jvm/backtype/storm/scheduler/Cluster.java

@@ -63,6 +63,10 @@ public void blacklistHost(String host) {
     public boolean isBlackListed(String supervisorId) {
         return blackListedHosts != null && blackListedHosts.contains(getHost(supervisorId));        
     }
+
+    public boolean isBlacklistedHost(String host) {
+        return blackListedHosts != null && blackListedHosts.contains(host);  
+    }
 
     public String getHost(String supervisorId) {
         return inimbus.getHostName(supervisors, supervisorId);