         (fn [e & _] (.printStackTrace e)))
 
       (assoc component
+        :open-session-kill-ch open-session-kill-ch
         :read-batch-ch read-batch-ch
         :decompress-batch-ch decompress-batch-ch
         :apply-fn-ch apply-fn-ch

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

      (extensions/touch-place sync completion-node))
    (assoc event :completed done?)))

(defn open-session-loop [read-ch kill-ch pipeline-data]
  (loop []
    (when (first (alts!! [kill-ch] :default true))
      (when-let [session (create-tx-session pipeline-data)]
        (>!! read-ch (munge-open-session pipeline-data session))
        (recur)))))

(defn read-batch-loop [read-ch decompress-ch]
  (loop []
