我是针对0.9中使用clojure中的javainterop的新kafka消费api编写的。到目前为止，我已经取得了成功，但现在我正在尝试用它编写一些单元测试 MockConsumer 以及 MockProducer . 我的考试总是不及格，因为 (first values) 是 nil .
我不明白为什么消费者看不到制作人发送给主题的任何信息。

(ns blah
  (:require [cheshire.core :as json]
            [clojure.test :refer [is testing deftest]])
  (:import [org.apache.kafka.clients.consumer MockConsumer OffsetResetStrategy]
           [org.apache.kafka.clients.producer MockProducer]
           [org.apache.kafka.common.serialization StringSerializer]
           [org.apache.kafka.clients.consumer KafkaConsumer ConsumerRecords ConsumerRecord]
           [org.apache.kafka.clients.producer KafkaProducer ProducerRecord]
           [java.util ArrayList]))

(defn send-message
  [producer topic value]
  (let [pr (ProducerRecord. topic value)]
    (.send producer pr)))

(defn messages
  "Return seq of messages from consumer."
  ([consumer] (messages consumer 100))
  ([consumer timeout]
   (println "poll consumer for messages")
   (let [records (seq (.poll consumer timeout))]
     (when records
       (map record->map records)))))

(deftest consuming
  (let [c (MockConsumer. (OffsetResetStrategy/EARLIEST))
        _ (.subscribe c (doto (ArrayList.) (.add "unittest")))
        p (MockProducer. true (StringSerializer.) (StringSerializer.))]
    (send-message p "unittest" (json/generate-string {:a 1 :b "two"}))
    (let [values (seq (messages c))]
      (is (= {:a 1 :b "two"}
          (first values))))
    (.close c)
    (.close p)))

有什么想法吗？