         key-deserializer (byte-array-deserializer)
         value-deserializer (byte-array-deserializer)
         consumer (consumer/make-consumer consumer-config key-deserializer value-deserializer)
+        kpartition (if partition
+                     (Integer/parseInt (str partition))
+                     (log-id @replica job-id peer-id task-id))
+        partitions (mapv :partition (metadata/partitions-for consumer topic))
+        n-partitions (count partitions)
+        done-unsupported? (and (> (count partitions) 1)
+                               (not (:kafka/partition task-map)))
+        _ (check-num-peers-equals-partitions task-map n-partitions)
         _ (assign-partitions! consumer [{:topic topic :partition kpartition}])
         _ (seek-offset! log consumer group-id topic kpartition task-map)
         offset (cp/next-offset consumer {:topic topic :partition kpartition})
-        poll-timeout-ms (or (:kafka/poll-timeout-ms task-map) (:kafka/poll-timeout-ms defaults))
         commit-interval (or (:kafka/commit-interval task-map) (:kafka/commit-interval defaults))
         commit-fut (future (commit-loop log group-id topic kpartition commit-interval pending-commits))]
     {:kafka/commit-fut commit-fut

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

;; 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
