                             (apply [this key value]
                               (println key value)
                               ))))
-        streams (KafkaStreams. builder props)]
+        streams (KafkaStreams. builder count-words-props)]
 
     (.start streams)
     (Thread/sleep 5000)


(require '[clojure.string :as str])

(defn client-props []
  (let [props (java.util.Properties.)]
    (.put props "bootstrap.servers" "kafka:9092")
    (.put props "acks" "all")
    (.put props "retries" (int 0))
    (.put props "batch.size" (int 16384))
    (.put props "linger.ms" (int 1))
    (.put props "buffer.memory" (int 33554432))
    (.put props "key.serializer" "org.apache.kafka.common.serialization.StringSerializer")
    (.put props "value.serializer" "org.apache.kafka.common.serialization.StringSerializer")

    props))

(defn produce-numbered-messages
  [& args]
  (let [topic "my-topic"
        props (client-props)
        producer (KafkaProducer. props)]
    (doseq
        [n (range 100)]
      (.send producer

                   .groupByKey
                   (.count "Counts"))
        xxx (.to counts (Serdes/String) (Serdes/Long) "streams-wordcount-output")
        streams (KafkaStreams. builder count-words-props)]

    (.start streams)
    (Thread/sleep 5000)
    (.close streams)))

(defn print-word-counts[& args]
  (let [builder (KStreamBuilder.)
        source (.stream builder (Serdes/String) (Serdes/Long)
                        (into-array String ["streams-wordcount-output"]))
        print-counts (-> source
