 
 (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))
+(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"
-        props (client-props)
-        producer (KafkaProducer. props)]
+        producer (KafkaProducer. client-props)]
     (doseq
         [n (range 100)]
       (.send producer

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

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

(defn print-word-counts[& args]
  (let [props (count-words-props)
        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"
        producer (KafkaProducer. client-props)]
    (doto producer
      (.send (ProducerRecord. topic "" "all streams lead to kafka"))
      (.send (ProducerRecord. topic "" "hello kafka streams"))
      (.send (ProducerRecord. topic "" "join kafka summit"))
      .close)))

(def count-words-props
  (doto (java.util.Properties.)
    (.put StreamsConfig/APPLICATION_ID_CONFIG, "streams-wordcount")
    (.put StreamsConfig/BOOTSTRAP_SERVERS_CONFIG, "kafka:9092")
    (.put StreamsConfig/KEY_SERDE_CLASS_CONFIG, (-> (Serdes/String) .getClass .getName))
    (.put 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 ConsumerConfig/AUTO_OFFSET_RESET_CONFIG, "earliest")))

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