         done-unsupported? (and (> (count partitions) 1)
                                (not (:kafka/partition task-map)))]
     {:kafka/read-ch ch
+     :kafka/retry-ch ch
      :kafka/reader-future reader-fut
      :kafka/pending-messages pending-messages
      :kafka/pending-commits pending-commits

        batch-timeout (arg-or-default :onyx/batch-timeout catalog-entry)
        chan-capacity (or (:kafka/chan-capacity catalog-entry) (:kafka/chan-capacity defaults))
        ch (chan chan-capacity)
        pending-messages (atom {})
        pending-commits (atom (sorted-set))
        drained? (atom false)]
    (->KafkaReadMessages
     max-pending batch-size batch-timeout
     pending-messages pending-commits drained? ch)))

(defn close-read-messages
  [{:keys [kafka/read-ch] :as pipeline} lifecycle]
  (future-cancel (:kafka/reader-future pipeline))
  (close! read-ch)
  ;; Drain read buffer to be safe, otherwise reader channel will stay blocked on put
  (while (clojure.core.async/poll! read-ch))
  {})

(defn inject-write-messages

        n-partitions (count partitions)
        _ (check-num-peers-equals-partitions task-map n-partitions)
        ch (:read-ch pipeline)
        retry-ch (:retry-ch pipeline)
        pending-messages (:pending-messages pipeline)
        pending-commits (:pending-commits pipeline)
        job-id (:onyx.core/job-id event)
