               :message {:n 0}})
 (>!! in-chan {:message {:n 1}})
 (>!! in-chan {:key "tarein"
-              :partition 1
               :message {:n 2}})
 (>!! in-chan :done)
 

  (fact (->> messages
             (sort-by (comp :n :value))
             (map (fn [msg] 
                    (select-keys (if-not (= "tarein" (:key msg)) 
                                   (assoc msg :partition nil)
                                   msg) 
                                 [:key :value :partition]))))
        => [{:key 1
             :partition nil
             :value {:n 0}} 
            {:key nil
             :partition nil
             :value {:n 1}} 
            {:key "tarein" 
             :partition 1
             :value {:n 2}}]))

(doseq [v-peer v-peers]


(with-open [zk (kadmin/zk-client zk-addr)]
  (kadmin/create-topic zk topic
                       {:partitions 1}))

(def workflow
  [[:in :identity]
