test topology for transactional topologies

Nathan Marz · Nathan Marz · commit 7989c581876f · 2012-01-25T23:45:51.000-08:00
diff --git a/src/jvm/backtype/storm/testing/CountingBatchBolt.java b/src/jvm/backtype/storm/testing/CountingBatchBolt.java
@@ -3,7 +3,6 @@
 import backtype.storm.task.TopologyContext;
 import backtype.storm.topology.OutputFieldsDeclarer;
 import backtype.storm.topology.base.BaseBatchBolt;
-import backtype.storm.transactional.TransactionAttempt;
 import backtype.storm.coordination.BatchOutputCollector;
 import backtype.storm.tuple.Fields;
 import backtype.storm.tuple.Tuple;
diff --git a/src/jvm/backtype/storm/testing/IdentityBolt.java b/src/jvm/backtype/storm/testing/IdentityBolt.java
@@ -0,0 +1,25 @@
+package backtype.storm.testing;
+
+import backtype.storm.topology.BasicOutputCollector;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.topology.base.BaseBasicBolt;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Tuple;
+
+public class IdentityBolt extends BaseBasicBolt {
+    Fields _fields;
+    
+    public IdentityBolt(Fields fields) {
+        _fields = fields;
+    }
+
+    @Override
+    public void execute(Tuple input, BasicOutputCollector collector) {
+        collector.emit(input.getValues());
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer declarer) {
+        declarer.declare(_fields);
+    }    
+}
diff --git a/test/clj/backtype/storm/transactional_test.clj b/test/clj/backtype/storm/transactional_test.clj
@@ -5,7 +5,8 @@
             TransactionalTopologyBuilder])
   (:import [backtype.storm.transactional.state TransactionalState RotatingTransactionalState RotatingTransactionalState$StateInitializer])
   (:import [backtype.storm.testing CountingBatchBolt MemoryTransactionalSpout
-            KeyedCountingBatchBolt KeyedCountingCommitterBolt KeyedSummingBatchBolt])
+            KeyedCountingBatchBolt KeyedCountingCommitterBolt KeyedSummingBatchBolt
+            IdentityBolt])
   (:use [backtype.storm bootstrap testing])
   (:use [backtype.storm.daemon common])  
   )
@@ -342,16 +343,34 @@
                                                  (Fields. ["word" "amt"])
                                                  2)
                       2))
+
+       (-> builder
+           (.setBolt "id1" (IdentityBolt. (Fields. ["tx" "word" "amt"])) 3)
+           (.shuffleGrouping "spout"))
+
+       (-> builder
+           (.setBolt "id2" (IdentityBolt. (Fields. ["tx" "word" "amt"])) 3)
+           (.shuffleGrouping "spout"))
        
        (-> builder
-           (.setBolt "count" (KeyedSummingBatchBolt.) 2)
-           (.fieldsGrouping "spout" (Fields. ["word"])))
+           (.setBolt "sum" (KeyedSummingBatchBolt.) 2)
+           (.fieldsGrouping "id1" (Fields. ["word"])))
+
+       (-> builder
+           (.setBolt "count" (KeyedCountingCommitterBolt.) 2)
+           (.fieldsGrouping "id2" (Fields. ["word"])))
+
+       (-> builder
+           (.setBolt "count2" (KeyedCountingCommitterBolt.) 3)
+           (.fieldsGrouping "sum" (Fields. ["key"]))
+           (.fieldsGrouping "count" (Fields. ["key"])))
 
        (bind builder (.buildTopologyBuilder builder))
        
        (-> builder
            (.setBolt "controller" controller 1)
-           (.directGrouping "count" Constants/COORDINATED_STREAM_ID))
+           (.directGrouping "count2" Constants/COORDINATED_STREAM_ID)
+           (.directGrouping "sum" Constants/COORDINATED_STREAM_ID))
 
        (add-transactional-data data
                                {0 [["dog" 3]
