 
 (def n-messages-total 100000)
 
-(def load-data-fut (future 
-                     (loop [ls (repeatedly n-messages-total (fn [] {:id (java.util.UUID/randomUUID)}))]
-                       (when-let [segment (first ls)]
-                         (swap! incomplete conj segment)
-                         (>!! in-chan segment)
-                         (recur (rest ls))))
-                     (>!! in-chan :done)
-                     (close! in-chan)))
-
-
 (try
   (helper-env/add-peers test-env 9)
-  (let [job-id (:job-id (helper-env/run-job test-env 
+  (let [load-data-ch (thread 
+                       (loop [ls (repeatedly n-messages-total (fn [] {:id (java.util.UUID/randomUUID)}))]
+                         (when-let [segment (first ls)]
+                           (swap! incomplete conj segment)
+                           (>!! in-chan segment)
+                           (recur (rest ls))))
+                       (>!! in-chan :done)
+                       (close! in-chan))
+
+        job-id (:job-id (helper-env/run-job test-env 
                                             {:catalog catalog
                                              :workflow workflow
                                              :task-scheduler :onyx.task-scheduler/balanced}))
-        _ (println "Done submitting")
         _ (Thread/sleep 4000)
-        task-ids (->> (-> test-env
-                          :replica
-                          deref
-                          :allocations
-                          (get job-id))
+        task-peers (-> test-env
+                       :replica
+                       deref
+                       :allocations
+                       (get job-id))
+
+        task-ids (->> task-peers
                       keys
                       (map (partial extensions/read-chunk (:log (:env test-env)) :task))
                       (map (juxt :name :id))

            [onyx.test-helper :refer [load-config]]
            [onyx.api]))

;comment 
(def config (load-config))
(def test-env (component/start (->TestEnv (:env-config config) 
                                          (:peer-config config))))

    (<!! mess-with-peers-ch)
    (helper-env/remove-n-peers test-env 9))

  (finally 
    (component/stop test-env)))
