我想用Kafka的 kafka.zk.EmbeddedZookeeper 和一个 kafka.server.KafkaServer 退回人 kafka.utils.TestUtils/createServer ，以运行kafka服务器进行测试。
但我遇到了一个路障，试图发送消息超时，并返回一个 KafkaProducer$Future 失败。下面是我正在使用的Kafka版本。下面的代码是clojure与kafka库的互操作。

[org.apache.kafka/kafka_2.11 "0.10.0.1"]
[org.apache.kafka/kafka-clients "0.10.1.0"]

这就是我要走的路。
zookeeper端口是随机分配的（参见此处）。
可以使用netcat成功创建zookeeper服务器并连接到它。
可以成功创建主题。
可以使用netcat成功创建kafka代理并连接到它。
第五步，就是流程失败的地方。
所以这个问题表明，传递一个正确的答案是很重要的 Time 对象。但mocktime看起来是一个合理的实现。以前有人处理过吗？

;; 1. Create Zookeeper
(require '[clojure.test :refer :all]
     '[kafkaesque.topics :as kt]
     '[kafkaesque.utils :as ku]
     '[clojure.pprint :refer [pprint]])

(import '[java.nio.file Files]
    '[kafka.zk EmbeddedZookeeper]
    '[kafka.server KafkaServer KafkaConfig]
    '[kafka.utils TestUtils Time MockTime])

(def zk-config {:zkhost "127.0.0.1"})
(def topic-name "client-test")
(def ^EmbeddedZookeeper zkServer (EmbeddedZookeeper.))

;; 2. Create Kafka Broker
(def zk-connect-str (str "127.0.0.1" ":" (.port zkServer)))
(def zku ((ZkUtils/apply (ZkUtils/createZkClient zk-connect-str 10000 8000) false)))
(def brokerhost "127.0.0.1")
(def brokerport "9092")

(def ^KafkaConfig config (KafkaConfig. {"zookeeper.connect" zk-connect-str
                       "broker.id" "0"
                       "log.dirs" (.toString
                           (.toAbsolutePath
                            (Files/createTempDirectory
                             "kafka-" (make-array java.nio.file.attribute.FileAttribute 0))))
                       "listeners" (str "PLAINTEXT://"  brokerhost  ":"  brokerport)}))

(def ^Time mock (MockTime.))
(def ^KafkaServer kafkaServer (TestUtils/createServer config mock))

;; 3. Create a Topic
(kt/create! zku topic-name 1 1 {})
(kt/topic-exists? zku topic-name)   ;; returns true

;; 4. Create a Producer and ProducerRecord
(def producer-a (kc/producer {"bootstrap.servers" "127.0.0.1:9092"
                 "acks"              "all"
                 "retries"           "0"
                 "batch.size"        "16384"
                 "linger.ms"         "1"
                 "buffer.memory"     "33554432"
                 "key.serializer"    "org.apache.kafka.common.serialization.StringSerializer"
                 "value.serializer"  "org.apache.kafka.common.serialization.StringSerializer"}))

(def message-key "k1")
(def message-value "foobar")
(def record-a (kc/producer-record topic-name 0 message-key message-value))

;; 5. Send a message
(def send-result (kc/send! producer-a record-a))  ;; Times out, and returns a KafkaProducer$Future failure.