                            2 "1"
                            3 "1"
                            4 "1"}
-                          {1 ["sup1" 1]
-                           2 ["sup1" 2]
-                           3 ["sup1" 3]
-                           4 ["sup1" 3]
+                          {[1] ["sup1" 1]
+                           [2] ["sup1" 2]
+                           [3] ["sup1" 3]
+                           [4] ["sup1" 3]
                            })
                         (advance-cluster-time cluster 2)
                         (heartbeat-workers cluster "sup1" [1 2 3])



(defn worker-assignment
  "Return [storm-id taskids]"
  [cluster supervisor-id port]
  (let [state (:storm-cluster-state cluster)
        slot-assigns (for [storm-id (.assignments state nil)]
                        (let [tasks (-> (.assignment-info state storm-id nil)
                                        :task->node+port
                                        reverse-map
                                        (get [supervisor-id port] ))]
                          (when tasks [storm-id tasks])
                          ))
        ret (find-first not-nil? slot-assigns)]
    (when-not ret
      (throw (RuntimeException. "Could not find assignment for worker")))
    ret
    ))

(defn heartbeat-worker [supervisor port storm-id task-ids]
  (let [conf (.get-conf supervisor)]
    (worker/do-heartbeat {:conf conf
                          :port port
                          :storm-id storm-id
                          :task-ids task-ids
                          :worker-id (find-worker-id conf port)})))

(defn heartbeat-workers [cluster supervisor-id ports]
  (let [sup (get-supervisor cluster supervisor-id)]
    (doseq [p ports]
      (let [[storm-id task-ids] (worker-assignment cluster supervisor-id p)]
        (heartbeat-worker sup p storm-id task-ids)
        ))
    ))

(defn validate-launched-once [launched supervisor->ports storm-id]
  (let [counts (map count (vals launched))

                          {1 "1"
                           2 "1"
                           3 "1"}
                          {[1] ["sup1" 3]
                           [2] ["sup1" 3]
                           [3] ["sup2" 2]
                           })
                        (advance-cluster-time cluster 2)
                        (heartbeat-workers cluster "sup1" [3])
