                    .groupByKey
                    (.count "Counts"))
         xxx (.to counts (Serdes/String) (Serdes/Long) "streams-wordcount-output")
-        streams (KafkaStreams. builder props)]
+        streams (KafkaStreams. builder count-words-props)]
 
     (.start streams)
     (Thread/sleep 5000)
     (.close streams)))
 
 (defn print-word-counts[& args]
-  (let [props (count-words-props)
-        builder (KStreamBuilder.)
+  (let [builder (KStreamBuilder.)
         source (.stream builder (Serdes/String) (Serdes/Long)
                         (into-array String ["streams-wordcount-output"]))
         print-counts (-> source

                            (apply [this key value]
                              (println key value)
                              ))))
        streams (KafkaStreams. builder props)]

    (.start streams)
    (Thread/sleep 5000)


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

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

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