      (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)
       (println "Controlled: " @tuples)
       (println "Captured:" (-> topo-info :capturer .getResults))
       (verify! {"sum" [[1 "dog" 3]
                        [1 "cat" 5]
                        [1 "mango" 6]

                           [1 "cat" 2]
                           [1 "mango" 2]
                           [1 "happy" 2]]})

       (add-transactional-data data
                               {0 [["a" 1]
                                   ["b" 2]
                                   ["c" 3]]
                                1 [["d" 4]
                                   ["c" 1]]
                                2 [["a" 2]
                                   ["e" 7]
                                   ["c" 11]]
                                3 [["a" 2]]})
       
       (ack-tx! 1)
       (tracked-wait topo-info 1)
       (verify! {"sum" [[3 "a" 5]
                        [3 "b" 2]
                        [3 "d" 4]
                        [3 "c" 1]
                        [3 "e" 7]]
                 "count" []
                 "count2" []})
       (ack-tx! 3)
       (ack-tx! 2)
       (tracked-wait topo-info 1)
       (verify! {"sum" []
                 "count" [[2 "apple" 1]
                          [2 "dog" 1]
                          [2 "zebra" 1]]
                 "count2" [[2 "apple" 2]
                           [2 "dog" 2]
                           [2 "zebra" 2]]})

       (fail-tx! 2)
       (tracked-wait topo-info 1)

       (verify! {"sum" [[2 "apple" 1]
                        [2 "dog" 3]
                        [2 "zebra" 1]]
                 "count" []
                 "count2" []})
       (ack-tx! 2)
       (tracked-wait topo-info 1)
       
       (verify! {"sum" []
                 "count" [[2 "apple" 1]
                          [2 "dog" 1]
                          [2 "zebra" 1]]
                 "count2" [[2 "apple" 2]
                           [2 "dog" 2]
                           [2 "zebra" 2]]})
       
       (ack-tx! 2)

       (tracked-wait topo-info 2)
       (verify! {"sum" [[4 "c" 14]]
                 "count" [[3 "a" 3]
                          [3 "b" 1]
                          [3 "d" 1]
                          [3 "c" 1]
                          [3 "e" 1]]
                 "count2" [[3 "a" 2]
                           [3 "b" 2]
                           [3 "d" 2]
                           [3 "c" 2]
                           [3 "e" 2]]})

       (-> topo-info :capturer .getAndClearResults)
       ))))

;; TODO: should test that it commits even when receiving no tuples (and test that finishBatch is called before commit in this case)
;; TODO: ;; * Test that it picks up where it left off when restarting the topology
;;      - run topology and restart it
