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 2016/10/01 04:09:28 UTC
[1/3] storm git commit: STORM-1870 Allow FluxShellBolt/Spout set
custom "componentConfig" via yaml
Repository: storm
Updated Branches:
refs/heads/1.x-branch 2bda84f26 -> d235f47bf
STORM-1870 Allow FluxShellBolt/Spout set custom "componentConfig" via yaml
* add addComponentConfig methods to FluxShellSpout and FluxShellBolt
* address unit tests
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/04d8d570
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/04d8d570
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/04d8d570
Branch: refs/heads/1.x-branch
Commit: 04d8d570def149b662e844042d54601784ffd997
Parents: 2bda84f
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Wed Aug 24 12:09:57 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Sat Oct 1 12:50:39 2016 +0900
----------------------------------------------------------------------
.../java/org/apache/storm/flux/TCKTest.java | 10 ++
.../test/resources/configs/bad_shell_test.yaml | 118 +++++++++++++++++++
.../src/test/resources/configs/shell_test.yaml | 14 +++
.../flux/wrappers/bolts/FluxShellBolt.java | 58 ++++++++-
.../flux/wrappers/spouts/FluxShellSpout.java | 55 ++++++++-
5 files changed, 251 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/04d8d570/external/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java
----------------------------------------------------------------------
diff --git a/external/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java b/external/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java
index 7a5ed7a..e0ab207 100644
--- a/external/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java
+++ b/external/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java
@@ -50,6 +50,16 @@ public class TCKTest {
topology.validate();
}
+ @Test(expected = IllegalArgumentException.class)
+ public void testBadShellComponents() throws Exception {
+ TopologyDef topologyDef = FluxParser.parseResource("/configs/bad_shell_test.yaml", false, true, null, false);
+ Config conf = FluxBuilder.buildConfig(topologyDef);
+ ExecutionContext context = new ExecutionContext(topologyDef, conf);
+ StormTopology topology = FluxBuilder.buildTopology(context);
+ assertNotNull(topology);
+ topology.validate();
+ }
+
@Test
public void testKafkaSpoutConfig() throws Exception {
TopologyDef topologyDef = FluxParser.parseResource("/configs/kafka_test.yaml", false, true, null, false);
http://git-wip-us.apache.org/repos/asf/storm/blob/04d8d570/external/flux/flux-core/src/test/resources/configs/bad_shell_test.yaml
----------------------------------------------------------------------
diff --git a/external/flux/flux-core/src/test/resources/configs/bad_shell_test.yaml b/external/flux/flux-core/src/test/resources/configs/bad_shell_test.yaml
new file mode 100644
index 0000000..0892ce7
--- /dev/null
+++ b/external/flux/flux-core/src/test/resources/configs/bad_shell_test.yaml
@@ -0,0 +1,118 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Test ability to wire together shell spouts/bolts
+---
+
+# topology definition
+# name to be used when submitting
+name: "shell-topology"
+
+# Components
+# Components are analagous to Spring beans. They are meant to be used as constructor,
+# property(setter), and builder arguments.
+#components:
+# - id: "myComponent"
+# className: "com.foo.bar.MyComponent"
+# constructorArgs:
+# - ...
+# properties:
+# foo: "bar"
+# bar: "foo"
+
+# NOTE: We may want to consider some level of spring integration. For example, allowing component references
+# to a spring `ApplicationContext`.
+
+# topology configuration
+# this will be passed to the submitter as a map of config options
+#
+config:
+ topology.workers: 1
+ # ...
+
+# spout definitions
+spouts:
+ - id: "sentence-spout"
+ className: "org.apache.storm.flux.wrappers.spouts.FluxShellSpout"
+ # shell spout constructor takes 2 arguments: String[], String[]
+ constructorArgs:
+ # command line
+ - ["node", "randomsentence.js"]
+ # output fields
+ - ["word"]
+ configMethods:
+ - name: "addComponentConfig"
+ args: ["rabbitmq.configfile", "etc/rabbit.yml", "hello"]
+ - name: "addComponentConfig"
+ args:
+ - "publisher.data_paths"
+ - ["actions", "hello"]
+ parallelism: 1
+ # ...
+
+# bolt definitions
+bolts:
+ - id: "splitsentence"
+ className: "org.apache.storm.flux.wrappers.bolts.FluxShellBolt"
+ constructorArgs:
+ # command line
+ - ["python", "splitsentence.py"]
+ # output fields
+ - ["word"]
+ configMethods:
+ - name: "addComponentConfig"
+ args: ["rabbitmq.configfile", "etc/rabbit.yml", "hello"]
+ - name: "addComponentConfig"
+ args:
+ - "publisher.data_paths"
+ - ["actions", "hello"]
+ parallelism: 1
+ # ...
+
+ - id: "log"
+ className: "org.apache.storm.flux.wrappers.bolts.LogInfoBolt"
+ parallelism: 1
+ # ...
+
+ - id: "count"
+ className: "org.apache.storm.testing.TestWordCounter"
+ parallelism: 1
+ # ...
+
+#stream definitions
+# stream definitions define connections between spouts and bolts.
+# note that such connections can be cyclical
+# custom stream groupings are also supported
+
+streams:
+ - name: "spout --> split" # name isn't used (placeholder for logging, UI, etc.)
+ from: "sentence-spout"
+ to: "splitsentence"
+ grouping:
+ type: SHUFFLE
+
+ - name: "split --> count"
+ from: "splitsentence"
+ to: "count"
+ grouping:
+ type: FIELDS
+ args: ["word"]
+
+ - name: "count --> log"
+ from: "count"
+ to: "log"
+ grouping:
+ type: SHUFFLE
http://git-wip-us.apache.org/repos/asf/storm/blob/04d8d570/external/flux/flux-core/src/test/resources/configs/shell_test.yaml
----------------------------------------------------------------------
diff --git a/external/flux/flux-core/src/test/resources/configs/shell_test.yaml b/external/flux/flux-core/src/test/resources/configs/shell_test.yaml
index d885975..dfab397 100644
--- a/external/flux/flux-core/src/test/resources/configs/shell_test.yaml
+++ b/external/flux/flux-core/src/test/resources/configs/shell_test.yaml
@@ -53,6 +53,13 @@ spouts:
- ["node", "randomsentence.js"]
# output fields
- ["word"]
+ configMethods:
+ - name: "addComponentConfig"
+ args: ["rabbitmq.configfile", "etc/rabbit.yml"]
+ - name: "addComponentConfig"
+ args:
+ - "publisher.data_paths"
+ - ["actions"]
parallelism: 1
# ...
@@ -65,6 +72,13 @@ bolts:
- ["python", "splitsentence.py"]
# output fields
- ["word"]
+ configMethods:
+ - name: "addComponentConfig"
+ args: ["rabbitmq.configfile", "etc/rabbit.yml"]
+ - name: "addComponentConfig"
+ args:
+ - "publisher.data_paths"
+ - ["actions"]
parallelism: 1
# ...
http://git-wip-us.apache.org/repos/asf/storm/blob/04d8d570/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
----------------------------------------------------------------------
diff --git a/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java b/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
index 457b8dd..05b8e7a 100644
--- a/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
+++ b/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
@@ -22,9 +22,10 @@ import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
-import java.util.Map;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
/**
* A generic `ShellBolt` implementation that allows you specify output fields
@@ -56,8 +57,59 @@ public class FluxShellBolt extends ShellBolt implements IRichBolt{
this(command);
this.setDefaultStream(outputFields);
}
-
-
+
+ /**
+ * Add configuration for this bolt. This method is called from YAML file:
+ *
+ * ```
+ * className: "org.apache.storm.flux.wrappers.bolts.FluxShellBolt"
+ * constructorArgs:
+ * # command line
+ * - ["python", "splitsentence.py"]
+ * # output fields
+ * - ["word"]
+ * configMethods:
+ * - name: "addComponentConfig"
+ * args: ["publisher.data_paths", "actions"]
+ * ```
+ *
+ * @param key
+ * @param value
+ */
+ public void addComponentConfig(String key, Object value) {
+ if (this.componentConfig == null) {
+ this.componentConfig = new HashMap<String, Object>();
+ }
+ this.componentConfig.put(key, value);
+ }
+
+ /**
+ * Add configuration for this bolt. This method is called from YAML file:
+ *
+ * ```
+ * className: "org.apache.storm.flux.wrappers.bolts.FluxShellBolt"
+ * constructorArgs:
+ * # command line
+ * - ["python", "splitsentence.py"]
+ * # output fields
+ * - ["word"]
+ * configMethods:
+ * - name: "addComponentConfig"
+ * args:
+ * - "publisher.data_paths"
+ * - ["actions"]
+ * ```
+ *
+ * @param key
+ * @param values
+ */
+ public void addComponentConfig(String key, List<Object> values) {
+ if (this.componentConfig == null) {
+ this.componentConfig = new HashMap<String, Object>();
+ }
+ this.componentConfig.put(key, values);
+ }
+
/**
* Set default stream outputFields, this method is called from YAML file:
*
http://git-wip-us.apache.org/repos/asf/storm/blob/04d8d570/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java
----------------------------------------------------------------------
diff --git a/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java b/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java
index d855f12..5fd378d 100644
--- a/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java
+++ b/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java
@@ -22,6 +22,7 @@ import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
+import java.util.List;
import java.util.Map;
import java.util.HashMap;
import java.util.Iterator;
@@ -56,7 +57,59 @@ public class FluxShellSpout extends ShellSpout implements IRichSpout {
this(args);
this.setDefaultStream(outputFields);
}
-
+
+ /**
+ * Add configuration for this spout. This method is called from YAML file:
+ *
+ * ```
+ * className: "org.apache.storm.flux.wrappers.bolts.FluxShellSpout"
+ * constructorArgs:
+ * # command line
+ * - ["python", "splitsentence.py"]
+ * # output fields
+ * - ["word"]
+ * configMethods:
+ * - name: "addComponentConfig"
+ * args: ["publisher.data_paths", "actions"]
+ * ```
+ *
+ * @param key
+ * @param value
+ */
+ public void addComponentConfig(String key, Object value) {
+ if (this.componentConfig == null) {
+ this.componentConfig = new HashMap<String, Object>();
+ }
+ this.componentConfig.put(key, value);
+ }
+
+ /**
+ * Add configuration for this spout. This method is called from YAML file:
+ *
+ * ```
+ * className: "org.apache.storm.flux.wrappers.bolts.FluxShellSpout"
+ * constructorArgs:
+ * # command line
+ * - ["python", "splitsentence.py"]
+ * # output fields
+ * - ["word"]
+ * configMethods:
+ * - name: "addComponentConfig"
+ * args:
+ * - "publisher.data_paths"
+ * - ["actions"]
+ * ```
+ *
+ * @param key
+ * @param values
+ */
+ public void addComponentConfig(String key, List<Object> values) {
+ if (this.componentConfig == null) {
+ this.componentConfig = new HashMap<String, Object>();
+ }
+ this.componentConfig.put(key, values);
+ }
+
/**
* Set default stream outputFields, this method is called from YAML file:
*
[3/3] storm git commit: add STORM-1870 to CHANGELOG
Posted by ka...@apache.org.
add STORM-1870 to CHANGELOG
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/d235f47b
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/d235f47b
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/d235f47b
Branch: refs/heads/1.x-branch
Commit: d235f47bf01a1ceeda01bd07dcc7c98f0be4cdd2
Parents: dd5923b
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Sat Oct 1 13:09:15 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Sat Oct 1 13:09:15 2016 +0900
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/d235f47b/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ee2a951..89ead89 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 1.1.0
+ * STORM-1870: Allow FluxShellBolt/Spout set custom "componentConfig" via yaml
* STORM-2126: fix NPE due to race condition in compute-new-sched-assign\u2026
* STORM-2124: show requested cpu mem for each component
* STORM-2089: Replace Consumer of ISqlTridentDataSource with SqlTridentConsumer
[2/3] storm git commit: Merge branch 'STORM-1870-1.x' into 1.x-branch
Posted by ka...@apache.org.
Merge branch 'STORM-1870-1.x' into 1.x-branch
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/dd5923bd
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/dd5923bd
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/dd5923bd
Branch: refs/heads/1.x-branch
Commit: dd5923bd028509e6c0ac1e923e08663b9298b328
Parents: 2bda84f 04d8d57
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Sat Oct 1 13:08:55 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Sat Oct 1 13:08:55 2016 +0900
----------------------------------------------------------------------
.../java/org/apache/storm/flux/TCKTest.java | 10 ++
.../test/resources/configs/bad_shell_test.yaml | 118 +++++++++++++++++++
.../src/test/resources/configs/shell_test.yaml | 14 +++
.../flux/wrappers/bolts/FluxShellBolt.java | 58 ++++++++-
.../flux/wrappers/spouts/FluxShellSpout.java | 55 ++++++++-
5 files changed, 251 insertions(+), 4 deletions(-)
----------------------------------------------------------------------