fix isolation scheduler to properly clear a machine of workers when putting an isolated topology there

Nathan Marz · Nathan Marz · commit a0e52c7d7fc9 · 2012-12-19T16:58:29.000-08:00
diff --git a/src/clj/backtype/storm/scheduler/IsolationScheduler.clj b/src/clj/backtype/storm/scheduler/IsolationScheduler.clj
@@ -96,6 +96,12 @@
        (LinkedList. <>)
        ))
 
+(defn- host->used-slots [^Cluster cluster]
+  (->> cluster
+       .getUsedSlots
+       (group-by #(.getHost cluster (.getNodeId ^WorkerSlot %)))
+       ))
+
 (defn- distribution->sorted-amts [distribution]
   (->> distribution
        (mapcat (fn [[val amt]] (repeat amt val)))
@@ -156,7 +162,8 @@
             ^Set worker-specs (get topology-worker-specs top-id)
             num-workers (count host-assignments)
             ]
-        (if (and (every? #(= (second %) top-id) assignments)
+        (if (and (contains? iso-ids-set top-id)
+                 (every? #(= (second %) top-id) assignments)
                  (contains? distribution num-workers)
                  (every? #(contains? worker-specs (nth % 2)) assignments))
           (do (decrement-distribution! distribution num-workers)
@@ -168,15 +175,16 @@
               ))
           )))
     
-    (let [^LinkedList sorted-assignable-hosts (host-assignable-slots cluster)]
+    (let [host->used-slots (host->used-slots cluster)
+          ^LinkedList sorted-assignable-hosts (host-assignable-slots cluster)]
       ;; TODO: can improve things further by ordering topologies in terms of who needs the least workers
       (doseq [[top-id worker-specs] topology-worker-specs
               :let [amts (distribution->sorted-amts (get topology-machine-distribution top-id))]]
         (doseq [amt amts
                 :let [[host host-slots] (.peek sorted-assignable-hosts)]]
           (when (and host-slots (>= (count host-slots) amt))
             (.poll sorted-assignable-hosts)
-            (.freeSlots cluster host-slots)
+            (.freeSlots cluster (get host->used-slots host))
             (doseq [slot (take amt host-slots)
                     :let [executors-set (remove-elem-from-set! worker-specs)]]
               (.assign cluster slot top-id executors-set))
diff --git a/src/jvm/backtype/storm/scheduler/Cluster.java b/src/jvm/backtype/storm/scheduler/Cluster.java
@@ -320,8 +320,10 @@ public void freeSlot(WorkerSlot slot) {
      * @param slots
      */
     public void freeSlots(Collection<WorkerSlot> slots) {
-        for (WorkerSlot slot : slots) {
-            this.freeSlot(slot);
+        if(slots!=null) {
+            for (WorkerSlot slot : slots) {
+                this.freeSlot(slot);
+            }
         }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -320,8 +320,10 @@ public void freeSlot(WorkerSlot slot) {`
`320`	`320`	`* @param slots`
`321`	`321`	`*/`
`322`	`322`	`public void freeSlots(Collection<WorkerSlot> slots) {`
`323`		`- for (WorkerSlot slot : slots) {`
`324`		`- this.freeSlot(slot);`
	`323`	`+ if(slots!=null) {`
	`324`	`+ for (WorkerSlot slot : slots) {`
	`325`	`+ this.freeSlot(slot);`
	`326`	`+ }`
`325`	`327`	`}`
`326`	`328`	`}`
`327`	`329`