       (is (= 2 (storm-num-workers state "mystorm"))) ;; because only 2 executors
       )))
 
+(deftest test-executor-assignments
+  (with-local-cluster [cluster :daemon-conf {SUPERVISOR-ENABLE false TOPOLOGY-ACKER-EXECUTORS 0}]
+    (let [nimbus (:nimbus cluster)
+          topology (thrift/mk-topology
+                    {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3 :conf {TOPOLOGY-TASKS 5})}
+                    {"2" (thrift/mk-bolt-spec {"1" :none} (TestPlannerBolt.) :parallelism-hint 8 :conf {TOPOLOGY-TASKS 2})
+                     "3" (thrift/mk-bolt-spec {"2" :none} (TestPlannerBolt.) :parallelism-hint 3)})
+          _ (submit-local-topology nimbus "mystorm" {TOPOLOGY-OPTIMIZE false TOPOLOGY-WORKERS 4} topology)
+          task-info (storm-component->task-info cluster "mystorm")
+          executor-info (->> (storm-component->executor-info cluster "mystorm")
+                             (map-val #(map executor-id->tasks %)))]
+      (check-consistency cluster "mystorm")
+      (is (= 5 (count (task-info "1"))))
+      (check-distribution (executor-info "1") [2 2 1])
+      
+      (is (= 2 (count (task-info "2"))))
+      (check-distribution (executor-info "2") [1 1])
+
+      (is (= 3 (count (task-info "3"))))
+      (check-distribution (executor-info "3") [1 1 1])
+      )))
+
 (deftest test-over-parallelism-assignment
   (with-local-cluster [cluster :supervisors 2 :ports-per-supervisor 5 :daemon-conf {SUPERVISOR-ENABLE false TOPOLOGY-ACKER-EXECUTORS 0}]
     (let [state (:storm-cluster-state cluster)

        (storm-task-info (from-json (.getTopologyConf nimbus storm-id)))
        reverse-map)))

(defn storm-num-workers [state storm-name]
  (let [storm-id (get-storm-id state storm-name)
        assignment (.assignment-info state storm-id nil)]

      (check-consistency cluster "test")
      )))

(defn check-executor-distribution [slot-executors distribution]
  (check-distribution (vals slot-executors) distribution))

(defn check-num-nodes [slot-executors num-nodes]
  (let [nodes (->> slot-executors keys (map first) set)]
