             (Thread/sleep 500)
             (recur @replica)))
 
-        (let [release-messages-ch (release-messages! messenger pipeline-data)
-              retry-messages-ch (retry-messages! messenger pipeline-data)
-              forward-completion-ch (thread (forward-completion-calls! pipeline-data completion-ch))
+        (let [cleanup-ch (chan 1)
+              release-messages-ch (release-messages! messenger pipeline-data cleanup-ch)
+              retry-messages-ch (retry-messages! messenger pipeline-data cleanup-ch)
+              forward-completion-ch (forward-completion-calls! pipeline-data completion-ch cleanup-ch)
               task-lifecycle-ch (thread (run-task-lifecycle pipeline-data seal-resp-ch kill-ch ex-f))
               listen-for-sealer-ch (listen-for-sealer job-id task-id pipeline-data seal-resp-ch outbox-ch)]
           (assoc component 
+                 :cleanup-ch cleanup-ch
                  :pipeline-data pipeline-data 
                  :seal-ch seal-resp-ch
                  :release-messages-ch release-messages-ch

(defn close-batch-resources [event]
  (merge event (l-ext/close-batch-resources* event)))

(defn release-messages! [messenger event]
  (go
   (loop []
     (when-let [id (<! (:release-ch messenger))]
       (p-ext/ack-message event id)
       (recur)))))

(defn retry-messages! [messenger event]
  (go
   (loop []
     (when-let [id (<! (:retry-ch messenger))]
       (p-ext/retry-message event id)
       (recur)))))

(defn forward-completion-calls! [event completion-ch]
  (try
    (loop []
      (when-let [{:keys [id peer-id]} (<!! completion-ch)]
        (let [peer-link (operation/peer-link event peer-id)]
          (extensions/internal-complete-message (:onyx.core/messenger event) event id peer-link)
          (recur))))
    (catch Exception e
      (timbre/fatal e))))

(defn handle-exception [e restart-ch outbox-ch job-id]
  (warn e)

      (l-ext/close-lifecycle-resources* event)

      (close! (:seal-ch component))
      (close! (:cleanup-ch component))
      
      ;; Ensure task operations are finished before closing peer connections
      (<!! (:task-lifecycle-ch component))
      (<!! (:listen-for-sealer-ch component))
      (<!! (:release-messages-ch component))
      (<!! (:retry-messages-ch component))

