add shell bolt pending limit in fresh branch

xiaokang · xiaokang · commit 45aebfe78b16 · 2013-11-05T14:24:37.000+08:00
diff --git a/storm-core/src/jvm/backtype/storm/Config.java b/storm-core/src/jvm/backtype/storm/Config.java
@@ -726,6 +726,12 @@ public class Config extends HashMap<String, Object> {
     public static final String TOPOLOGY_NAME="topology.name";
     public static final Object TOPOLOGY_NAME_SCHEMA = String.class;
 
+    /**
+     * Max pending tuples in one ShellBolt
+     */
+    public static final String TOPOLOGY_SHELLBOLT_MAX_PENDING="topology.shellbolt.max.pending";
+    public static final Object TOPOLOGY_SHELLBOLT_MAX_PENDING_SCHEMA = Number.class;
+
     /**
      * The root directory in ZooKeeper for metadata about TransactionalSpouts.
      */
diff --git a/storm-core/src/jvm/backtype/storm/task/ShellBolt.java b/storm-core/src/jvm/backtype/storm/task/ShellBolt.java
@@ -1,5 +1,6 @@
 package backtype.storm.task;
 
+import backtype.storm.Config;
 import backtype.storm.generated.ShellComponent;
 import backtype.storm.tuple.MessageId;
 import backtype.storm.tuple.Tuple;
@@ -71,6 +72,10 @@ public ShellBolt(String... command) {
 
     public void prepare(Map stormConf, TopologyContext context,
                         final OutputCollector collector) {
+        Object maxPending = stormConf.get(Config.TOPOLOGY_SHELLBOLT_MAX_PENDING);
+        if (maxPending != null && maxPending instanceof Number) {
+           this._pendingWrites = new LinkedBlockingQueue(((Number)maxPending).intValue());
+        }
         _rand = new Random();
         _process = new ShellProcess(_command);
         _collector = collector;
