You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2014/07/01 22:40:38 UTC
[07/24] git commit: merge 0.9.2 incubating
merge 0.9.2 incubating
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/bc391722
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/bc391722
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/bc391722
Branch: refs/heads/master
Commit: bc391722098c1e66f6ebe0377a3de6b51a7cf38c
Parents: e2a7f7c c89fb82
Author: JuDasheng <ju...@meituan.com>
Authored: Tue Jun 10 17:43:29 2014 +0800
Committer: JuDasheng <ju...@meituan.com>
Committed: Tue Jun 10 17:43:29 2014 +0800
----------------------------------------------------------------------
.gitignore | 4 +-
CHANGELOG.md | 64 +
DEVELOPER.md | 313 +++++
NOTICE | 6 +-
README.markdown | 12 +-
SECURITY.md | 74 ++
bin/storm | 59 +-
bin/storm-config.cmd | 27 +-
bin/storm.cmd | 59 +-
conf/defaults.yaml | 11 +-
conf/storm.yaml.example | 2 +-
conf/storm_env.ini | 30 +
examples/storm-starter/README.markdown | 127 ++
.../multilang/resources/splitsentence.py | 24 +
.../multilang/resources/splitsentence.rb | 26 +
.../storm-starter/multilang/resources/storm.py | 221 ++++
.../storm-starter/multilang/resources/storm.rb | 200 +++
examples/storm-starter/pom.xml | 166 +++
.../src/clj/storm/starter/clj/word_count.clj | 95 ++
.../jvm/storm/starter/BasicDRPCTopology.java | 78 ++
.../jvm/storm/starter/ExclamationTopology.java | 87 ++
.../src/jvm/storm/starter/ManualDRPC.java | 68 +
.../jvm/storm/starter/PrintSampleStream.java | 58 +
.../src/jvm/storm/starter/ReachTopology.java | 196 +++
.../src/jvm/storm/starter/RollingTopWords.java | 78 ++
.../jvm/storm/starter/SingleJoinExample.java | 64 +
.../storm/starter/TransactionalGlobalCount.java | 173 +++
.../jvm/storm/starter/TransactionalWords.java | 246 ++++
.../jvm/storm/starter/WordCountTopology.java | 107 ++
.../storm/starter/bolt/AbstractRankerBolt.java | 110 ++
.../starter/bolt/IntermediateRankingsBolt.java | 58 +
.../src/jvm/storm/starter/bolt/PrinterBolt.java | 37 +
.../storm/starter/bolt/RollingCountBolt.java | 142 ++
.../jvm/storm/starter/bolt/SingleJoinBolt.java | 114 ++
.../storm/starter/bolt/TotalRankingsBolt.java | 59 +
.../starter/spout/RandomSentenceSpout.java | 64 +
.../storm/starter/spout/TwitterSampleSpout.java | 164 +++
.../tools/NthLastModifiedTimeTracker.java | 70 +
.../src/jvm/storm/starter/tools/Rankable.java | 32 +
.../starter/tools/RankableObjectWithFields.java | 148 +++
.../src/jvm/storm/starter/tools/Rankings.java | 156 +++
.../starter/tools/SlidingWindowCounter.java | 119 ++
.../storm/starter/tools/SlotBasedCounter.java | 118 ++
.../jvm/storm/starter/trident/TridentReach.java | 156 +++
.../storm/starter/trident/TridentWordCount.java | 85 ++
.../src/jvm/storm/starter/util/StormRunner.java | 39 +
.../jvm/storm/starter/util/TupleHelpers.java | 33 +
.../bolt/IntermediateRankingsBoltTest.java | 146 +++
.../starter/bolt/RollingCountBoltTest.java | 113 ++
.../starter/bolt/TotalRankingsBoltTest.java | 147 +++
.../storm/starter/tools/MockTupleHelpers.java | 40 +
.../tools/NthLastModifiedTimeTrackerTest.java | 125 ++
.../tools/RankableObjectWithFieldsTest.java | 252 ++++
.../jvm/storm/starter/tools/RankingsTest.java | 368 ++++++
.../starter/tools/SlidingWindowCounterTest.java | 106 ++
.../starter/tools/SlotBasedCounterTest.java | 181 +++
external/storm-kafka/CHANGELOG.md | 13 +
external/storm-kafka/README.md | 28 +
external/storm-kafka/pom.xml | 138 ++
.../storm-kafka/src/jvm/storm/kafka/Broker.java | 81 ++
.../src/jvm/storm/kafka/BrokerHosts.java | 25 +
.../jvm/storm/kafka/DynamicBrokersReader.java | 147 +++
.../kafka/DynamicPartitionConnections.java | 94 ++
.../jvm/storm/kafka/FailedFetchException.java | 29 +
.../src/jvm/storm/kafka/KafkaConfig.java | 52 +
.../src/jvm/storm/kafka/KafkaError.java | 43 +
.../src/jvm/storm/kafka/KafkaSpout.java | 195 +++
.../src/jvm/storm/kafka/KafkaUtils.java | 243 ++++
.../src/jvm/storm/kafka/KeyValueScheme.java | 28 +
.../kafka/KeyValueSchemeAsMultiScheme.java | 37 +
.../src/jvm/storm/kafka/Partition.java | 64 +
.../jvm/storm/kafka/PartitionCoordinator.java | 28 +
.../src/jvm/storm/kafka/PartitionManager.java | 265 ++++
.../src/jvm/storm/kafka/SpoutConfig.java | 36 +
.../src/jvm/storm/kafka/StaticCoordinator.java | 51 +
.../src/jvm/storm/kafka/StaticHosts.java | 38 +
.../storm/kafka/StaticPartitionConnections.java | 52 +
.../jvm/storm/kafka/StringKeyValueScheme.java | 37 +
.../src/jvm/storm/kafka/StringScheme.java | 46 +
.../src/jvm/storm/kafka/ZkCoordinator.java | 113 ++
.../src/jvm/storm/kafka/ZkHosts.java | 36 +
.../src/jvm/storm/kafka/ZkState.java | 116 ++
.../src/jvm/storm/kafka/bolt/KafkaBolt.java | 89 ++
.../jvm/storm/kafka/trident/Coordinator.java | 50 +
.../storm/kafka/trident/DefaultCoordinator.java | 31 +
.../trident/GlobalPartitionInformation.java | 98 ++
.../storm/kafka/trident/IBatchCoordinator.java | 26 +
.../jvm/storm/kafka/trident/IBrokerReader.java | 25 +
.../src/jvm/storm/kafka/trident/MaxMetric.java | 40 +
.../kafka/trident/OpaqueTridentKafkaSpout.java | 59 +
.../storm/kafka/trident/StaticBrokerReader.java | 36 +
.../trident/TransactionalTridentKafkaSpout.java | 58 +
.../storm/kafka/trident/TridentKafkaConfig.java | 37 +
.../kafka/trident/TridentKafkaEmitter.java | 268 ++++
.../jvm/storm/kafka/trident/ZkBrokerReader.java | 70 +
.../storm/kafka/DynamicBrokersReaderTest.java | 155 +++
.../src/test/storm/kafka/KafkaErrorTest.java | 39 +
.../src/test/storm/kafka/KafkaTestBroker.java | 75 ++
.../src/test/storm/kafka/KafkaUtilsTest.java | 225 ++++
.../storm/kafka/StringKeyValueSchemeTest.java | 38 +
.../src/test/storm/kafka/TestUtils.java | 20 +
.../src/test/storm/kafka/ZkCoordinatorTest.java | 130 ++
.../test/storm/kafka/bolt/KafkaBoltTest.java | 171 +++
logback/cluster.xml | 2 +-
pom.xml | 69 +-
storm-core/pom.xml | 49 +-
.../clj/backtype/storm/command/rebalance.clj | 7 +-
storm-core/src/clj/backtype/storm/config.clj | 2 +-
.../src/clj/backtype/storm/daemon/common.clj | 6 +-
.../src/clj/backtype/storm/daemon/drpc.clj | 17 +-
.../src/clj/backtype/storm/daemon/executor.clj | 43 +-
.../src/clj/backtype/storm/daemon/logviewer.clj | 41 +-
.../src/clj/backtype/storm/daemon/nimbus.clj | 9 -
.../clj/backtype/storm/daemon/supervisor.clj | 66 +-
.../src/clj/backtype/storm/daemon/worker.clj | 76 +-
storm-core/src/clj/backtype/storm/disruptor.clj | 10 +-
.../src/clj/backtype/storm/messaging/loader.clj | 81 +-
.../src/clj/backtype/storm/messaging/local.clj | 20 +-
storm-core/src/clj/backtype/storm/testing.clj | 5 +-
storm-core/src/clj/backtype/storm/timer.clj | 7 +-
storm-core/src/clj/backtype/storm/ui/core.clj | 1209 +++++++-----------
.../src/clj/backtype/storm/ui/helpers.clj | 5 +-
storm-core/src/clj/backtype/storm/util.clj | 17 +-
storm-core/src/clj/backtype/storm/zookeeper.clj | 10 +-
storm-core/src/jvm/backtype/storm/Config.java | 38 +-
.../jvm/backtype/storm/ConfigValidation.java | 17 +
.../src/jvm/backtype/storm/StormSubmitter.java | 176 ++-
.../backtype/storm/messaging/IConnection.java | 15 +-
.../backtype/storm/messaging/netty/Client.java | 333 +++--
.../backtype/storm/messaging/netty/Context.java | 57 +-
.../storm/messaging/netty/ControlMessage.java | 6 +-
.../storm/messaging/netty/MessageBatch.java | 52 +-
.../storm/messaging/netty/MessageDecoder.java | 108 +-
.../netty/NettyRenameThreadFactory.java | 35 +
.../backtype/storm/messaging/netty/Server.java | 147 ++-
.../netty/StormClientErrorHandler.java | 41 +
.../messaging/netty/StormClientHandler.java | 121 --
.../netty/StormClientPipelineFactory.java | 2 +-
.../messaging/netty/StormServerHandler.java | 40 +-
.../jvm/backtype/storm/multilang/BoltMsg.java | 63 +
.../backtype/storm/multilang/ISerializer.java | 65 +
.../storm/multilang/JsonSerializer.java | 162 +++
.../storm/multilang/NoOutputException.java | 23 +
.../jvm/backtype/storm/multilang/ShellMsg.java | 105 ++
.../jvm/backtype/storm/multilang/SpoutMsg.java | 34 +
.../jvm/backtype/storm/spout/ShellSpout.java | 86 +-
.../src/jvm/backtype/storm/task/ShellBolt.java | 115 +-
.../backtype/storm/task/TopologyContext.java | 9 +
.../storm/testing/TestEventLogSpout.java | 139 ++
.../storm/testing/TestEventOrderCheckBolt.java | 76 ++
.../transactional/state/TransactionalState.java | 2 +-
.../backtype/storm/utils/DisruptorQueue.java | 9 +-
.../jvm/backtype/storm/utils/ShellProcess.java | 169 ++-
.../src/jvm/backtype/storm/utils/Time.java | 18 +-
.../backtype/storm/utils/TransferDrainer.java | 113 ++
.../src/jvm/backtype/storm/utils/Utils.java | 117 +-
storm-core/src/jvm/storm/trident/Stream.java | 5 +-
.../trident/spout/RichSpoutBatchTriggerer.java | 1 +
.../jvm/storm/trident/state/map/OpaqueMap.java | 6 +-
.../trident/state/map/RemovableMapState.java | 8 +
.../storm/trident/testing/MemoryMapState.java | 27 +-
.../topology/state/TransactionalState.java | 2 +-
.../jvm/storm/trident/util/TridentUtils.java | 14 +-
storm-core/src/ui/public/component.html | 88 ++
storm-core/src/ui/public/index.html | 73 ++
storm-core/src/ui/public/js/arbor-graphics.js | 51 +
storm-core/src/ui/public/js/arbor-tween.js | 86 ++
storm-core/src/ui/public/js/arbor.js | 67 +
storm-core/src/ui/public/js/jquery.mustache.js | 592 +++++++++
storm-core/src/ui/public/js/purl.js | 267 ++++
storm-core/src/ui/public/js/script.js | 51 +-
storm-core/src/ui/public/js/visualization.js | 403 ++++++
.../templates/component-page-template.html | 152 +++
.../public/templates/index-page-template.html | 62 +
.../public/templates/json-error-template.html | 4 +
.../templates/topology-page-template.html | 128 ++
storm-core/src/ui/public/topology.html | 90 ++
.../test/clj/backtype/storm/config_test.clj | 20 +
.../test/clj/backtype/storm/drpc_test.clj | 27 +-
.../clj/backtype/storm/integration_test.clj | 9 +-
.../storm/messaging/netty_unit_test.clj | 46 +-
.../test/clj/backtype/storm/messaging_test.clj | 35 +-
.../test/clj/backtype/storm/metrics_test.clj | 24 +
.../test/clj/backtype/storm/multilang_test.clj | 6 +-
.../test/clj/backtype/storm/nimbus_test.clj | 16 +-
.../test/clj/backtype/storm/supervisor_test.clj | 66 +-
.../test/clj/backtype/storm/utils_test.clj | 13 +-
.../test/clj/storm/trident/integration_test.clj | 2 +-
.../test/clj/storm/trident/state_test.clj | 33 +-
storm-dist/binary/NOTICE | 9 +-
storm-dist/binary/pom.xml | 40 +-
storm-dist/binary/src/main/assembly/binary.xml | 41 +
storm-dist/source/pom.xml | 38 +-
193 files changed, 14785 insertions(+), 1693 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/bc391722/storm-core/src/jvm/backtype/storm/spout/ShellSpout.java
----------------------------------------------------------------------
diff --cc storm-core/src/jvm/backtype/storm/spout/ShellSpout.java
index f54cf72,d6a18e7..368f16a
--- a/storm-core/src/jvm/backtype/storm/spout/ShellSpout.java
+++ b/storm-core/src/jvm/backtype/storm/spout/ShellSpout.java
@@@ -18,18 -18,15 +18,18 @@@
package backtype.storm.spout;
import backtype.storm.generated.ShellComponent;
+import backtype.storm.metric.api.IMetric;
+import backtype.storm.metric.api.rpc.IShellMetric;
+ import backtype.storm.multilang.ShellMsg;
+ import backtype.storm.multilang.SpoutMsg;
import backtype.storm.task.TopologyContext;
import backtype.storm.utils.ShellProcess;
import java.util.Map;
import java.util.List;
import java.io.IOException;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
- import org.json.simple.JSONObject;
public class ShellSpout implements ISpout {
@@@ -38,9 -35,8 +38,11 @@@
private SpoutOutputCollector _collector;
private String[] _command;
private ShellProcess _process;
+
+ private TopologyContext _context;
+
+ private SpoutMsg _spoutMsg;
+
public ShellSpout(ShellComponent component) {
this(component.get_execution_command(), component.get_script());
}
@@@ -48,118 -44,75 +50,110 @@@
public ShellSpout(String... command) {
_command = command;
}
-
+
public void open(Map stormConf, TopologyContext context,
SpoutOutputCollector collector) {
- _process = new ShellProcess(_command);
_collector = collector;
+ _context = context;
- try {
- Number subpid = _process.launch(stormConf, context);
- LOG.info("Launched subprocess with pid " + subpid);
- } catch (IOException e) {
- throw new RuntimeException("Error when launching multilang subprocess\n" + _process.getErrorsString(), e);
- }
+ _process = new ShellProcess(_command);
+
+ Number subpid = _process.launch(stormConf, context);
+ LOG.info("Launched subprocess with pid " + subpid);
}
public void close() {
_process.destroy();
}
- private JSONObject _next;
public void nextTuple() {
- if (_next == null) {
- _next = new JSONObject();
- _next.put("command", "next");
+ if (_spoutMsg == null) {
+ _spoutMsg = new SpoutMsg();
}
-
- querySubprocess(_next);
+ _spoutMsg.setCommand("next");
+ _spoutMsg.setId("");
+ querySubprocess();
}
- private JSONObject _ack;
public void ack(Object msgId) {
- if (_ack == null) {
- _ack = new JSONObject();
- _ack.put("command", "ack");
+ if (_spoutMsg == null) {
+ _spoutMsg = new SpoutMsg();
}
-
- _ack.put("id", msgId);
- querySubprocess(_ack);
+ _spoutMsg.setCommand("ack");
+ _spoutMsg.setId(msgId);
+ querySubprocess();
}
- private JSONObject _fail;
public void fail(Object msgId) {
- if (_fail == null) {
- _fail = new JSONObject();
- _fail.put("command", "fail");
+ if (_spoutMsg == null) {
+ _spoutMsg = new SpoutMsg();
}
-
- _fail.put("id", msgId);
- querySubprocess(_fail);
+ _spoutMsg.setCommand("fail");
+ _spoutMsg.setId(msgId);
+ querySubprocess();
}
+
+ private void handleMetrics(Map action) {
+ //get metric name
+ Object nameObj = action.get("name");
+ if (nameObj == null || !(nameObj instanceof String) ) {
+ throw new RuntimeException("Receive Metrics name is null or is not String");
+ }
+ String name = (String) nameObj;
+ if (name.isEmpty()) {
+ throw new RuntimeException("Receive Metrics name is empty");
+ }
+
+ //get metric by name
+ IMetric iMetric = _context.getRegisteredMetricByName(name);
+ if (iMetric == null) {
+ throw new RuntimeException("Not find metric by name["+name+"] ");
+ }
+ if ( !(iMetric instanceof IShellMetric)) {
+ throw new RuntimeException("Metric["+name+"] is not IShellMetric, can not call by RPC");
+ }
+ IShellMetric iShellMetric = (IShellMetric)iMetric;
+
+ //call updateMetricFromRPC with params
+ Object paramsObj = action.get("params");
+ try {
+ iShellMetric.updateMetricFromRPC(paramsObj);
+ } catch (RuntimeException re) {
+ throw re;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
- private void querySubprocess(Object query) {
+ private void querySubprocess() {
try {
- _process.writeMessage(query);
+ _process.writeSpoutMsg(_spoutMsg);
while (true) {
- JSONObject action = _process.readMessage();
- String command = (String) action.get("command");
+ ShellMsg shellMsg = _process.readShellMsg();
+ String command = shellMsg.getCommand();
if (command.equals("sync")) {
return;
} else if (command.equals("log")) {
- String msg = (String) action.get("msg");
+ String msg = shellMsg.getMsg();
LOG.info("Shell msg: " + msg);
} else if (command.equals("emit")) {
- String stream = (String) action.get("stream");
- if (stream == null) stream = Utils.DEFAULT_STREAM_ID;
- Long task = (Long) action.get("task");
- List<Object> tuple = (List) action.get("tuple");
- Object messageId = (Object) action.get("id");
- if (task == null) {
+ String stream = shellMsg.getStream();
+ Long task = shellMsg.getTask();
+ List<Object> tuple = shellMsg.getTuple();
+ Object messageId = shellMsg.getId();
+ if (task == 0) {
List<Integer> outtasks = _collector.emit(stream, tuple, messageId);
- Object need_task_ids = action.get("need_task_ids");
- if (need_task_ids == null || ((Boolean) need_task_ids).booleanValue()) {
- _process.writeMessage(outtasks);
+ if (shellMsg.areTaskIdsNeeded()) {
+ _process.writeTaskIds(outtasks);
}
} else {
- _collector.emitDirect((int)task.longValue(), stream, tuple, messageId);
+ _collector.emitDirect((int) task.longValue(), stream, tuple, messageId);
}
+ } else if (command.equals("metrics")) {
+ handleMetrics(action);
+ } else {
+ throw new RuntimeException("Unknown command received: " + command);
}
}
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/bc391722/storm-core/src/jvm/backtype/storm/task/ShellBolt.java
----------------------------------------------------------------------
diff --cc storm-core/src/jvm/backtype/storm/task/ShellBolt.java
index de3c3f2,81aca02..50d6e20
--- a/storm-core/src/jvm/backtype/storm/task/ShellBolt.java
+++ b/storm-core/src/jvm/backtype/storm/task/ShellBolt.java
@@@ -19,15 -19,13 +19,16 @@@ package backtype.storm.task
import backtype.storm.Config;
import backtype.storm.generated.ShellComponent;
+import backtype.storm.metric.api.IMetric;
+import backtype.storm.metric.api.rpc.IShellMetric;
+import backtype.storm.tuple.MessageId;
import backtype.storm.tuple.Tuple;
- import backtype.storm.utils.Utils;
import backtype.storm.utils.ShellProcess;
- import java.io.IOException;
+ import backtype.storm.multilang.BoltMsg;
+ import backtype.storm.multilang.ShellMsg;
+
import java.util.ArrayList;
- import java.util.Arrays;
+ import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import static java.util.concurrent.TimeUnit.SECONDS;
@@@ -77,11 -73,9 +76,11 @@@ public class ShellBolt implements IBol
private volatile Throwable _exception;
private LinkedBlockingQueue _pendingWrites = new LinkedBlockingQueue();
private Random _rand;
-
+
private Thread _readerThread;
private Thread _writerThread;
+
+ private TopologyContext _context;
public ShellBolt(ShellComponent component) {
this(component.get_execution_command(), component.get_script());
@@@ -98,42 -92,32 +97,42 @@@
this._pendingWrites = new LinkedBlockingQueue(((Number)maxPending).intValue());
}
_rand = new Random();
- _process = new ShellProcess(_command);
_collector = collector;
++<<<<<<< HEAD
+ _context = context;
++=======
+ _process = new ShellProcess(_command);
++>>>>>>> upstream/master
- try {
- //subprocesses must send their pid first thing
- Number subpid = _process.launch(stormConf, context);
- LOG.info("Launched subprocess with pid " + subpid);
- } catch (IOException e) {
- throw new RuntimeException("Error when launching multilang subprocess\n" + _process.getErrorsString(), e);
- }
+ //subprocesses must send their pid first thing
+ Number subpid = _process.launch(stormConf, context);
+ LOG.info("Launched subprocess with pid " + subpid);
// reader
_readerThread = new Thread(new Runnable() {
public void run() {
while (_running) {
try {
- JSONObject action = _process.readMessage();
- if (action == null) {
- // ignore sync
- }
+ ShellMsg shellMsg = _process.readShellMsg();
- String command = (String) action.get("command");
+ String command = shellMsg.getCommand();
if(command.equals("ack")) {
- handleAck(action);
+ handleAck(shellMsg.getId());
} else if (command.equals("fail")) {
- handleFail(action);
+ handleFail(shellMsg.getId());
} else if (command.equals("error")) {
- handleError(action);
+ handleError(shellMsg.getMsg());
} else if (command.equals("log")) {
- String msg = (String) action.get("msg");
+ String msg = shellMsg.getMsg();
LOG.info("Shell msg: " + msg);
} else if (command.equals("emit")) {
++<<<<<<< HEAD
+ handleEmit(action);
+ } else if (command.equals("metrics")) {
+ handleMetrics(action);
++=======
+ handleEmit(shellMsg);
++>>>>>>> upstream/master
}
} catch (InterruptedException e) {
} catch (Throwable t) {
@@@ -192,9 -179,8 +194,14 @@@
_process.destroy();
_inputs.clear();
}
++<<<<<<< HEAD
+
+ private void handleAck(Map action) {
+ String id = (String) action.get("id");
++=======
+
+ private void handleAck(Object id) {
++>>>>>>> upstream/master
Tuple acked = _inputs.remove(id);
if(acked==null) {
throw new RuntimeException("Acked a non-existent or already acked/failed id: " + id);
@@@ -242,41 -219,10 +240,42 @@@
_pendingWrites.put(outtasks);
}
} else {
- _collector.emitDirect((int)task.longValue(), stream, anchors, tuple);
+ _collector.emitDirect((int) shellMsg.getTask(),
+ shellMsg.getStream(), anchors, shellMsg.getTuple());
}
}
+
+ private void handleMetrics(Map action) {
+ //get metric name
+ Object nameObj = action.get("name");
+ if (nameObj == null || !(nameObj instanceof String) ) {
+ throw new RuntimeException("Receive Metrics name is null or is not String");
+ }
+ String name = (String) nameObj;
+ if (name.isEmpty()) {
+ throw new RuntimeException("Receive Metrics name is empty");
+ }
+
+ //get metric by name
+ IMetric iMetric = _context.getRegisteredMetricByName(name);
+ if (iMetric == null) {
+ throw new RuntimeException("Not find metric by name["+name+"] ");
+ }
+ if ( !(iMetric instanceof IShellMetric)) {
+ throw new RuntimeException("Metric["+name+"] is not IShellMetric, can not call by RPC");
+ }
+ IShellMetric iShellMetric = (IShellMetric)iMetric;
+
+ //call updateMetricFromRPC with params
+ Object paramsObj = action.get("params");
+ try {
+ iShellMetric.updateMetricFromRPC(paramsObj);
+ } catch (RuntimeException re) {
+ throw re;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
private void die(Throwable exception) {
_exception = exception;
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/bc391722/storm-core/src/jvm/backtype/storm/task/TopologyContext.java
----------------------------------------------------------------------