        ;;       (println (read-tuples results "count"))
        (-> topo-info :capturer .getAndClearResults)
        ))))
+
+{0 [["dog" 3]
+    ["cat" 4]
+    ["apple" 1]
+    ["dog" 3]
+    ["banana" 0]]
+ 1 [["cat" 1]
+    ["mango" 4]]
+ 2 [["happy" 11]
+    ["mango" 2]
+    ["zebra" 1]]}

                        (.createTopology builder)))
       (submit-local-topology (:nimbus cluster)
                              "transactional-test"
                              {TOPOLOGY-MAX-SPOUT-PENDING 2
                               TOPOLOGY-DEBUG true}
                              (:topology topo-info))

       (bind ack-tx (fn [txid]
                      (let [[to-ack not-to-ack] (separate
                                                 #(-> %
                                                      (.getValue 0)
                                                      .getTransactionId
                                                      (= txid))
                                                 @tuples)]
                        (reset! tuples not-to-ack)
                        (doseq [t to-ack]
                          (.ack @collector t)))))

       ;; only check default streams
       (bind verify! (fn [expected]

        ))))

(defn normalize-tx-tuple [values]
  (-> values vec (update 0 #(-> % .getTransactionId .intValue))))

(defn verify-and-reset! [expected-map emitted-map-atom]
  (let [results @emitted-map-atom]
