         (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]} task-map
+  (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)}

        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)]
    {: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/read-ch ch
     :kafka/pending-messages (atom {})}))

(defmethod p-ext/read-batch :kafka/read-messages
  [{:keys [kafka/read-ch kafka/pending-messages onyx.core/task-map] :as event}]


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

