     (.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)
+    ))

  (:use [clojure test])
  (:import [backtype.storm.drpc ReturnResults DRPCSpout
            LinearDRPCTopologyBuilder])
  (:import [backtype.storm.coordination CoordinatedBolt$FinishedCallback])
  (:import [backtype.storm LocalDRPC LocalCluster])
  (:import [backtype.storm.tuple Fields])
  (:use [backtype.storm bootstrap testing])
  (:use [backtype.storm.daemon common])
  (:use [backtype.storm clojure])

    (.addBolt builder exclamation-bolt-drpc 3)
    (.submitTopology cluster
                     "builder-test"
                     {}
                     (.createLocalTopology builder drpc))
    (is (= "aaa!!!" (.execute drpc "test" "aaa")))
    (is (= "b!!!" (.execute drpc "test" "b")))
