         task-id (:onyx.core/task-id event)
         reader-fut (future (reader-loop m client-id group-id topic partition partitions task-map 
                                         (:onyx.core/replica event) job-id peer-id task-id
-                                        ch pending-commits))]
+                                        ch pending-commits))
+        done-unsupported? (and (> (count partitions) 1)
+                               (not (:kafka/partition task-map)))]
+
     {:kafka/read-ch ch
      :kafka/reader-future reader-fut
      :kafka/pending-messages pending-messages
-     :kafka/pending-commits pending-commits}))
+     :kafka/pending-commits pending-commits
+     :kafka/done-unsupported? done-unsupported?}))
 
 (defn all-done? [messages]
   (empty? (remove #(= :done (:message %))
                   messages)))
 
-(defrecord KafkaReadMessages [max-pending batch-size batch-timeout partitions done-unsupported? 
-                              pending-messages pending-commits drained? read-ch]
+(defrecord KafkaReadMessages
+    [max-pending batch-size batch-timeout pending-messages
+     pending-commits drained? read-ch]
   p-ext/Pipeline
   (write-batch
     [this event]

        (swap! pending-messages assoc (:id m) m))
      (when (and (all-done? (vals @pending-messages))
                 (all-done? batch)
		 (zero? (count (.buf read-ch)))
                 (or (not (empty? @pending-messages))
                     (not (empty? batch))))
        (if done-unsupported? 
          (throw (UnsupportedOperationException. ":done is not supported for auto assigned kafka partitions. 
                                                 (:kafka/partition must be supplied)"))
          (reset! drained? true)))

        batch-timeout (arg-or-default :onyx/batch-timeout catalog-entry)
        chan-capacity (or (:kafka/chan-capacity catalog-entry) (:kafka/chan-capacity defaults))
        m {"zookeeper.connect" (:kafka/zookeeper catalog-entry)}
        ch (chan chan-capacity)
        pending-messages (atom {})
        pending-commits (atom (sorted-set))
        drained? (atom false)]
    (->KafkaReadMessages
     max-pending batch-size batch-timeout
     pending-messages pending-commits drained? ch)))

(defn close-read-messages
  [{:keys [kafka/read-ch] :as pipeline} lifecycle]
