        (recur (dec n))))))
 
 (defrecord KafkaReadMessages
-  [max-pending batch-size batch-timeout pending-messages pending-commits drained? iter retry-ch segment-fn]
+  [task-map max-pending batch-size batch-timeout pending-messages pending-commits drained? iter retry-ch segment-fn]
   p-ext/Pipeline
   (write-batch
     [this event]

                        :n-peers n-peers
                        :min-peers min-peers
                        :max-peers max-peers
                        :task-map task-map})] 
        (log/error e)
        (throw e)))))

            (when-not offset
              (throw (ex-info "Offset missing for existing partition when using :kafka/start-offsets" 
                              {:missing-partition kpartition
                               :recoverable? false
                               :kafka/start-offsets start-offsets})))
            (seek-to-offset! consumer {:topic topic :partition kpartition} offset))
          (cp/seek-to-beginning-offset! consumer [{:topic topic :partition kpartition}]))))))
