changed FailedBatchException to FailedException and moved it to topology namespace so it could be used with basicbolt as well

Nathan Marz · Nathan Marz · commit bfb22f946cb8 · 2012-01-27T20:49:06.000-08:00
diff --git a/src/jvm/backtype/storm/coordination/BatchBoltExecutor.java b/src/jvm/backtype/storm/coordination/BatchBoltExecutor.java
@@ -4,6 +4,7 @@
 import backtype.storm.coordination.CoordinatedBolt.TimeoutCallback;
 import backtype.storm.task.OutputCollector;
 import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.FailedException;
 import backtype.storm.topology.IRichBolt;
 import backtype.storm.topology.OutputFieldsDeclarer;
 import backtype.storm.tuple.Tuple;
@@ -40,7 +41,7 @@ public void execute(Tuple input) {
         try {
              bolt.execute(input);
             _collector.ack(input);
-        } catch(FailedBatchException e) {
+        } catch(FailedException e) {
             LOG.error("Failed to process tuple in batch", e);
             _collector.fail(input);                
         }
diff --git a/src/jvm/backtype/storm/coordination/CoordinatedBolt.java b/src/jvm/backtype/storm/coordination/CoordinatedBolt.java
@@ -1,5 +1,6 @@
 package backtype.storm.coordination;
 
+import backtype.storm.topology.FailedException;
 import java.util.Map.Entry;
 import backtype.storm.tuple.Values;
 import backtype.storm.generated.GlobalStreamId;
@@ -238,7 +239,7 @@ private boolean checkFinishId(Tuple tup) {
                     track.finished = true;
                     _tracked.remove(id);
                 }
-            } catch(FailedBatchException e) {
+            } catch(FailedException e) {
                 LOG.error("Failed to finish batch", e);
                 track.failed = true;
                 ret = true;
diff --git a/src/jvm/backtype/storm/coordination/FailedBatchException.java b/src/jvm/backtype/storm/coordination/FailedBatchException.java
diff --git a/src/jvm/backtype/storm/topology/BasicBoltExecutor.java b/src/jvm/backtype/storm/topology/BasicBoltExecutor.java
@@ -3,10 +3,11 @@
 import backtype.storm.task.OutputCollector;
 import backtype.storm.task.TopologyContext;
 import backtype.storm.tuple.Tuple;
-import java.util.HashMap;
 import java.util.Map;
+import org.apache.log4j.Logger;
 
 public class BasicBoltExecutor implements IRichBolt {
+    public static Logger LOG = Logger.getLogger(BasicBoltExecutor.class);    
     
     private IBasicBolt _bolt;
     private transient BasicOutputCollector _collector;
@@ -27,8 +28,13 @@ public void prepare(Map stormConf, TopologyContext context, OutputCollector coll
 
     public void execute(Tuple input) {
         _collector.setContext(input);
-        _bolt.execute(input, _collector);
-        _collector.getOutputter().ack(input);
+        try {
+            _bolt.execute(input, _collector);
+            _collector.getOutputter().ack(input);
+        } catch(FailedException e) {
+            LOG.warn("Failed to process tuple", e);
+            _collector.getOutputter().fail(input);
+        }
     }
 
     public void cleanup() {
diff --git a/src/jvm/backtype/storm/topology/FailedException.java b/src/jvm/backtype/storm/topology/FailedException.java
@@ -0,0 +1,19 @@
+package backtype.storm.topology;
+
+public class FailedException extends RuntimeException {
+    public FailedException() {
+        super();
+    }
+    
+    public FailedException(String msg) {
+        super(msg);
+    }
+    
+    public FailedException(String msg, Throwable cause) {
+        super(msg, cause);
+    }
+
+    public FailedException(Throwable cause) {
+        super(cause);
+    }
+}
diff --git a/src/jvm/backtype/storm/transactional/TransactionalSpoutBatchExecutor.java b/src/jvm/backtype/storm/transactional/TransactionalSpoutBatchExecutor.java
@@ -1,9 +1,9 @@
 package backtype.storm.transactional;
 
-import backtype.storm.coordination.FailedBatchException;
 import backtype.storm.coordination.BatchOutputCollectorImpl;
 import backtype.storm.task.OutputCollector;
 import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.FailedException;
 import backtype.storm.topology.IRichBolt;
 import backtype.storm.topology.OutputFieldsDeclarer;
 import backtype.storm.tuple.Tuple;
@@ -34,7 +34,7 @@ public void execute(Tuple input) {
         try {
             _emitter.emitBatch(attempt, input.getValue(1), _collector);
             _collector.ack(input);
-        } catch(FailedBatchException e) {
+        } catch(FailedException e) {
             LOG.warn("Failed to emit batch for transaction", e);
             _collector.fail(input);
         }
diff --git a/src/jvm/backtype/storm/transactional/TransactionalSpoutCoordinator.java b/src/jvm/backtype/storm/transactional/TransactionalSpoutCoordinator.java
@@ -1,9 +1,9 @@
 package backtype.storm.transactional;
 
-import backtype.storm.coordination.FailedBatchException;
 import backtype.storm.Config;
 import backtype.storm.spout.SpoutOutputCollector;
 import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.FailedException;
 import backtype.storm.topology.IRichSpout;
 import backtype.storm.topology.OutputFieldsDeclarer;
 import backtype.storm.transactional.state.RotatingTransactionalState;
@@ -137,7 +137,7 @@ private void sync() {
                     curr = nextTransactionId(curr);
                 }
             }     
-        } catch(FailedBatchException e) {
+        } catch(FailedException e) {
             LOG.warn("Failed to get metadata for a transaction", e);
         }
     }
diff --git a/test/clj/backtype/storm/drpc_test.clj b/test/clj/backtype/storm/drpc_test.clj
@@ -2,7 +2,8 @@
   (:use [clojure test])
   (:import [backtype.storm.drpc ReturnResults DRPCSpout
             LinearDRPCTopologyBuilder])
-  (:import [backtype.storm.coordination CoordinatedBolt$FinishedCallback FailedBatchException])
+  (:import [backtype.storm.topology FailedException])
+  (:import [backtype.storm.coordination CoordinatedBolt$FinishedCallback])
   (:import [backtype.storm LocalDRPC LocalCluster])
   (:import [backtype.storm.tuple Fields])
   (:import [backtype.storm.generated DRPCExecutionException])
@@ -182,7 +183,7 @@
             (ack! collector tuple))
    CoordinatedBolt$FinishedCallback
    (finishedId [this id]
-               (throw (FailedBatchException.))
+               (throw (FailedException.))
                )))
 
 (deftest test-drpc-fail-finish

Original file line number	Diff line number	Diff line change
`@@ -1,9 +1,9 @@`
`1`	`1`	`package backtype.storm.transactional;`
`2`	`2`
`3`		`-import backtype.storm.coordination.FailedBatchException;`
`4`	`3`	`import backtype.storm.Config;`
`5`	`4`	`import backtype.storm.spout.SpoutOutputCollector;`
`6`	`5`	`import backtype.storm.task.TopologyContext;`
	`6`	`+import backtype.storm.topology.FailedException;`
`7`	`7`	`import backtype.storm.topology.IRichSpout;`
`8`	`8`	`import backtype.storm.topology.OutputFieldsDeclarer;`
`9`	`9`	`import backtype.storm.transactional.state.RotatingTransactionalState;`
`@@ -137,7 +137,7 @@ private void sync() {`
`137`	`137`	`curr = nextTransactionId(curr);`
`138`	`138`	`}`
`139`	`139`	`}`
`140`		`- } catch(FailedBatchException e) {`
	`140`	`+ } catch(FailedException e) {`
`141`	`141`	`LOG.warn("Failed to get metadata for a transaction", e);`
`142`	`142`	`}`
`143`	`143`	`}`