   http://kafka.apache.org/082/javadoc/org/apache/kafka/clients/producer/package-summary.html"}
   clj-kafka.new.producer
   (:refer-clojure :exclude [send])
-  (:import [org.apache.kafka.clients.producer Callback KafkaProducer ProducerRecord RecordMetadata]
+  (:import [java.util.concurrent Future TimeUnit TimeoutException]
+           [org.apache.kafka.clients.producer Callback KafkaProducer ProducerRecord RecordMetadata]
            [org.apache.kafka.common.serialization Serializer ByteArraySerializer StringSerializer]))
 
 (defn string-serializer [] (StringSerializer.))

   :partition (.partition m)
   :offfset (.offset m)})

(defn send
  "Asynchronously send a record to Kafka. Returns a map with `:topic`,
  `:partition` and `:offset` keys, wrapped in a delay. Derefing delay
  will block until the underlying future completes. Optionally provide
  a callback fn that will be called when the operation completes.
  Callback should be a fn of two arguments, a map as above, and an
  exception. Exception will be nil if operation succeeded.

  For details on behaviour, see http://kafka.apache.org/082/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html#send(org.apache.kafka.clients.producer.ProducerRecord,org.apache.kafka.clients.producer.Callback)"
  ([^KafkaProducer producer record]
   (let [fut (.send producer record)]
     (map-future-val fut record-metadata->map)))
  ([^KafkaProducer producer record callback]
   (let [fut (.send producer record (reify Callback
                                      (onCompletion [_ metadata exception]
                                        (callback (and metadata (record-metadata->map metadata)) exception))))]
     (map-future-val fut record-metadata->map))))
