         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 [_]


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


              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 consumer m topic kpartition commit-interval pending-commits))]
          (log/info "Opening Kafka consumer" task-map)
          (log/info (str "Kafka consumer is starting at offset " offset))
          (try
            (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)))
            (finally
             (future-cancel commit-fut))))
        (catch InterruptedException e
          (throw e))
        (catch Throwable e
