Allow to specify minimum bytes and maximum wait for simple consumer API

When requesting new messages with the simple consumer API and message is
available, by default, the call to `messages` will return immediatly. If
called in a loop to wait for new messages, a user will have to sleep
between each call to avoid a tight loop, increasing the latency. The
Java API exposes the ability to wait for a minimum of bytes and a
maximum of time before returning no record. We expose those knobs in
`fetch-request` and `messages`.

src/clj_kafka/consumer/simple.clj

@@ -16,14 +16,16 @@
                    client-id))
 
 (defn fetch-request
-  [client-id topic ^Long partition offset fetch-size]
+  [client-id topic ^Long partition offset fetch-size & {:keys [max-wait min-bytes]}]
   (.build (doto (FetchRequestBuilder. )
             (.clientId client-id)
-            (.addFetch topic (Integer/valueOf partition) offset fetch-size))))
+            (.addFetch topic (Integer/valueOf partition) offset fetch-size)
+            (#(when max-wait (.maxWait % max-wait)))
+            (#(when min-bytes (.minBytes % min-bytes))))))
 
 (defn messages
-  [^SimpleConsumer consumer client-id topic partition offset fetch-size]
-  (let [fetch (fetch-request client-id topic partition offset fetch-size)]
+  [^SimpleConsumer consumer client-id topic partition offset fetch-size & more]
+  (let [fetch (apply fetch-request client-id topic partition offset fetch-size more)]
     (map to-clojure (iterator-seq (.iterator (.messageSet ^FetchResponse (.fetch consumer ^FetchRequest fetch)
                                                           topic
                                                           partition))))))