         broker (get (id->broker m) (first brokers))
         consumer (kc/consumer (:host broker) (:port broker) client-id)
         offset (starting-offset consumer topic partition group-id task-map)
-        messages (kc/messages consumer "onyx" topic partition offset fetch-size)]
+        messages (kc/messages consumer "onyx" topic partition offset fetch-size)
+        ch (chan (sliding-buffer chan-capacity))
+        pending-messages (atom {})
+        pending-commits (atom (sorted-set))
+        reader-fut (future
+                     (try
+                       (log/debug "Opening Kafka resource " task-map)
+                       (loop [ms messages]
+                         (if (first ms)
+                           (>!! ch {:message (.payload ^Message (first ms))
+                                    :offset (.offset ^int (first ms))})
+                           (Thread/sleep (:kafka/empty-read-back-off task-map)))
+                         (recur (rest ms)))
+                       (catch InterruptedException e)
+                       (catch Throwable e
+                         (fatal e))))
+        commit-fut (future
+                     (try
+                       (loop []
+                         (Thread/sleep (:kafka/commit-interval task-map))
+                         (let [offset (highest-offset-to-commit @(pending-commits))]
+                           (kzk/set-offset! m consumer topic partition offset)
+                           (swap! pending-commits (fn [coll] (remove (fn [k] (<= k offset)) coll)))
+                           (recur)))
+                       (catch InterruptedException e)
+                       (catch Throwable e
+                         (fatal e))))]
     {:kafka/consumer consumer
-     :kafka/messages messages}))
-
-(defn inject-read-messages
-  [{:keys [onyx.core/task-map] :as pipeline} lifecycle]
-  (let [config {"zookeeper.connect" (:kafka/zookeeper task-map)
-                "group.id" (:kafka/group-id task-map)
-                "auto.offset.reset" (:kafka/offset-reset task-map)
-                "auto.commit.enable" "true"}
-        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))]
-                           (if (= (:value (first ms)) :done)
-                             (>!! ch :done)
-                             (>!! ch {:content (:value (first ms))}))
-                           (recur (rest ms))))
-                       (catch InterruptedException e)
-                       (catch Exception e
-                         (fatal e)))
-                     (log/debug "Stopping Kafka consumer and cleaning up"))
+     :kafka/messages messages
      :kafka/read-ch ch
-     :kafka/pending-messages (atom {})}))
+     :kafka/reader-future reader-fut
+     :kafka/commit-future commit-fut
+     :kafka/pending-messages pending-messages
+     :kafka/pending-commits pending-commits}))
 
 (defmethod p-ext/read-batch :kafka/read-messages
   [{:keys [kafka/read-ch kafka/pending-messages onyx.core/task-map] :as event}]

            [cheshire.core :refer [parse-string]]
            [zookeeper :as zk]
            [onyx.peer.pipeline-extensions :as p-ext]
            [taoensso.timbre :as log :refer [fatal]]))

(defn id->broker [m]
  (k/with-resource [z (zk/connect (get m "zookeeper.connect"))]

        (kc/topic-offset consumer topic partition :latest)
        :else (kzk/committed-offset m group-id topic (str partition))))


(defn highest-offset-to-commit [offsets]
  (->> (partition-all 2 1 offsets)
       (partition-by #(- (or (second %) (first %)) (first %)))
       (first)
       (last)
       (last)))

(defn start-kafka-consumer
  [{:keys [onyx.core/task-map] :as event} lifecycle]
  (let [{:keys [kafka/topic kafka/partition kafka/group-id fetch-size chan-capacity]} task-map
        partition (str partition)
        client-id "onyx"
        m {"zookeeper.connect" (:kafka/zookeeper task-map)}
