tests for throwing failedbatchexception in finishedId

Nathan Marz · Nathan Marz · commit 941ef6362ba7 · 2012-01-26T19:23:29.000-08:00
diff --git a/test/clj/backtype/storm/drpc_test.clj b/test/clj/backtype/storm/drpc_test.clj
@@ -2,9 +2,10 @@
   (:use [clojure test])
   (:import [backtype.storm.drpc ReturnResults DRPCSpout
             LinearDRPCTopologyBuilder])
-  (:import [backtype.storm.coordination CoordinatedBolt$FinishedCallback])
+  (:import [backtype.storm.coordination CoordinatedBolt$FinishedCallback FailedBatchException])
   (:import [backtype.storm LocalDRPC LocalCluster])
   (:import [backtype.storm.tuple Fields])
+  (:import [backtype.storm.generated DRPCExecutionException])
   (:use [backtype.storm bootstrap testing])
   (:use [backtype.storm.daemon common])
   (:use [backtype.storm clojure])
@@ -29,7 +30,7 @@
                                 exclamation-bolt)
                    "3" (bolt-spec {"2" :shuffle}
                                 (ReturnResults.))})]
-    (.submitTopology cluster "test" {TOPOLOGY-DEBUG true} topology)
+    (.submitTopology cluster "test" {} topology)
 
     (is (= "aaa!!!" (.execute drpc "test" "aaa")))
     (is (= "b!!!" (.execute drpc "test" "b")))
@@ -55,7 +56,7 @@
     (.addBolt builder exclamation-bolt-drpc 3)
     (.submitTopology cluster
                      "builder-test"
-                     {TOPOLOGY-DEBUG true}
+                     {}
                      (.createLocalTopology builder drpc))
     (is (= "aaa!!!" (.execute drpc "test" "aaa")))
     (is (= "b!!!" (.execute drpc "test" "b")))
@@ -173,3 +174,31 @@
     (.shutdown cluster)
     (.shutdown drpc)
     ))
+
+(defbolt fail-finish-bolt ["request" "result"] {:prepare true}
+  [conf context collector]
+  (bolt
+   (execute [tuple]
+            (ack! collector tuple))
+   CoordinatedBolt$FinishedCallback
+   (finishedId [this id]
+               (throw (FailedBatchException.))
+               )))
+
+(deftest test-drpc-fail-finish
+  (let [drpc (LocalDRPC.)
+        cluster (LocalCluster.)
+        builder (LinearDRPCTopologyBuilder. "fail2")
+        ]
+    (.addBolt builder fail-finish-bolt 3)
+
+    (.submitTopology cluster
+                     "fail2"
+                     {}
+                     (.createLocalTopology builder drpc))
+    
+    (is (thrown? DRPCExecutionException (.execute drpc "fail2" "2")))
+
+    (.shutdown cluster)
+    (.shutdown drpc)
+    ))
