make isolationscheduler kill workers for all non-isolated topologies in case of squeezed cluster to make room for remaining isolated topology workers
diff --git a/src/clj/backtype/storm/scheduler/IsolationScheduler.clj b/src/clj/backtype/storm/scheduler/IsolationScheduler.clj
index 7e722cd..bd15f18 100644
--- a/src/clj/backtype/storm/scheduler/IsolationScheduler.clj
+++ b/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)
))
diff --git a/src/jvm/backtype/storm/scheduler/Cluster.java b/src/jvm/backtype/storm/scheduler/Cluster.java
index daececc..951bd3a 100644
--- a/src/jvm/backtype/storm/scheduler/Cluster.java
+++ b/src/jvm/backtype/storm/scheduler/Cluster.java
@@ -63,6 +63,10 @@
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);
