         (is (= (.value record) "all streams lead to kafka"))))))
 
 (deftest test-word-count-topology
-  (testing "It works"
+  (testing "Complete table written to output topic"
     (let [builder (KStreamBuilder.)
           _ (build-count-words builder)
           driver (ProcessorTopologyTestDriver. count-words-props builder

                string-serializer string-serializer)
      (.process driver "streams-file-input" "" "join kafka summit"
                string-serializer string-serializer)
      (let [record (.readOutput driver "streams-wordcount-output")]
        (println (-> driver (.getKeyValueStore "Counts") .all iterator-seq))
        (is (some? record))))))

  (:require [clojure.test :refer :all]
            [clstreams.core :refer :all])
  (:import [org.apache.kafka.test ProcessorTopologyTestDriver]
           [org.apache.kafka.common.serialization Serdes]
           [org.apache.kafka.streams.kstream KStreamBuilder]))

(def string-serializer (-> (Serdes/String) .serializer))
(def string-deserializer (-> (Serdes/String) .deserializer))

(def long-deserializer (-> (Serdes/Long) .deserializer))

(defn key-value-map [key-value-objs]
  (into {} (map #(vector (.key %) (.value %)) key-value-objs)))

(defn collect-output [record-producing-fn]
  (->> record-producing-fn repeatedly (take-while identity) key-value-map))

(deftest test-copy-topology
  (testing "Garbage in, garbage out..."
    (let [builder (KStreamBuilder.)
          _ (build-copy builder)
          driver (ProcessorTopologyTestDriver. count-words-props builder
