                    .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

(defn produce-words
  [& args]
  (let [topic "streams-file-input"
        props (client-props)
        producer (KafkaProducer. props)]
    (doto producer
      (.send (ProducerRecord. topic "" "all streams lead to kafka"))
      (.send (ProducerRecord. topic "" "hello kafka streams"))
      (.send (ProducerRecord. topic "" "join kafka summit"))
      .close)))

(defn count-words-props []
  (let [props (java.util.Properties.)]
    (.put props StreamsConfig/APPLICATION_ID_CONFIG, "streams-wordcount")
    (.put props StreamsConfig/BOOTSTRAP_SERVERS_CONFIG, "kafka:9092")
    (.put props StreamsConfig/KEY_SERDE_CLASS_CONFIG, (-> (Serdes/String) .getClass .getName))
    (.put props StreamsConfig/VALUE_SERDE_CLASS_CONFIG, (-> (Serdes/String) .getClass .getName))

    ; setting offset reset to earliest so that we can re-run the demo
    ; code with the same pre-loaded data
    ; Note: To re-run the demo, you need to use the offset reset tool:
    ; https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Application+Reset+Tool
    (.put props ConsumerConfig/AUTO_OFFSET_RESET_CONFIG, "earliest")

    props))

(defn count-words
  [& args]
  (let [props (count-words-props)
        builder (KStreamBuilder.)
        source (.stream builder (into-array String ["streams-file-input"]))
        counts (-> source
                   (.flatMapValues

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

    (.start streams)
    (Thread/sleep 5000)
