         (.set watermark (max (.get watermark) (.timestamp rec)))
 
         (cond (= :done deserialized) ;; TODO: Remove this in favor of target-offsets
-              (do (set! drained true) nil)
+              (do (set! drained true) 
+                  nil)
 
               deserialized
               (let [new-offset (.offset rec)

        (throw (ex-info "Could not locate any Kafka brokers to connect to."
                        {:recoverable? true
                         :zk-addr zk-addr}))))))

(defn start-kafka-consumer
  [event lifecycle]


  (stop [this event]
    (when consumer
      (.close consumer)
      (set! consumer nil))
    this)

