               part (.partition rec)]
           (.set watermark (max (.get watermark) (.timestamp rec)))
 
-          (cond (= :done deserialized) ;; TODO: Remove this in favor of target-offsets
+          ;; TODO: remove support for :done in favor of target-offsets
+          (cond (= :done deserialized)
                 (do (swap! drained assoc (.partition rec) :emitted)
                     nil)
 

        (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}]
  (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))

(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]
