 
   (stop [this event]
     (when consumer
-      (.close ^KafkaConsumer consumer)
+      (.close consumer)
       (set! consumer nil))
     this)
 

(defn all-partitions-paused?
  [^KafkaConsumer consumer kpartitions]
  (let [paused (into #{}
                     (map #(.partition %))
                     (.paused consumer))]
    (= paused (set kpartitions))))

(deftype KafkaReadMessages
    [log-prefix task-map topic ^:unsynchronized-mutable kpartitions batch-timeout
     deserializer-fn segment-fn ^AtomicLong watermark ^AtomicLong lag-gauge ^:unsynchronized-mutable consumer
     ^:unsynchronized-mutable iter ^:unsynchronized-mutable partition->offset ^:unsynchronized-mutable drained
     target-offsets]
  p/Plugin
  (start [this event]
    (let [{:keys [kafka/bootstrap-servers kafka/group-id kafka/consumer-opts]} task-map

        (.set watermark (max (.get watermark) (.timestamp rec)))

        (cond (= :done deserialized) ;; TODO: Remove this in favor of target-offsets
              (do (set! drained true) 
                  nil)

              deserialized
              (let [new-offset (.offset rec)
