An example of clj-kafka 0.2.6-0.8 library usage

kafka_client.clj

(ns kafka-client
  "An example of clj-kafka 0.2.6-0.8 library usage"
  (:require [clj-kafka.core :as kafka]
            [clj-kafka.zk :as zk]
            [clj-kafka.consumer.zk :as zk-consumer]
            [clj-kafka.producer :as producer]
            [clojure.test :refer [is deftest run-tests]]
            [clojure.edn :as edn]))

(def zk-config {"zookeeper.connect" "localhost:2181"})

(deftest sending-a-message-from-producer-to-consumer
  (let [{:keys [host port]} ((comp first zk/brokers) zk-config)
        kafka-broker-address (str host ":" port)
        message {:some "data"}
        topic "test-topic"]
    (with-open [kafka-producer (producer/producer
                                {"metadata.broker.list" kafka-broker-address})]
      (->> message
           prn-str
           (.getBytes)
           (producer/message topic)
           (producer/send-message kafka-producer)))
    (kafka/with-resource [kafka-consumer
                          (zk-consumer/consumer
                           (merge zk-config
                                  {"group.id" "clj-kafka.consumer"}))]
      zk-consumer/shutdown
      (->> topic
           (zk-consumer/messages kafka-consumer)
           first
           :value
           (String.)
           edn/read-string
           (= message)
           is))))

(run-tests *ns*)