         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

  (retry-segment
    [_ _ segment-id]
    (when-let [msg (get @pending-messages segment-id)]
      (>!! read-ch (t/input (random-uuid) (:message msg)))
      (swap! pending-messages dissoc segment-id)))

  (pending?


(defrecord KafkaReadMessages
    [max-pending batch-size batch-timeout pending-messages
     pending-commits drained? read-ch retry-ch]
  p-ext/Pipeline
  (write-batch
    [this event]
