         :commit-tx-ch commit-tx-ch
         :close-resources-ch close-resources-ch
         :complete-task-ch complete-task-ch
-        :reset-payload-node-ch reset-payload-node-ch        
+        :reset-payload-node-ch reset-payload-node-ch
         
-        :open-session-loop (future (open-session-loop read-batch-ch pipeline-data))
+        :open-session-loop (thread (open-session-loop read-batch-ch open-session-kill-ch pipeline-data))
         :read-batch-loop (thread (read-batch-loop read-batch-ch decompress-batch-ch))
         :decompress-batch-loop (thread (decompress-batch-loop decompress-batch-ch apply-fn-ch))
         :apply-fn-loop (thread (apply-fn-loop apply-fn-ch compress-batch-ch))

  (stop [component]
    (prn "Stopping Task Pipeline")

    (future-cancel (:open-session-loop component))

    (close! (:read-batch-ch component))
    (close! (:decompress-batch-ch component))
    (close! (:apply-fn-ch component))

  (start [component]
    (prn "Starting Task Pipeline")
    
    (let [open-session-kill-ch (chan 1)
          read-batch-ch (chan 1)
          decompress-batch-ch (chan 1)
          apply-fn-ch (chan 1)
          compress-batch-ch (chan 1)
