You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2015/07/13 16:57:18 UTC
[1/3] storm git commit: Minor type cleanup things in
storm.task.TopologyContext.java
Repository: storm
Updated Branches:
refs/heads/0.10.x-branch f342328f7 -> 701d103f0
Minor type cleanup things in storm.task.TopologyContext.java
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/aa280a68
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/aa280a68
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/aa280a68
Branch: refs/heads/0.10.x-branch
Commit: aa280a680fd0eacb34fb9b6aad8b04fec4b56cb2
Parents: f342328
Author: Dan Blanchard <da...@parsely.com>
Authored: Mon Jul 6 16:45:29 2015 -0400
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Mon Jul 13 23:35:30 2015 +0900
----------------------------------------------------------------------
.../backtype/storm/task/TopologyContext.java | 38 ++++++++++----------
1 file changed, 20 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/aa280a68/storm-core/src/jvm/backtype/storm/task/TopologyContext.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/task/TopologyContext.java b/storm-core/src/jvm/backtype/storm/task/TopologyContext.java
index 8124651..b3c6d22 100644
--- a/storm-core/src/jvm/backtype/storm/task/TopologyContext.java
+++ b/storm-core/src/jvm/backtype/storm/task/TopologyContext.java
@@ -29,6 +29,7 @@ import backtype.storm.metric.api.CombinedMetric;
import backtype.storm.state.ISubscribedState;
import backtype.storm.tuple.Fields;
import backtype.storm.utils.Utils;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -36,6 +37,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+
import org.apache.commons.lang.NotImplementedException;
import org.json.simple.JSONValue;
@@ -61,7 +63,7 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo
Map<String, Map<String, Fields>> componentToStreamToFields,
String stormId, String codeDir, String pidDir, Integer taskId,
Integer workerPort, List<Integer> workerTasks, Map<String, Object> defaultResources,
- Map<String, Object> userResources, Map<String, Object> executorData, Map registeredMetrics,
+ Map<String, Object> userResources, Map<String, Object> executorData, Map<Integer, Map<Integer, Map<String, IMetric>>> registeredMetrics,
clojure.lang.Atom openOrPrepareWasCalled) {
super(topology, stormConf, taskToComponent, componentToSortedTasks,
componentToStreamToFields, stormId, codeDir, pidDir,
@@ -154,11 +156,11 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo
}
/**
- * Gets the declared output fields for the specified stream id for the
+ * Gets the declared output fields for all streams for the
* component this task is a part of.
*/
public Map<String, List<String>> getThisOutputFieldsForStreams() {
- Map<String, List<String>> streamToFields = new HashMap<String, List<String>>();
+ Map<String, List<String>> streamToFields = new HashMap<>();
for (String stream : this.getThisStreams()) {
streamToFields.put(stream, this.getThisOutputFields(stream).toList());
}
@@ -178,7 +180,7 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo
* accesses which resource in a distributed resource to ensure an even distribution.
*/
public int getThisTaskIndex() {
- List<Integer> tasks = new ArrayList<Integer>(getComponentTasks(getThisComponentId()));
+ List<Integer> tasks = new ArrayList<>(getComponentTasks(getThisComponentId()));
Collections.sort(tasks);
for(int i=0; i<tasks.size(); i++) {
if(tasks.get(i) == getThisTaskId()) {
@@ -232,7 +234,7 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo
}
private static Map<String, Object> groupingToJSONableMap(Grouping grouping) {
- Map groupingMap = new HashMap<String, Object>();
+ Map<String, Object> groupingMap = new HashMap<>();
groupingMap.put("type", grouping.getSetField().toString());
if (grouping.is_set_fields()) {
groupingMap.put("fields", grouping.get_fields());
@@ -242,18 +244,18 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo
@Override
public String toJSONString() {
- Map obj = new HashMap();
+ Map<String, Object> obj = new HashMap<>();
obj.put("task->component", this.getTaskToComponent());
obj.put("taskid", this.getThisTaskId());
obj.put("componentid", this.getThisComponentId());
- List<String> streamList = new ArrayList<String>();
+ List<String> streamList = new ArrayList<>();
streamList.addAll(this.getThisStreams());
obj.put("streams", streamList);
obj.put("stream->outputfields", this.getThisOutputFieldsForStreams());
// Convert targets to a JSON serializable format
- Map<String, Map> stringTargets = new HashMap<String, Map>();
+ Map<String, Map<String, Object>> stringTargets = new HashMap<>();
for (Map.Entry<String, Map<String, Grouping>> entry : this.getThisTargets().entrySet()) {
- Map stringTargetMap = new HashMap<String, Object>();
+ Map<String, Object> stringTargetMap = new HashMap<>();
for (Map.Entry<String, Grouping> innerEntry : entry.getValue().entrySet()) {
stringTargetMap.put(innerEntry.getKey(), groupingToJSONableMap(innerEntry.getValue()));
}
@@ -261,12 +263,12 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo
}
obj.put("stream->target->grouping", stringTargets);
// Convert sources to a JSON serializable format
- Map<String, Map<String, Object>> stringSources = new HashMap<String, Map<String, Object>>();
+ Map<String, Map<String, Object>> stringSources = new HashMap<>();
for (Map.Entry<GlobalStreamId, Grouping> entry : this.getThisSources().entrySet()) {
GlobalStreamId gid = entry.getKey();
Map<String, Object> stringSourceMap = stringSources.get(gid.get_componentId());
if (stringSourceMap == null) {
- stringSourceMap = new HashMap<String, Object>();
+ stringSourceMap = new HashMap<>();
stringSources.put(gid.get_componentId(), stringSourceMap);
}
stringSourceMap.put(gid.get_streamId(), groupingToJSONableMap(entry.getValue()));
@@ -301,17 +303,17 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo
throw new RuntimeException("The same metric name `" + name + "` was registered twice." );
}
- Map m1 = _registeredMetrics;
+ Map<Integer, Map<Integer, Map<String, IMetric>>> m1 = _registeredMetrics;
if(!m1.containsKey(timeBucketSizeInSecs)) {
- m1.put(timeBucketSizeInSecs, new HashMap());
+ m1.put(timeBucketSizeInSecs, new HashMap<Integer, Map<String, IMetric>>());
}
- Map m2 = (Map)m1.get(timeBucketSizeInSecs);
+ Map<Integer, Map<String, IMetric>> m2 = m1.get(timeBucketSizeInSecs);
if(!m2.containsKey(_taskId)) {
- m2.put(_taskId, new HashMap());
+ m2.put(_taskId, new HashMap<String, IMetric>());
}
- Map m3 = (Map)m2.get(_taskId);
+ Map<String, IMetric> m3 = m2.get(_taskId);
if(m3.containsKey(name)) {
throw new RuntimeException("The same metric name `" + name + "` was registered twice." );
} else {
@@ -346,13 +348,13 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo
}
/*
- * Convinience method for registering ReducedMetric.
+ * Convenience method for registering ReducedMetric.
*/
public ReducedMetric registerMetric(String name, IReducer reducer, int timeBucketSizeInSecs) {
return registerMetric(name, new ReducedMetric(reducer), timeBucketSizeInSecs);
}
/*
- * Convinience method for registering CombinedMetric.
+ * Convenience method for registering CombinedMetric.
*/
public CombinedMetric registerMetric(String name, ICombiner combiner, int timeBucketSizeInSecs) {
return registerMetric(name, new CombinedMetric(combiner), timeBucketSizeInSecs);
[3/3] storm git commit: add STORM-928 to CHANGELOG.md
Posted by ka...@apache.org.
add STORM-928 to CHANGELOG.md
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/701d103f
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/701d103f
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/701d103f
Branch: refs/heads/0.10.x-branch
Commit: 701d103f0673ac42a6522415910e1d0b89957792
Parents: 6141d74
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Mon Jul 13 23:33:25 2015 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Mon Jul 13 23:36:01 2015 +0900
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/701d103f/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2f390ca..47362e2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,7 @@
* STORM-843: [storm-redis] Add Javadoc to storm-redis
* STORM-866: Use storm.log.dir instead of storm.home in log4j2 config
* STORM-810: PartitionManager in storm-kafka should commit latest offset before close
+ * STORM-928: Add sources->streams->fields map to Multi-Lang Handshake
## 0.10.0-beta1
* STORM-873: Flux does not handle diamond topologies
[2/3] storm git commit: Add source->stream->fields mapping to
multi-lang handshake.
Posted by ka...@apache.org.
Add source->stream->fields mapping to multi-lang handshake.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/6141d741
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/6141d741
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/6141d741
Branch: refs/heads/0.10.x-branch
Commit: 6141d741bcef5a6fe7069fabc31008cc5084b655
Parents: aa280a6
Author: Dan Blanchard <da...@parsely.com>
Authored: Tue Jul 7 10:25:58 2015 -0400
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Mon Jul 13 23:35:37 2015 +0900
----------------------------------------------------------------------
docs/documentation/Multilang-protocol.md | 9 ++++++--
.../backtype/storm/task/TopologyContext.java | 23 ++++++++++++++++++++
2 files changed, 30 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/6141d741/docs/documentation/Multilang-protocol.md
----------------------------------------------------------------------
diff --git a/docs/documentation/Multilang-protocol.md b/docs/documentation/Multilang-protocol.md
index 017ad32..2a90059 100644
--- a/docs/documentation/Multilang-protocol.md
+++ b/docs/documentation/Multilang-protocol.md
@@ -66,7 +66,7 @@ The initial handshake is the same for both types of shell components:
"4": "example-bolt2"
},
"taskid": 3,
- // Everything below this line is only available in Storm 0.11.0+
+ // Everything below this line is only available in Storm 0.10.0+
"componentid": "example-bolt"
"stream->target->grouping": {
"default": {
@@ -82,6 +82,11 @@ The initial handshake is the same for both types of shell components:
}
}
}
+ "source->stream->fields": {
+ "example-spout": {
+ "default": ["word"]
+ }
+ }
}
}
```
@@ -90,7 +95,7 @@ Your script should create an empty file named with its PID in this directory. e.
the PID is 1234, so an empty file named 1234 is created in the directory. This
file lets the supervisor know the PID so it can shutdown the process later on.
-As of Storm 0.11.0, the context sent by Storm to shell components has been
+As of Storm 0.10.0, the context sent by Storm to shell components has been
enhanced substantially to include all aspects of the topology context available
to JVM components. One key addition is the ability to determine a shell
component's source and targets (i.e., inputs and outputs) in the topology via
http://git-wip-us.apache.org/repos/asf/storm/blob/6141d741/storm-core/src/jvm/backtype/storm/task/TopologyContext.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/task/TopologyContext.java b/storm-core/src/jvm/backtype/storm/task/TopologyContext.java
index b3c6d22..cefa207 100644
--- a/storm-core/src/jvm/backtype/storm/task/TopologyContext.java
+++ b/storm-core/src/jvm/backtype/storm/task/TopologyContext.java
@@ -191,6 +191,28 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo
}
/**
+ * Gets the declared input fields for this component.
+ *
+ * @return A map from sources to streams to fields.
+ */
+ public Map<String, Map<String, List<String>>> getThisInputFields() {
+ Map<String, Map<String, List<String>>> outputMap = new HashMap<>();
+ for (Map.Entry<GlobalStreamId, Grouping> entry : this.getThisSources().entrySet()) {
+ String componentId = entry.getKey().get_componentId();
+ Set<String> streams = getComponentStreams(componentId);
+ for (String stream : streams) {
+ Map<String, List<String>> streamFieldMap = outputMap.get(componentId);
+ if (streamFieldMap == null) {
+ streamFieldMap = new HashMap<>();
+ outputMap.put(componentId, streamFieldMap);
+ }
+ streamFieldMap.put(stream, getComponentOutputFields(componentId, stream).toList());
+ }
+ }
+ return outputMap;
+ }
+
+ /**
* Gets the declared inputs to this component.
*
* @return A map from subscribed component/stream to the grouping subscribed with.
@@ -274,6 +296,7 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo
stringSourceMap.put(gid.get_streamId(), groupingToJSONableMap(entry.getValue()));
}
obj.put("source->stream->grouping", stringSources);
+ obj.put("source->stream->fields", this.getThisInputFields());
return JSONValue.toJSONString(obj);
}