         max-pending (or (:onyx/max-pending task-map) 10000)
         batch-size (:onyx/batch-size task-map)
         max-segments (min (- max-pending pending) batch-size)
-        ms (or (:onyx/batch-timeout task-map) 500)
+        ms (or (:onyx/batch-timeout task-map) 50)
         timeout-ch (timeout ms)
         batch (->> (range max-segments)
                    (map (fn [_]

              messages (kc/messages consumer "onyx" topic kpartition offset fetch-size)
              empty-read-back-off (or (:kafka/empty-read-back-off task-map) (:kafka/empty-read-back-off defaults))
              commit-interval (or (:kafka/commit-interval task-map) (:kafka/commit-interval defaults))
              commit-fut (future (commit-loop m topic kpartition commit-interval pending-commits))]
          (log/debug "Opening Kafka consumer" task-map)
          (log/debug (str "Kafka consumer is starting at offset " offset))
          (loop [ms messages]
            (if (first ms)
              (>!! ch {:message (read-string (String. (.value ^KafkaMessage (first ms)) "UTF-8"))
                       :offset (.offset ^int (first ms))})
              (Thread/sleep empty-read-back-off))
            (recur (rest ms))))
        (catch InterruptedException e
          (throw e))
        (catch Throwable e

       (last)))

(defn commit-loop [consumer m topic kpartition commit-interval pending-commits]
  (try
    (loop []
      (Thread/sleep commit-interval)
      (when-let [offset (highest-offset-to-commit @pending-commits)]
        (log/info "Writing offset to ZooKeeper: " offset)
        (kzk/set-offset! m consumer topic kpartition offset)
        (swap! pending-commits (fn [coll] (remove (fn [k] (<= k offset)) coll))))
      (recur))
    (catch InterruptedException e
      (throw e))
    (catch Throwable e
      (fatal e))))

(defn reader-loop [m client-id group-id topic partitions kpartition task-map ch pending-commits]
  (try
