         (do (set! iter (.iterator ^ConsumerRecords (.poll ^Consumer consumer remaining-ms)))
             nil))))) 
 
-(defn read-messages [{:keys [onyx.core/task-map onyx.core/log-prefix onyx.core/monitoring] :as event}]
+(defn read-messages [{:keys [onyx.core/task-map onyx.core/log-prefix onyx.core/monitoring]}]
   (let [{:keys [kafka/topic kafka/deserializer-fn kafka/target-offsets]} task-map
         batch-timeout (arg-or-default :onyx/batch-timeout task-map)
         wrap-message? (or (:kafka/wrap-with-metadata? task-map) (:kafka/wrap-with-metadata? defaults))

  (onCompletion [_ v exception]
    (when exception (reset! e exception))))

(defn write-messages [{:keys [onyx.core/task-map onyx.core/log-prefix] :as event}]
  (let [_ (s/validate onyx.tasks.kafka/KafkaOutputTaskMap task-map)
        brokers (or (:kafka/bootstrap-servers task-map) (find-brokers task-map))
        request-size (or (get task-map :kafka/request-size) (get write-defaults :kafka/request-size))

(ns onyx.plugin.kafka
  (:require [onyx.plugin.partition-assignment :refer [partitions-for-slot]]
            [onyx.kafka.helpers :as h]
            [taoensso.timbre :as log :refer [fatal info]]
            [onyx.static.default-vals :refer [arg-or-default]]
            [onyx.plugin.protocols :as p]
            [onyx.static.util :refer [kw->fn]]
            [onyx.tasks.kafka]
            [schema.core :as s])
  (:import [java.util.concurrent.atomic AtomicLong]
           [org.apache.kafka.clients.consumer ConsumerRecords ConsumerRecord]
           [org.apache.kafka.clients.consumer KafkaConsumer ConsumerRebalanceListener Consumer]
