                          (doseq [t to-ack]
                            (.ack @collector t)))))
 
+       (bind fail-tx! (fn [txid]
+                        (let [[to-fail not-to-fail] (separate
+                                                     #(-> %
+                                                          (.getValue 0)
+                                                          .getTransactionId
+                                                          (= txid))
+                                                     @tuples)]
+                          (reset! tuples not-to-fail)
+                          (doseq [t to-fail]
+                            (.fail @collector t)))))
+
        ;; only check default streams
        (bind verify! (fn [expected]
                        (let [results (-> topo-info :capturer .getResults)]

     (RegisteredGlobalState/clearState id#)
    ))

;; cluster-map topology :mock-sources {} :storm-conf {}
(deftest test-transactional-topology
  (with-tracked-cluster [cluster]
    (with-controller-bolt [controller collector tuples]

                         )))

       (tracked-wait topo-info 2)
       (verify! {"sum" [[1 "dog" 3]
                        [1 "cat" 5]
                        [1 "mango" 6]
