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

      (when (and (all-done? (vals @pending-messages))
                 (all-done? batch)
                 (zero? (count (.buf read-ch)))
                 (or (not (empty? @pending-messages))
                     (not (empty? batch))))
        (if (:kafka/done-unsupported? event)


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