         commit-interval (or (:kafka/commit-interval task-map) (:kafka/commit-interval defaults))
         client-id "onyx"
         zk-utils (k-admin/make-zk-utils {:servers (:kafka/zookeeper task-map)} false)
-        partitions (first (vals (k-partitions/partitions-for zk-utils [topic])))
-        n-partitions (count partitions)
-        _ (check-num-peers-equals-partitions task-map n-partitions)
         retry-ch (:retry-ch pipeline)
         pending-messages (:pending-messages pipeline)
         pending-commits (:pending-commits pipeline)
         job-id (:onyx.core/job-id event)
         peer-id (:onyx.core/id event)
         task-id (:onyx.core/task-id event)
-        done-unsupported? (and (> (count partitions) 1)
-                               (not (:kafka/partition task-map)))
-        kpartition (if partition
-                     (Integer/parseInt (str partition))
-                     (log-id @replica job-id peer-id task-id))
         consumer-config {:bootstrap.servers (find-brokers (:kafka/zookeeper task-map))
                          :group.id group-id
                          :enable.auto.commit false

            [franzy.clients.producer.types :refer [make-producer-record]]
            [franzy.clients.consumer.protocols :as proto]
            [franzy.clients.consumer.client :as consumer]
            [franzy.clients.consumer.protocols :refer [assign-partitions! commit-offsets-sync!
                                                       poll! seek-to-offset!] :as cp]
            [taoensso.timbre :as log :refer [fatal info]]

;; kafka operations

(defn id->broker [zk-addr]
  (with-open [zk-utils (k-admin/make-zk-utils {:servers zk-addr} false)]
    (reduce
     (fn [result {:keys [id endpoints]}]
       (assoc
