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

        (.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)

(defn all-partitions-paused?
  [^KafkaConsumer consumer kpartitions]
  (let [paused (into #{}
                     (map #(.partition ^org.apache.kafka.common.PartitionInfo %))
                     (.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 ^KafkaConsumer ^:unsynchronized-mutable consumer
     ^:unsynchronized-mutable iter ^:unsynchronized-mutable partition->offset ^:unsynchronized-mutable drained
     target-offsets]
  PluginMeta
  (metadata [this]
    {:consumer consumer 
     :beginning-offsets (into {} 
                              (map (fn [[topic-partition offset]]
                                     [{:topic (.topic topic-partition)
                                       :partition (.partition topic-partition)}
                                      offset]) 
                                   (.beginningOffsets consumer (.assignment consumer))))
     :end-offsets (into {} 
                        (map (fn [[topic-partition offset]]
                               [{:topic (.topic topic-partition)
                                 :partition (.partition topic-partition)}
                                offset]) 
                             (.endOffsets consumer (.assignment consumer))))
     :partitions kpartitions 
     :partition->offset partition->offset})
  p/Plugin
  (start [this event]
    (let [{:keys [kafka/bootstrap-servers kafka/group-id kafka/consumer-opts]} task-map
