         ch (chan (:kafka/chan-capacity task-map))]
     {:kafka/future (future
                      (try
+                       (log/debug "Opening Kafka resource " config)
                        (k/with-resource [c (zk/consumer config)]
                          zk/shutdown
                          (loop [ms (zk/messages c (:kafka/topic task-map))]

            [clj-kafka.core :as k]
            [onyx.peer.task-lifecycle-extensions :as l-ext]
            [onyx.peer.pipeline-extensions :as p-ext]
            [taoensso.timbre :refer [fatal]]))

(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)))
                     (log/debug "Stopping Kafka consumer and cleaning up"))
     :kafka/ch ch}))

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