storm-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ptgo...@apache.org
Subject [16/24] git commit: add shell bolt pending limit in fresh branch
Date Tue, 19 Nov 2013 15:26:53 GMT
add shell bolt pending limit in fresh branch


Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/45aebfe7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/45aebfe7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/45aebfe7

Branch: refs/heads/master
Commit: 45aebfe78b1643bb950d48b08c33e0fafef09b17
Parents: 6efadbe
Author: Kang Xiao <kxiao.tiger@gmail.com>
Authored: Tue Nov 5 14:24:37 2013 +0800
Committer: Kang Xiao <kxiao.tiger@gmail.com>
Committed: Tue Nov 5 14:24:37 2013 +0800

----------------------------------------------------------------------
 storm-core/src/jvm/backtype/storm/Config.java         | 6 ++++++
 storm-core/src/jvm/backtype/storm/task/ShellBolt.java | 5 +++++
 2 files changed, 11 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/45aebfe7/storm-core/src/jvm/backtype/storm/Config.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/Config.java b/storm-core/src/jvm/backtype/storm/Config.java
index 89e17d2..98c2da6 100644
--- a/storm-core/src/jvm/backtype/storm/Config.java
+++ b/storm-core/src/jvm/backtype/storm/Config.java
@@ -727,6 +727,12 @@ public class Config extends HashMap<String, Object> {
     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.
      */
     public static final String TRANSACTIONAL_ZOOKEEPER_ROOT="transactional.zookeeper.root";

http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/45aebfe7/storm-core/src/jvm/backtype/storm/task/ShellBolt.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/task/ShellBolt.java b/storm-core/src/jvm/backtype/storm/task/ShellBolt.java
index 854aa8f..6e72da6 100644
--- 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 class ShellBolt implements IBolt {
 
     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;


Mime
View raw message