             [cheshire.core :refer [parse-string]]
             [zookeeper :as zk]
             [onyx.peer.pipeline-extensions :as p-ext]
-            [taoensso.timbre :as log :refer [fatal]]))
+            [taoensso.timbre :as log :refer [fatal]])
+  (:import [kafka.message Message]))
 
 (defn id->broker [m]
   (k/with-resource [z (zk/connect (get m "zookeeper.connect"))]


(defn close-read-messages
  [{:keys [kafka/read-ch] :as pipeline} lifecycle]
  (future-cancel (:kafka/future pipeline))
  (close! read-ch)
  {})


        timeout-ch (timeout ms)
        batch (->> (range max-segments)
                   (map (fn [_]
                          (let [result (first (alts!! [read-ch timeout-ch] :priority true))
                                ]
                            {:id (java.util.UUID/randomUUID)
                             :input :kafka
                             :message (read-string (String. (:message result) "UTF-8"))
                             :offset (:offset result)})))
                   (remove (comp nil? :message)))]
    (doseq [m batch]
      (swap! pending-messages assoc (:id m) (select-keys m [:message :offset])))
    {:onyx.core/batch batch}))

(defmethod p-ext/ack-message :kafka/read-messages
  [{:keys [kafka/pending-messages kafka/pending-commits onyx.core/log onyx.core/task-id]} message-id]
  (swap! pending-commits conj (:offset (get pending-messages message-id)))
  (swap! pending-messages dissoc message-id))

(defmethod p-ext/retry-message :kafka/read-messages
