You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by pt...@apache.org on 2013/11/19 16:26:53 UTC
[16/24] git commit: add shell bolt pending limit in fresh branch
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 <kx...@gmail.com>
Authored: Tue Nov 5 14:24:37 2013 +0800
Committer: Kang Xiao <kx...@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;