         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)

      (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)

(ns onyx.plugin.kafka
  (:require [clojure.core.async :as a :refer [chan >!! <!! close! timeout alts!! sliding-buffer]]
            [franzy.admin.zookeeper.client :as k-admin]
            [franzy.admin.cluster :as k-cluster]
            [franzy.admin.partitions :as k-partitions]
