             [clj-kafka.core :as k]
             [onyx.peer.task-lifecycle-extensions :as l-ext]
             [onyx.peer.pipeline-extensions :as p-ext]
-            [taoensso.timbre :refer [fatal]]))
+            [taoensso.timbre :refer [fatal] :as log]))
 
 (defmethod l-ext/inject-lifecycle-resources :kafka/read-messages
   [_ {:keys [onyx.core/task-map] :as pipeline}]

                           (recur (rest ms))))
                       (catch InterruptedException e)
                       (catch Exception e
                         (fatal e))))
     :kafka/ch ch}))

(defmethod p-ext/read-batch [:input :kafka]

  [{:keys [onyx.core/decompressed]}]
  {:onyx.core/results decompressed})

(defmethod l-ext/close-lifecycle-resources :kafka/read-messages
  [_ {:keys [kafka/ch] :as pipeline}]
  (future-cancel (:kafka/future pipeline))
  (close! ch)
  {})

(defmethod l-ext/inject-lifecycle-resources :kafka/write-messages
