You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@heron.apache.org by GitBox <gi...@apache.org> on 2018/04/05 14:35:50 UTC
[GitHub] kramasamy closed pull request #2849: Eco for heron topologies
kramasamy closed pull request #2849: Eco for heron topologies
URL: https://github.com/apache/incubator-heron/pull/2849
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/eco-examples/src/java/BUILD b/eco-heron-examples/src/java/BUILD
similarity index 92%
rename from eco-examples/src/java/BUILD
rename to eco-heron-examples/src/java/BUILD
index 1ac3dbe9d6..1051d3259a 100644
--- a/eco-examples/src/java/BUILD
+++ b/eco-heron-examples/src/java/BUILD
@@ -2,7 +2,7 @@ package(default_visibility = ["//visibility:public"])
filegroup(
name = "heron-eco-examples-support",
- srcs = glob(["**/*.yaml", "**/*.properties"]),
+ srcs = glob(["**/*.yaml"]),
)
java_binary(
diff --git a/eco-heron-examples/src/java/com/twitter/heron/examples/eco/EvenAndOddBolt.java b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/EvenAndOddBolt.java
new file mode 100644
index 0000000000..bd60a493c4
--- /dev/null
+++ b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/EvenAndOddBolt.java
@@ -0,0 +1,77 @@
+// Copyright 2018 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.examples.eco;
+
+
+import java.util.Map;
+
+import com.twitter.heron.api.bolt.BasicOutputCollector;
+import com.twitter.heron.api.bolt.IBasicBolt;
+import com.twitter.heron.api.topology.OutputFieldsDeclarer;
+import com.twitter.heron.api.topology.TopologyContext;
+
+import com.twitter.heron.api.tuple.Fields;
+import com.twitter.heron.api.tuple.Tuple;
+
+import static com.twitter.heron.api.utils.Utils.tuple;
+
+@SuppressWarnings({"serial", "rawtypes", "unchecked"})
+public class EvenAndOddBolt implements IBasicBolt {
+
+
+ @Override
+ public void prepare(Map stormConf, TopologyContext context) {
+
+ }
+
+ protected int getTupleValue(Tuple t, int idx) {
+ return (int) t.getValues().get(idx);
+ }
+
+ @Override
+ public void execute(Tuple input, BasicOutputCollector collector) {
+ int number = getTupleValue(input, 0);
+
+ if (number % 2 == 0) {
+ System.out.println("Emitting to evens stream: " + number);
+ collector.emit("evens", tuple(input.getValues().get(0)));
+
+ } else {
+ System.out.println("emitting to odds stream: " + number);
+ collector.emit("odds", tuple(input.getValues().get(0)));
+ }
+
+ collector.emit(tuple(input.getValues().get(0)));
+
+
+
+ }
+
+ @Override
+ public void cleanup() {
+
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declareStream("evens", new Fields("evens"));
+ declarer.declareStream("odds", new Fields("odds"));
+ declarer.declare(new Fields("number"));
+ }
+
+ @Override
+ public Map<String, Object> getComponentConfiguration() {
+ return null;
+ }
+}
diff --git a/eco-heron-examples/src/java/com/twitter/heron/examples/eco/LogInfoBolt.java b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/LogInfoBolt.java
new file mode 100644
index 0000000000..d7be2c6c69
--- /dev/null
+++ b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/LogInfoBolt.java
@@ -0,0 +1,41 @@
+// Copyright 2018 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.examples.eco;
+
+
+import java.util.logging.Logger;
+
+import com.twitter.heron.api.bolt.BaseBasicBolt;
+import com.twitter.heron.api.bolt.BasicOutputCollector;
+import com.twitter.heron.api.topology.OutputFieldsDeclarer;
+import com.twitter.heron.api.tuple.Tuple;
+
+/**
+ * Simple bolt that does nothing other than LOG.info() every tuple received.
+ *
+ */
+@SuppressWarnings("serial")
+public class LogInfoBolt extends BaseBasicBolt {
+ private static final Logger LOG = Logger.getLogger(LogInfoBolt.class.getName());
+
+ @Override
+ public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
+ LOG.info("{ }" + tuple);
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+
+ }
+}
diff --git a/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestFibonacciSpout.java b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestFibonacciSpout.java
new file mode 100644
index 0000000000..fe19768a63
--- /dev/null
+++ b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestFibonacciSpout.java
@@ -0,0 +1,65 @@
+// Copyright 2018 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.examples.eco;
+
+import java.util.Map;
+import java.util.Random;
+import java.util.logging.Logger;
+
+import com.twitter.heron.api.spout.BaseRichSpout;
+import com.twitter.heron.api.spout.SpoutOutputCollector;
+import com.twitter.heron.api.topology.OutputFieldsDeclarer;
+import com.twitter.heron.api.topology.TopologyContext;
+
+import com.twitter.heron.api.tuple.Fields;
+import com.twitter.heron.api.tuple.Values;
+import com.twitter.heron.api.utils.Utils;
+
+@SuppressWarnings({"serial", "rawtypes", "HiddenField"})
+public class TestFibonacciSpout extends BaseRichSpout {
+ private static final Logger LOG = Logger.getLogger(TestFibonacciSpout.class.getName());
+ private TestPropertyHolder holder;
+ private SpoutOutputCollector collector;
+
+ public TestFibonacciSpout(TestPropertyHolder holder) {
+ this.holder = holder;
+ }
+
+ @Override
+ public void open(Map<String, Object> conf, TopologyContext context,
+ SpoutOutputCollector collector) {
+ this.collector = collector;
+ }
+
+ @Override
+ public void nextTuple() {
+ Utils.sleep(100);
+ final int[] words = new int[] {0, 1, 2, 3, 5, 8, 13, 21, 34};
+ final Random rand = new Random();
+ final int number = words[rand.nextInt(words.length)];
+ final String property = holder.getProperty();
+ final int numberProperty = holder.getNumberProperty();
+ final String publicProperty = holder.publicProperty;
+ LOG.info("Constructor Args: " + property);
+ LOG.info("Property set by setter: " + numberProperty);
+ LOG.info("Property set by public field: " + publicProperty);
+ LOG.info("Emitting: number " + number);
+ collector.emit(new Values(number));
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("number"));
+ }
+}
diff --git a/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestIBasicPrintBolt.java b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestIBasicPrintBolt.java
new file mode 100644
index 0000000000..4f23f4d0aa
--- /dev/null
+++ b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestIBasicPrintBolt.java
@@ -0,0 +1,68 @@
+// Copyright 2018 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.examples.eco;
+
+import java.util.Map;
+
+import com.twitter.heron.api.bolt.BasicOutputCollector;
+import com.twitter.heron.api.bolt.IBasicBolt;
+import com.twitter.heron.api.topology.OutputFieldsDeclarer;
+import com.twitter.heron.api.topology.TopologyContext;
+import com.twitter.heron.api.tuple.Fields;
+import com.twitter.heron.api.tuple.Tuple;
+
+import static com.twitter.heron.api.utils.Utils.tuple;
+
+@SuppressWarnings({"serial", "rawtypes", "unchecked", "HiddenField"})
+public class TestIBasicPrintBolt implements IBasicBolt {
+
+ public String someProperty = "set ";
+
+ public TestUnits testUnits;
+
+ @Override
+ public void prepare(Map stormConf, TopologyContext context) {
+
+ }
+
+ public void sampleConfigurationMethod(String someProperty, TestUnits testUnits) {
+ this.someProperty += someProperty;
+ this.testUnits = testUnits;
+ }
+
+ @Override
+ public void execute(Tuple input, BasicOutputCollector collector) {
+ System.out.println("The configuration method has set \"someProperty\" to : "
+ + this.someProperty);
+ System.out.println("The configuration method has set TestUnits to " + testUnits);
+ System.out.println("Emitting : " + input);
+ collector.emit(tuple(input.getValues().get(0)));
+
+ }
+
+ @Override
+ public void cleanup() {
+
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("ibasic-print"));
+ }
+
+ @Override
+ public Map<String, Object> getComponentConfiguration() {
+ return null;
+ }
+}
diff --git a/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestNameCounter.java b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestNameCounter.java
new file mode 100644
index 0000000000..ba02368bc8
--- /dev/null
+++ b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestNameCounter.java
@@ -0,0 +1,61 @@
+// Copyright 2018 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.examples.eco;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.twitter.heron.api.bolt.BaseBasicBolt;
+import com.twitter.heron.api.bolt.BasicOutputCollector;
+import com.twitter.heron.api.topology.OutputFieldsDeclarer;
+import com.twitter.heron.api.topology.TopologyContext;
+import com.twitter.heron.api.tuple.Fields;
+import com.twitter.heron.api.tuple.Tuple;
+
+import static com.twitter.heron.api.utils.Utils.tuple;
+@SuppressWarnings({"serial", "rawtypes"})
+public class TestNameCounter extends BaseBasicBolt {
+
+ private Map<String, Integer> counts;
+
+ @Override
+ public void prepare(Map map, TopologyContext topologyContext) {
+ counts = new HashMap<>();
+ }
+
+
+ protected String getTupleValue(Tuple t, int idx) {
+ return (String) t.getValues().get(idx);
+ }
+
+ public void execute(Tuple input, BasicOutputCollector collector) {
+ String word = getTupleValue(input, 0);
+ int count = 0;
+ if (counts.containsKey(word)) {
+ count = counts.get(word);
+ }
+ count++;
+ counts.put(word, count);
+ collector.emit(tuple(word, count));
+ }
+
+ public void cleanup() {
+
+ }
+
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("name", "count"));
+ }
+
+}
diff --git a/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestNameSpout.java b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestNameSpout.java
new file mode 100644
index 0000000000..355d9b181e
--- /dev/null
+++ b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestNameSpout.java
@@ -0,0 +1,81 @@
+// Copyright 2017 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.examples.eco;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+
+import com.twitter.heron.api.Config;
+import com.twitter.heron.api.spout.BaseRichSpout;
+import com.twitter.heron.api.spout.SpoutOutputCollector;
+import com.twitter.heron.api.topology.OutputFieldsDeclarer;
+import com.twitter.heron.api.topology.TopologyContext;
+import com.twitter.heron.api.tuple.Fields;
+import com.twitter.heron.api.tuple.Values;
+import com.twitter.heron.api.utils.Utils;
+
+@SuppressWarnings({"serial", "HiddenField"})
+public class TestNameSpout extends BaseRichSpout {
+ private boolean isdistributed;
+ private SpoutOutputCollector collector;
+
+ public TestNameSpout() {
+ this(true);
+ }
+
+ public TestNameSpout(boolean isDistributed) {
+ isdistributed = isDistributed;
+ }
+
+ public void open(Map<String, Object> conf, TopologyContext context,
+ SpoutOutputCollector collector) {
+ this.collector = collector;
+ }
+
+ public void close() {
+
+ }
+
+ public void nextTuple() {
+ Utils.sleep(100);
+ final String[] words = new String[] {"marge", "homer", "bart", "simpson", "lisa"};
+ final Random rand = new Random();
+ final String word = words[rand.nextInt(words.length)];
+ collector.emit(new Values(word));
+ }
+
+ public void ack(Object msgId) {
+
+ }
+
+ public void fail(Object msgId) {
+
+ }
+
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("name"));
+ }
+
+ @Override
+ public Map<String, Object> getComponentConfiguration() {
+ if (!isdistributed) {
+ Map<String, Object> ret = new HashMap<String, Object>();
+ ret.put(Config.TOPOLOGY_STMGRS, 1);
+ return ret;
+ } else {
+ return null;
+ }
+ }
+}
diff --git a/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestPrintBolt.java b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestPrintBolt.java
new file mode 100644
index 0000000000..b48cf31287
--- /dev/null
+++ b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestPrintBolt.java
@@ -0,0 +1,33 @@
+// Copyright 2018 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.examples.eco;
+
+import com.twitter.heron.api.bolt.BaseBasicBolt;
+import com.twitter.heron.api.bolt.BasicOutputCollector;
+import com.twitter.heron.api.topology.OutputFieldsDeclarer;
+import com.twitter.heron.api.tuple.Tuple;
+
+@SuppressWarnings("serial")
+public class TestPrintBolt extends BaseBasicBolt {
+
+ @Override
+ public void execute(Tuple tuple, BasicOutputCollector collector) {
+ System.out.println(tuple);
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer ofd) {
+ }
+
+}
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/TestPropertyHolder.java b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestPropertyHolder.java
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/TestPropertyHolder.java
rename to eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestPropertyHolder.java
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/TestUnits.java b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestUnits.java
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/TestUnits.java
rename to eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestUnits.java
diff --git a/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestWindowBolt.java b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestWindowBolt.java
new file mode 100644
index 0000000000..34d5c76d7a
--- /dev/null
+++ b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/TestWindowBolt.java
@@ -0,0 +1,46 @@
+// Copyright 2018 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.examples.eco;
+
+import java.util.Map;
+
+import com.twitter.heron.api.bolt.BaseWindowedBolt;
+import com.twitter.heron.api.bolt.OutputCollector;
+import com.twitter.heron.api.topology.OutputFieldsDeclarer;
+import com.twitter.heron.api.topology.TopologyContext;
+import com.twitter.heron.api.tuple.Fields;
+import com.twitter.heron.api.tuple.Values;
+import com.twitter.heron.api.windowing.TupleWindow;
+
+@SuppressWarnings({"serial", "HiddenField"})
+public class TestWindowBolt extends BaseWindowedBolt {
+ private OutputCollector collector;
+
+
+ @Override
+ public void prepare(Map<String, Object> topoConf, TopologyContext context,
+ OutputCollector collector) {
+ this.collector = collector;
+ }
+
+ @Override
+ public void execute(TupleWindow inputWindow) {
+ collector.emit(new Values(inputWindow.get().size()));
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("count"));
+ }
+}
diff --git a/eco-heron-examples/src/java/com/twitter/heron/examples/eco/heron_fibonacci.yaml b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/heron_fibonacci.yaml
new file mode 100644
index 0000000000..7a9f9a2618
--- /dev/null
+++ b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/heron_fibonacci.yaml
@@ -0,0 +1,75 @@
+# Copyright 2017 Twitter. All rights reserved.
+#
+# Licensed 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.
+
+---
+
+name: "heron-fibonacci-topology"
+type: "heron"
+
+config:
+ topology.workers: 1
+
+components:
+ - id: "property-holder"
+ className: "com.twitter.heron.examples.eco.TestPropertyHolder"
+ constructorArgs:
+ - "some argument"
+ properties:
+ - name: "numberProperty"
+ value: 11
+ - name: "publicProperty"
+ value: "This is public property"
+
+spouts:
+ - id: "spout-1"
+ className: "com.twitter.heron.examples.eco.TestFibonacciSpout"
+ constructorArgs:
+ - ref: "property-holder"
+ parallelism: 1
+
+bolts:
+ - id: "even-and-odd-bolt"
+ className: "com.twitter.heron.examples.eco.EvenAndOddBolt"
+ parallelism: 1
+
+ - id: "ibasic-print-bolt"
+ className: "com.twitter.heron.examples.eco.TestIBasicPrintBolt"
+ parallelism: 1
+ configMethods:
+ - name: "sampleConfigurationMethod"
+ args:
+ - "${ecoPropertyOne}"
+ - MB
+
+ - id: "sys-out-bolt"
+ className: "com.twitter.heron.examples.eco.TestPrintBolt"
+ parallelism: 1
+
+streams:
+ - from: "spout-1"
+ to: "even-and-odd-bolt"
+ grouping:
+ type: SHUFFLE
+
+ - from: "even-and-odd-bolt"
+ to: "ibasic-print-bolt"
+ grouping:
+ type: SHUFFLE
+ streamId: "odds"
+
+ - from: "even-and-odd-bolt"
+ to: "sys-out-bolt"
+ grouping:
+ type: SHUFFLE
+ streamId: "evens"
diff --git a/eco-heron-examples/src/java/com/twitter/heron/examples/eco/heron_windowing.yaml b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/heron_windowing.yaml
new file mode 100644
index 0000000000..49adf3ccc1
--- /dev/null
+++ b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/heron_windowing.yaml
@@ -0,0 +1,66 @@
+# Copyright 2017 Twitter. All rights reserved.
+#
+# Licensed 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.
+
+---
+
+name: "heron-sliding-window-topology"
+type: "heron"
+
+components:
+ - id: "windowLength"
+ className: "com.twitter.heron.api.bolt.BaseWindowedBolt$Count"
+ constructorArgs:
+ - 5
+ - id: "slidingInterval"
+ className: "com.twitter.heron.api.bolt.BaseWindowedBolt$Count"
+ constructorArgs:
+ - 3
+
+config:
+ topology.workers: 1
+
+# spout definitions
+spouts:
+ - id: "spout-1"
+ className: "com.twitter.heron.examples.eco.TestNameSpout"
+ parallelism: 1
+
+# bolt definitions
+bolts:
+ - id: "bolt-1"
+ className: "com.twitter.heron.examples.eco.TestWindowBolt"
+ configMethods:
+ - name: "withWindow"
+ args: [ref: "windowLength", ref: "slidingInterval"]
+ parallelism: 1
+ - id: "bolt-2"
+ className: "com.twitter.heron.examples.eco.TestPrintBolt"
+ parallelism: 1
+
+
+#stream definitions
+# stream definitions define connections between spouts and bolts.
+# note that such connections can be cyclical
+streams:
+
+ - from: "spout-1"
+ to: "bolt-1"
+ grouping:
+ type: FIELDS
+ args: ["word"]
+
+ - from: "bolt-1"
+ to: "bolt-2"
+ grouping:
+ type: SHUFFLE
diff --git a/eco-heron-examples/src/java/com/twitter/heron/examples/eco/heron_wordcount.yaml b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/heron_wordcount.yaml
new file mode 100644
index 0000000000..a195699165
--- /dev/null
+++ b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/heron_wordcount.yaml
@@ -0,0 +1,73 @@
+# Copyright 2017 Twitter. All rights reserved.
+#
+# Licensed 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.
+
+---
+
+# topology definition
+# name to be used when submitting
+name: "heron-simple-wordcount-topology"
+type: "heron"
+
+# topology configuration
+# this will be passed to the submitter as a map of config options
+#
+config:
+ topology.workers: 2
+ topology.component.resourcemap:
+
+ - id: "spout-1"
+ ram: 256MB # The minimum value for a component's specified ram is 256MB
+ cpu: 0.5
+ disk: 4GB
+
+ - id: "bolt-1"
+ ram: 256MB # The minimum value for a component's specified ram is 256MB
+ cpu: 0.5
+ disk: 2GB
+
+ topology.component.jvmoptions:
+
+ - id: "spout-1"
+ options: ["-XX:NewSize=300m", "-Xms2g"]
+
+# spout definitions
+spouts:
+ - id: "spout-1"
+ className: "com.twitter.heron.examples.eco.TestNameSpout"
+ parallelism: 1
+
+# bolt definitions
+bolts:
+ - id: "bolt-1"
+ className: "com.twitter.heron.examples.eco.TestNameCounter"
+ parallelism: 1
+
+ - id: "bolt-2"
+ className: "com.twitter.heron.examples.eco.LogInfoBolt"
+ parallelism: 1
+
+#stream definitions
+# stream definitions define connections between spouts and bolts.
+# note that such connections can be cyclical
+streams:
+ - from: "spout-1"
+ to: "bolt-1"
+ grouping:
+ type: FIELDS
+ args: ["word"]
+
+ - from: "bolt-1"
+ to: "bolt-2"
+ grouping:
+ type: SHUFFLE
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/sample.properties b/eco-heron-examples/src/java/com/twitter/heron/examples/eco/sample.properties
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/sample.properties
rename to eco-heron-examples/src/java/com/twitter/heron/examples/eco/sample.properties
diff --git a/eco-storm-examples/src/java/BUILD b/eco-storm-examples/src/java/BUILD
new file mode 100644
index 0000000000..6ef748d8d3
--- /dev/null
+++ b/eco-storm-examples/src/java/BUILD
@@ -0,0 +1,25 @@
+package(default_visibility = ["//visibility:public"])
+
+filegroup(
+ name = "storm-eco-examples-support",
+ srcs = glob(["**/*.yaml", "**/*.properties"]),
+)
+
+java_binary(
+ name='storm-eco-examples-unshaded',
+ srcs = glob(["com/twitter/heron/examples/eco/**/*.java"]),
+ deps = [
+ "//heron/api/src/java:api-java-low-level",
+ "//heron/api/src/java:api-java",
+ "//heron/common/src/java:basics-java",
+ "//storm-compatibility/src/java:storm-compatibility-java",
+ ],
+ create_executable = 0,
+)
+
+genrule(
+ name = 'storm-eco-examples',
+ srcs = [":storm-eco-examples-unshaded_deploy.jar"],
+ outs = ["storm-eco-examples.jar"],
+ cmd = "cp $< $@",
+)
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/EvenAndOddBolt.java b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/EvenAndOddBolt.java
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/EvenAndOddBolt.java
rename to eco-storm-examples/src/java/com/twitter/heron/examples/eco/EvenAndOddBolt.java
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/LogInfoBolt.java b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/LogInfoBolt.java
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/LogInfoBolt.java
rename to eco-storm-examples/src/java/com/twitter/heron/examples/eco/LogInfoBolt.java
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/TestFibonacciSpout.java b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestFibonacciSpout.java
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/TestFibonacciSpout.java
rename to eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestFibonacciSpout.java
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/TestIBasicPrintBolt.java b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestIBasicPrintBolt.java
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/TestIBasicPrintBolt.java
rename to eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestIBasicPrintBolt.java
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/TestNameCounter.java b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestNameCounter.java
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/TestNameCounter.java
rename to eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestNameCounter.java
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/TestNameSpout.java b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestNameSpout.java
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/TestNameSpout.java
rename to eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestNameSpout.java
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/TestPrintBolt.java b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestPrintBolt.java
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/TestPrintBolt.java
rename to eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestPrintBolt.java
diff --git a/eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestPropertyHolder.java b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestPropertyHolder.java
new file mode 100644
index 0000000000..2020fe489d
--- /dev/null
+++ b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestPropertyHolder.java
@@ -0,0 +1,48 @@
+// Copyright 2018 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.examples.eco;
+
+import java.io.Serializable;
+
+@SuppressWarnings("serial")
+public class TestPropertyHolder implements Serializable {
+
+ private String property;
+
+ private int numberProperty;
+
+ public String publicProperty;
+
+
+ public TestPropertyHolder(String property) {
+ this.property = property;
+ }
+
+ public String getProperty() {
+ return property;
+ }
+
+ public void setProperty(String property) {
+ this.property = property;
+ }
+
+ public int getNumberProperty() {
+ return numberProperty;
+ }
+
+ public void setNumberProperty(int numberProperty) {
+ this.numberProperty = numberProperty;
+ }
+
+}
diff --git a/eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestUnits.java b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestUnits.java
new file mode 100644
index 0000000000..f1d469b6c0
--- /dev/null
+++ b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestUnits.java
@@ -0,0 +1,33 @@
+// Copyright 2018 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.examples.eco;
+
+/**
+ * Sample unit of measure enum
+ */
+public enum TestUnits {
+ MB("MB"),
+ GB("GB"),
+ B("B");
+
+ private String value;
+
+ TestUnits(String value) {
+ this.value = value;
+ }
+
+ public String getValue() {
+ return value;
+ }
+}
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/TestWindowBolt.java b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestWindowBolt.java
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/TestWindowBolt.java
rename to eco-storm-examples/src/java/com/twitter/heron/examples/eco/TestWindowBolt.java
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/fibonacci.yaml b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/fibonacci.yaml
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/fibonacci.yaml
rename to eco-storm-examples/src/java/com/twitter/heron/examples/eco/fibonacci.yaml
diff --git a/eco-storm-examples/src/java/com/twitter/heron/examples/eco/sample.properties b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/sample.properties
new file mode 100644
index 0000000000..65a85c4df6
--- /dev/null
+++ b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/sample.properties
@@ -0,0 +1,18 @@
+# // Copyright 2018 Twitter. All rights reserved.
+# //
+# // Licensed 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.
+
+ecoPropertyOne=thisValueWasSetFromAPropertiesFile
+
+ecoPropertyTwo=1234
+
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/simple_windowing.yaml b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/simple_windowing.yaml
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/simple_windowing.yaml
rename to eco-storm-examples/src/java/com/twitter/heron/examples/eco/simple_windowing.yaml
diff --git a/eco-examples/src/java/com/twitter/heron/examples/eco/simple_wordcount.yaml b/eco-storm-examples/src/java/com/twitter/heron/examples/eco/simple_wordcount.yaml
similarity index 100%
rename from eco-examples/src/java/com/twitter/heron/examples/eco/simple_wordcount.yaml
rename to eco-storm-examples/src/java/com/twitter/heron/examples/eco/simple_wordcount.yaml
diff --git a/eco/src/java/BUILD b/eco/src/java/BUILD
index d5c81b9b25..158fe1a214 100644
--- a/eco/src/java/BUILD
+++ b/eco/src/java/BUILD
@@ -11,8 +11,57 @@ eco_deps = [
"//heron/common/src/java:basics-java",
]
+java_library(
+ name = "eco-defs-java",
+ srcs = glob(["com/twitter/heron/eco/definition/*.java"]),
+ deps = eco_deps,
+)
+
+java_library(
+ name = "eco-parser-java",
+ srcs = glob(["com/twitter/heron/eco/parser/*.java"]),
+ deps = eco_deps + [":eco-defs-java"],
+)
+
+java_library(
+ name = "eco-builder-java",
+ srcs = glob(["com/twitter/heron/eco/builder/*.java"]),
+ deps = eco_deps + [":eco-defs-java"],
+)
+
+java_library(
+ name = "eco-storm-builder-java",
+ srcs = glob(["com/twitter/heron/eco/builder/storm/*.java"]),
+ deps = eco_deps + [
+ ":eco-builder-java",
+ ":eco-defs-java",
+ ],
+)
+
+java_library(
+ name = "eco-heron-builder-java",
+ srcs = glob(["com/twitter/heron/eco/builder/heron/*.java"]),
+ deps = eco_deps + [
+ ":eco-builder-java",
+ ":eco-defs-java",
+ ],
+)
+
+java_library(
+ name = "eco-submit-java",
+ srcs = glob(["com/twitter/heron/eco/submit/*.java"]),
+ deps = eco_deps,
+)
+
java_library(
name = "eco-java",
- srcs = glob(["com/twitter/heron/eco/**/*.java"]),
- deps = eco_deps
+ srcs = glob(["com/twitter/heron/eco/*.java"]),
+ deps = eco_deps + [
+ ":eco-defs-java",
+ ":eco-parser-java",
+ ":eco-builder-java",
+ ":eco-heron-builder-java",
+ ":eco-storm-builder-java",
+ ":eco-submit-java",
+ ],
)
diff --git a/eco/src/java/com/twitter/heron/eco/Eco.java b/eco/src/java/com/twitter/heron/eco/Eco.java
index f5cb6bbbfb..77b4dd8906 100644
--- a/eco/src/java/com/twitter/heron/eco/Eco.java
+++ b/eco/src/java/com/twitter/heron/eco/Eco.java
@@ -15,6 +15,7 @@
import java.io.File;
import java.io.FileInputStream;
+import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.commons.cli.CommandLine;
@@ -23,17 +24,13 @@
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
-import org.apache.storm.topology.TopologyBuilder;
import com.twitter.heron.api.Config;
import com.twitter.heron.eco.builder.BoltBuilder;
import com.twitter.heron.eco.builder.BuilderUtility;
import com.twitter.heron.eco.builder.ComponentBuilder;
import com.twitter.heron.eco.builder.ConfigBuilder;
-import com.twitter.heron.eco.builder.EcoBuilder;
import com.twitter.heron.eco.builder.ObjectBuilder;
-import com.twitter.heron.eco.builder.SpoutBuilder;
-import com.twitter.heron.eco.builder.StreamBuilder;
import com.twitter.heron.eco.definition.BoltDefinition;
import com.twitter.heron.eco.definition.EcoExecutionContext;
import com.twitter.heron.eco.definition.EcoTopologyDefinition;
@@ -50,16 +47,22 @@
private static final String ENV_PROPS = "env-props";
private static final String ECO_CONFIG_FILE = "eco-config-file";
- private EcoBuilder ecoBuilder;
private EcoParser ecoParser;
private EcoSubmitter ecoSubmitter;
- public Eco(EcoBuilder ecoBuilder, EcoParser ecoParser, EcoSubmitter ecoSubmitter) {
- this.ecoBuilder = ecoBuilder;
+ public Eco(EcoParser ecoParser, EcoSubmitter ecoSubmitter) {
this.ecoParser = ecoParser;
this.ecoSubmitter = ecoSubmitter;
}
+ /**
+ * Submit an ECO topology
+ *
+ * @param fileInputStream The input stream associated with ECO topology definition file
+ * @param propertiesFile The optional key-value property file for optional property substitution.
+ * @param envFilter The optional flag to tell ECO to perform environment variable substitution
+ * @throws Exception
+ */
public void submit(FileInputStream fileInputStream,
FileInputStream propertiesFile, boolean envFilter)
throws Exception {
@@ -67,21 +70,61 @@ public void submit(FileInputStream fileInputStream,
.parseFromInputStream(fileInputStream, propertiesFile, envFilter);
String topologyName = topologyDefinition.getName();
+ String topologyType = topologyDefinition.getType();
- Config topologyConfig = ecoBuilder
- .buildConfig(topologyDefinition);
+ if ("storm".equals(topologyType)) {
+ System.out.println("topology type is Storm");
+ com.twitter.heron.eco.builder.storm.EcoBuilder ecoBuilder =
+ new com.twitter.heron.eco.builder.storm.EcoBuilder(
+ new com.twitter.heron.eco.builder.storm.SpoutBuilder(),
+ new BoltBuilder(),
+ new com.twitter.heron.eco.builder.storm.StreamBuilder(),
+ new ComponentBuilder(),
+ new ConfigBuilder());
+
+ Config topologyConfig = ecoBuilder
+ .buildConfig(topologyDefinition);
+
+ EcoExecutionContext executionContext
+ = new EcoExecutionContext(topologyDefinition, topologyConfig);
- EcoExecutionContext executionContext
- = new EcoExecutionContext(topologyDefinition, topologyConfig);
+ printTopologyInfo(executionContext);
- printTopologyInfo(executionContext);
+ ObjectBuilder objectBuilder = new ObjectBuilder();
+ objectBuilder.setBuilderUtility(new BuilderUtility());
+
+ org.apache.storm.topology.TopologyBuilder builder = ecoBuilder
+ .buildTopologyBuilder(executionContext, objectBuilder);
+ ecoSubmitter.submitStormTopology(topologyName, topologyConfig, builder.createTopology());
+ } else if ("heron".equals(topologyType)) {
+ System.out.println("topology type is Heron");
+ com.twitter.heron.eco.builder.heron.EcoBuilder ecoBuilder =
+ new com.twitter.heron.eco.builder.heron.EcoBuilder(
+ new com.twitter.heron.eco.builder.heron.SpoutBuilder(),
+ new BoltBuilder(),
+ new com.twitter.heron.eco.builder.heron.StreamBuilder(),
+ new ComponentBuilder(),
+ new ConfigBuilder());
- ObjectBuilder objectBuilder = new ObjectBuilder();
- objectBuilder.setBuilderUtility(new BuilderUtility());
- TopologyBuilder builder = ecoBuilder
- .buildTopologyBuilder(executionContext, objectBuilder);
+ Config topologyConfig = ecoBuilder
+ .buildConfig(topologyDefinition);
- ecoSubmitter.submitTopology(topologyName, topologyConfig, builder.createTopology());
+ EcoExecutionContext executionContext
+ = new EcoExecutionContext(topologyDefinition, topologyConfig);
+
+ printTopologyInfo(executionContext);
+
+ ObjectBuilder objectBuilder = new ObjectBuilder();
+ objectBuilder.setBuilderUtility(new BuilderUtility());
+
+ com.twitter.heron.api.topology.TopologyBuilder builder = ecoBuilder
+ .buildTopologyBuilder(executionContext, objectBuilder);
+ ecoSubmitter.submitHeronTopology(topologyName, topologyConfig, builder.createTopology());
+ } else {
+ LOG.log(Level.SEVERE,
+ String.format("Unknown topology type \'%s\' for topology %s, not submitted",
+ topologyType, topologyName));
+ }
}
public static void main(String[] args) throws Exception {
@@ -109,15 +152,7 @@ public static void main(String[] args) throws Exception {
Boolean filterFromEnv = cmd.hasOption(ENV_PROPS);
- Eco eco = new Eco(
- new EcoBuilder(
- new SpoutBuilder(),
- new BoltBuilder(),
- new StreamBuilder(),
- new ComponentBuilder(),
- new ConfigBuilder()),
- new EcoParser(),
- new EcoSubmitter());
+ Eco eco = new Eco(new EcoParser(), new EcoSubmitter());
eco.submit(fin, propsInputStream, filterFromEnv);
}
diff --git a/eco/src/java/com/twitter/heron/eco/builder/BoltBuilder.java b/eco/src/java/com/twitter/heron/eco/builder/BoltBuilder.java
index 6af6ac9b90..5259db5f9f 100644
--- a/eco/src/java/com/twitter/heron/eco/builder/BoltBuilder.java
+++ b/eco/src/java/com/twitter/heron/eco/builder/BoltBuilder.java
@@ -21,7 +21,7 @@
public class BoltBuilder {
- protected void buildBolts(EcoExecutionContext executionContext,
+ public void buildBolts(EcoExecutionContext executionContext,
ObjectBuilder objectBuilder)
throws IllegalAccessException, InstantiationException, ClassNotFoundException,
NoSuchFieldException, InvocationTargetException {
diff --git a/eco/src/java/com/twitter/heron/eco/builder/ComponentBuilder.java b/eco/src/java/com/twitter/heron/eco/builder/ComponentBuilder.java
index 9fe602303a..3e7be34a81 100644
--- a/eco/src/java/com/twitter/heron/eco/builder/ComponentBuilder.java
+++ b/eco/src/java/com/twitter/heron/eco/builder/ComponentBuilder.java
@@ -20,7 +20,7 @@
import com.twitter.heron.eco.definition.EcoExecutionContext;
public class ComponentBuilder {
- protected void buildComponents(EcoExecutionContext context, ObjectBuilder objectBuilder)
+ public void buildComponents(EcoExecutionContext context, ObjectBuilder objectBuilder)
throws ClassNotFoundException,
IllegalAccessException, InstantiationException,
NoSuchFieldException, InvocationTargetException {
diff --git a/eco/src/java/com/twitter/heron/eco/builder/ConfigBuilder.java b/eco/src/java/com/twitter/heron/eco/builder/ConfigBuilder.java
index bc8dd43ca1..787ecb0820 100644
--- a/eco/src/java/com/twitter/heron/eco/builder/ConfigBuilder.java
+++ b/eco/src/java/com/twitter/heron/eco/builder/ConfigBuilder.java
@@ -44,7 +44,12 @@
private static final Integer MINIMUM_BYTES = 256000000;
private static final Integer MINIMUM_MB = 256;
- protected Config buildConfig(EcoTopologyDefinition topologyDefinition)
+ /**
+ * Build the config for a ECO topology definition
+ *
+ * @param topologyDefinition - ECO topology definition
+ */
+ public Config buildConfig(EcoTopologyDefinition topologyDefinition)
throws IllegalArgumentException {
Map<String, Object> configMap = topologyDefinition.getConfig();
diff --git a/eco/src/java/com/twitter/heron/eco/builder/heron/EcoBuilder.java b/eco/src/java/com/twitter/heron/eco/builder/heron/EcoBuilder.java
new file mode 100644
index 0000000000..e21cde98c8
--- /dev/null
+++ b/eco/src/java/com/twitter/heron/eco/builder/heron/EcoBuilder.java
@@ -0,0 +1,79 @@
+// Copyright 2017 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.eco.builder.heron;
+
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.logging.Logger;
+
+import com.twitter.heron.api.Config;
+import com.twitter.heron.api.topology.TopologyBuilder;
+
+import com.twitter.heron.eco.builder.BoltBuilder;
+import com.twitter.heron.eco.builder.ComponentBuilder;
+import com.twitter.heron.eco.builder.ConfigBuilder;
+import com.twitter.heron.eco.builder.ObjectBuilder;
+
+import com.twitter.heron.eco.definition.EcoExecutionContext;
+import com.twitter.heron.eco.definition.EcoTopologyDefinition;
+
+
+public class EcoBuilder {
+
+ private SpoutBuilder spoutBuilder;
+
+ private BoltBuilder boltBuilder;
+
+ private StreamBuilder streamBuilder;
+
+ private ComponentBuilder componentBuilder;
+
+ private ConfigBuilder configBuilder;
+
+ private static final Logger LOG = Logger.getLogger(EcoBuilder.class.getName());
+
+ public EcoBuilder(SpoutBuilder spoutBuilder, BoltBuilder boltBuilder,
+ StreamBuilder streamBuilder, ComponentBuilder componentBuilder,
+ ConfigBuilder configBuilder) {
+ this.spoutBuilder = spoutBuilder;
+ this.boltBuilder = boltBuilder;
+ this.streamBuilder = streamBuilder;
+ this.componentBuilder = componentBuilder;
+ this.configBuilder = configBuilder;
+ }
+
+ public TopologyBuilder buildTopologyBuilder(EcoExecutionContext executionContext,
+ ObjectBuilder objectBuilder)
+ throws InstantiationException, IllegalAccessException,
+ ClassNotFoundException,
+ NoSuchFieldException, InvocationTargetException {
+
+ TopologyBuilder builder = new TopologyBuilder();
+ LOG.info("Building components");
+ componentBuilder.buildComponents(executionContext, objectBuilder);
+ LOG.info("Building spouts");
+ spoutBuilder.buildSpouts(executionContext, builder, objectBuilder);
+ LOG.info("Building bolts");
+ boltBuilder.buildBolts(executionContext, objectBuilder);
+ LOG.info("Building streams");
+ streamBuilder.buildStreams(executionContext, builder, objectBuilder);
+
+ return builder;
+ }
+
+ public Config buildConfig(EcoTopologyDefinition topologyDefinition) throws Exception {
+ LOG.info("Building topology config");
+ return this.configBuilder.buildConfig(topologyDefinition);
+ }
+}
diff --git a/eco/src/java/com/twitter/heron/eco/builder/heron/SpoutBuilder.java b/eco/src/java/com/twitter/heron/eco/builder/heron/SpoutBuilder.java
new file mode 100644
index 0000000000..2289a56a1b
--- /dev/null
+++ b/eco/src/java/com/twitter/heron/eco/builder/heron/SpoutBuilder.java
@@ -0,0 +1,43 @@
+// Copyright 2017 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.eco.builder.heron;
+
+import java.lang.reflect.InvocationTargetException;
+
+import com.twitter.heron.api.spout.IRichSpout;
+import com.twitter.heron.api.topology.TopologyBuilder;
+
+import com.twitter.heron.eco.builder.ObjectBuilder;
+
+import com.twitter.heron.eco.definition.EcoExecutionContext;
+import com.twitter.heron.eco.definition.EcoTopologyDefinition;
+import com.twitter.heron.eco.definition.ObjectDefinition;
+
+
+public class SpoutBuilder {
+
+ protected void buildSpouts(EcoExecutionContext executionContext,
+ TopologyBuilder builder,
+ ObjectBuilder objectBuilder)
+ throws ClassNotFoundException, InstantiationException, IllegalAccessException,
+ NoSuchFieldException, InvocationTargetException {
+ EcoTopologyDefinition topologyDefinition = executionContext.getTopologyDefinition();
+
+ for (ObjectDefinition def: topologyDefinition.getSpouts()) {
+ Object obj = objectBuilder.buildObject(def, executionContext);
+ builder.setSpout(def.getId(), (IRichSpout) obj, def.getParallelism());
+ executionContext.addSpout(def.getId(), obj);
+ }
+ }
+}
diff --git a/eco/src/java/com/twitter/heron/eco/builder/heron/StreamBuilder.java b/eco/src/java/com/twitter/heron/eco/builder/heron/StreamBuilder.java
new file mode 100644
index 0000000000..f48852343f
--- /dev/null
+++ b/eco/src/java/com/twitter/heron/eco/builder/heron/StreamBuilder.java
@@ -0,0 +1,129 @@
+// Copyright 2017 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.eco.builder.heron;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.twitter.heron.api.bolt.IBasicBolt;
+import com.twitter.heron.api.bolt.IRichBolt;
+import com.twitter.heron.api.bolt.IWindowedBolt;
+import com.twitter.heron.api.grouping.CustomStreamGrouping;
+import com.twitter.heron.api.topology.BoltDeclarer;
+import com.twitter.heron.api.topology.TopologyBuilder;
+import com.twitter.heron.api.tuple.Fields;
+import com.twitter.heron.api.utils.Utils;
+
+import com.twitter.heron.eco.builder.ObjectBuilder;
+
+import com.twitter.heron.eco.definition.ComponentStream;
+import com.twitter.heron.eco.definition.EcoExecutionContext;
+import com.twitter.heron.eco.definition.EcoTopologyDefinition;
+import com.twitter.heron.eco.definition.GroupingDefinition;
+import com.twitter.heron.eco.definition.ObjectDefinition;
+import com.twitter.heron.eco.definition.StreamDefinition;
+
+public class StreamBuilder {
+
+ protected void buildStreams(EcoExecutionContext executionContext, TopologyBuilder builder,
+ ObjectBuilder objectBuilder)
+ throws IllegalAccessException, InstantiationException, ClassNotFoundException,
+ NoSuchFieldException, InvocationTargetException {
+ EcoTopologyDefinition topologyDefinition = executionContext.getTopologyDefinition();
+ Map<String, ComponentStream> componentStreams = new HashMap<>();
+
+ HashMap<String, BoltDeclarer> declarers = new HashMap<>();
+ for (StreamDefinition stream : topologyDefinition.getStreams()) {
+ Object boltObj = executionContext.getBolt(stream.getTo());
+ BoltDeclarer declarer = declarers.get(stream.getTo());
+ if (boltObj instanceof IRichBolt) {
+ if (declarer == null) {
+ declarer = builder.setBolt(stream.getTo(),
+ (IRichBolt) boltObj,
+ topologyDefinition.parallelismForBolt(stream.getTo()));
+ declarers.put(stream.getTo(), declarer);
+ }
+ } else if (boltObj instanceof IBasicBolt) {
+ if (declarer == null) {
+ declarer = builder.setBolt(
+ stream.getTo(),
+ (IBasicBolt) boltObj,
+ topologyDefinition.parallelismForBolt(stream.getTo()));
+ declarers.put(stream.getTo(), declarer);
+ }
+ } else if (boltObj instanceof IWindowedBolt) {
+ if (declarer == null) {
+ declarer = builder.setBolt(
+ stream.getTo(),
+ (IWindowedBolt) boltObj,
+ topologyDefinition.parallelismForBolt(stream.getTo()));
+ declarers.put(stream.getTo(), declarer);
+ }
+ } else {
+ throw new IllegalArgumentException("Class does not appear to be a bolt: "
+ + boltObj.getClass().getName());
+ }
+
+ GroupingDefinition grouping = stream.getGrouping();
+ // if the streamId is defined, use it for the grouping,
+ // otherwise assume default stream
+ String streamId = grouping.getStreamId() == null
+ ? Utils.DEFAULT_STREAM_ID : grouping.getStreamId();
+
+
+ switch (grouping.getType()) {
+ case SHUFFLE:
+ declarer.shuffleGrouping(stream.getFrom(), streamId);
+ break;
+ case FIELDS:
+ List<String> groupingArgs = grouping.getArgs();
+ if (groupingArgs == null) {
+ throw new IllegalArgumentException("You must supply arguments for Fields grouping");
+ }
+ declarer.fieldsGrouping(stream.getFrom(), streamId, new Fields(groupingArgs));
+ break;
+ case ALL:
+ declarer.allGrouping(stream.getFrom(), streamId);
+ break;
+ case GLOBAL:
+ declarer.globalGrouping(stream.getFrom(), streamId);
+ break;
+ case NONE:
+ declarer.noneGrouping(stream.getFrom(), streamId);
+ break;
+ case CUSTOM:
+ declarer.customGrouping(stream.getFrom(), streamId,
+ buildCustomStreamGrouping(stream.getGrouping().getCustomClass(),
+ executionContext,
+ objectBuilder));
+ break;
+ default:
+ throw new UnsupportedOperationException("unsupported grouping type: " + grouping);
+ }
+ }
+ executionContext.setStreams(componentStreams);
+ }
+
+ private CustomStreamGrouping buildCustomStreamGrouping(ObjectDefinition objectDefinition,
+ EcoExecutionContext executionContext,
+ ObjectBuilder objectBuilder)
+ throws ClassNotFoundException,
+ IllegalAccessException, InstantiationException, NoSuchFieldException,
+ InvocationTargetException {
+ Object grouping = objectBuilder.buildObject(objectDefinition, executionContext);
+ return (CustomStreamGrouping) grouping;
+ }
+}
diff --git a/eco/src/java/com/twitter/heron/eco/builder/EcoBuilder.java b/eco/src/java/com/twitter/heron/eco/builder/storm/EcoBuilder.java
similarity index 91%
rename from eco/src/java/com/twitter/heron/eco/builder/EcoBuilder.java
rename to eco/src/java/com/twitter/heron/eco/builder/storm/EcoBuilder.java
index 06f7a477f0..4900e73fa4 100644
--- a/eco/src/java/com/twitter/heron/eco/builder/EcoBuilder.java
+++ b/eco/src/java/com/twitter/heron/eco/builder/storm/EcoBuilder.java
@@ -11,7 +11,7 @@
// 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.
-package com.twitter.heron.eco.builder;
+package com.twitter.heron.eco.builder.storm;
import java.lang.reflect.InvocationTargetException;
@@ -20,6 +20,11 @@
import org.apache.storm.topology.TopologyBuilder;
import com.twitter.heron.api.Config;
+import com.twitter.heron.eco.builder.BoltBuilder;
+import com.twitter.heron.eco.builder.ComponentBuilder;
+import com.twitter.heron.eco.builder.ConfigBuilder;
+import com.twitter.heron.eco.builder.ObjectBuilder;
+
import com.twitter.heron.eco.definition.EcoExecutionContext;
import com.twitter.heron.eco.definition.EcoTopologyDefinition;
diff --git a/eco/src/java/com/twitter/heron/eco/builder/SpoutBuilder.java b/eco/src/java/com/twitter/heron/eco/builder/storm/SpoutBuilder.java
similarity index 94%
rename from eco/src/java/com/twitter/heron/eco/builder/SpoutBuilder.java
rename to eco/src/java/com/twitter/heron/eco/builder/storm/SpoutBuilder.java
index 19dbfc95a1..775d30d9e7 100644
--- a/eco/src/java/com/twitter/heron/eco/builder/SpoutBuilder.java
+++ b/eco/src/java/com/twitter/heron/eco/builder/storm/SpoutBuilder.java
@@ -11,13 +11,15 @@
// 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.
-package com.twitter.heron.eco.builder;
+package com.twitter.heron.eco.builder.storm;
import java.lang.reflect.InvocationTargetException;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.TopologyBuilder;
+import com.twitter.heron.eco.builder.ObjectBuilder;
+
import com.twitter.heron.eco.definition.EcoExecutionContext;
import com.twitter.heron.eco.definition.EcoTopologyDefinition;
import com.twitter.heron.eco.definition.ObjectDefinition;
diff --git a/eco/src/java/com/twitter/heron/eco/builder/StreamBuilder.java b/eco/src/java/com/twitter/heron/eco/builder/storm/StreamBuilder.java
similarity index 98%
rename from eco/src/java/com/twitter/heron/eco/builder/StreamBuilder.java
rename to eco/src/java/com/twitter/heron/eco/builder/storm/StreamBuilder.java
index 3ff50b3b2d..6777890aaa 100644
--- a/eco/src/java/com/twitter/heron/eco/builder/StreamBuilder.java
+++ b/eco/src/java/com/twitter/heron/eco/builder/storm/StreamBuilder.java
@@ -11,7 +11,7 @@
// 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.
-package com.twitter.heron.eco.builder;
+package com.twitter.heron.eco.builder.storm;
import java.lang.reflect.InvocationTargetException;
import java.util.HashMap;
@@ -27,6 +27,8 @@
import org.apache.storm.tuple.Fields;
import org.apache.storm.utils.Utils;
+import com.twitter.heron.eco.builder.ObjectBuilder;
+
import com.twitter.heron.eco.definition.ComponentStream;
import com.twitter.heron.eco.definition.EcoExecutionContext;
import com.twitter.heron.eco.definition.EcoTopologyDefinition;
diff --git a/eco/src/java/com/twitter/heron/eco/definition/EcoTopologyDefinition.java b/eco/src/java/com/twitter/heron/eco/definition/EcoTopologyDefinition.java
index b5b8ad9f25..132bb0d150 100644
--- a/eco/src/java/com/twitter/heron/eco/definition/EcoTopologyDefinition.java
+++ b/eco/src/java/com/twitter/heron/eco/definition/EcoTopologyDefinition.java
@@ -22,6 +22,7 @@
public class EcoTopologyDefinition {
private String name;
+ private String type;
private Map<String, Object> config = new HashMap<>();
private Map<String, SpoutDefinition> spouts = new LinkedHashMap<>();
private Map<String, BoltDefinition> bolts = new LinkedHashMap<>();
@@ -94,10 +95,25 @@ public void setConfig(Map<String, Object> config) {
}
public String getName() {
-
return name;
}
+ public String getType() {
+ if (type == null || "storm".equals(type)) {
+ return "storm";
+ }
+
+ if ("heron".equals(type)) {
+ return "heron";
+ }
+
+ return null;
+ }
+
+ public void setType(String type) {
+ this.type = type;
+ }
+
public void setName(String name) {
this.name = name;
}
diff --git a/eco/src/java/com/twitter/heron/eco/submit/EcoSubmitter.java b/eco/src/java/com/twitter/heron/eco/submit/EcoSubmitter.java
index 07d65e52bf..8383bc4e0a 100644
--- a/eco/src/java/com/twitter/heron/eco/submit/EcoSubmitter.java
+++ b/eco/src/java/com/twitter/heron/eco/submit/EcoSubmitter.java
@@ -13,17 +13,27 @@
// limitations under the License.
package com.twitter.heron.eco.submit;
+
import org.apache.storm.StormSubmitter;
-import org.apache.storm.generated.AlreadyAliveException;
-import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.generated.StormTopology;
import com.twitter.heron.api.Config;
+import com.twitter.heron.api.HeronSubmitter;
+import com.twitter.heron.api.HeronTopology;
public class EcoSubmitter {
- public void submitTopology(String topologyName, Config topologyConfig, StormTopology topology)
- throws AlreadyAliveException, InvalidTopologyException {
+ public void submitStormTopology(String topologyName,
+ Config topologyConfig, StormTopology topology)
+ throws org.apache.storm.generated.AlreadyAliveException,
+ org.apache.storm.generated.InvalidTopologyException {
StormSubmitter.submitTopology(topologyName, topologyConfig, topology);
}
+
+ public void submitHeronTopology(String topologyName,
+ Config topologyConfig, HeronTopology topology)
+ throws com.twitter.heron.api.exception.AlreadyAliveException,
+ com.twitter.heron.api.exception.InvalidTopologyException {
+ HeronSubmitter.submitTopology(topologyName, topologyConfig, topology);
+ }
}
diff --git a/eco/tests/java/BUILD b/eco/tests/java/BUILD
index efcc0bcb5c..388df82ae6 100644
--- a/eco/tests/java/BUILD
+++ b/eco/tests/java/BUILD
@@ -7,6 +7,12 @@ test_deps_files = [
heron_local_deps = [
"//eco/src/java:eco-java",
+ "//eco/src/java:eco-submit-java",
+ "//eco/src/java:eco-builder-java",
+ "//eco/src/java:eco-defs-java",
+ "//eco/src/java:eco-parser-java",
+ "//eco/src/java:eco-storm-builder-java",
+ "//eco/src/java:eco-heron-builder-java",
"//heron/api/src/java:api-java-low-level",
"//storm-compatibility/src/java:storm-compatibility-java",
]
@@ -14,8 +20,15 @@ heron_local_deps = [
eco_test_deps = heron_local_deps + test_deps_files
java_test(
- name = "EcoBuilderTest",
- srcs = glob(["com/twitter/heron/eco/builder/EcoBuilderTest.java"]),
+ name = "StormEcoBuilderTest",
+ srcs = glob(["com/twitter/heron/eco/builder/storm/StormEcoBuilderTest.java"]),
+ deps = eco_test_deps,
+ size = "small",
+)
+
+java_test(
+ name = "HeronEcoBuilderTest",
+ srcs = glob(["com/twitter/heron/eco/builder/heron/HeronEcoBuilderTest.java"]),
deps = eco_test_deps,
size = "small",
)
@@ -49,15 +62,29 @@ java_test(
)
java_test(
- name = "SpoutBuilderTest",
- srcs = glob(["com/twitter/heron/eco/builder/SpoutBuilderTest.java"]),
+ name = "StormSpoutBuilderTest",
+ srcs = glob(["com/twitter/heron/eco/builder/storm/StormSpoutBuilderTest.java"]),
+ deps = eco_test_deps,
+ size = "small"
+)
+
+java_test(
+ name = "HeronSpoutBuilderTest",
+ srcs = glob(["com/twitter/heron/eco/builder/heron/HeronSpoutBuilderTest.java"]),
+ deps = eco_test_deps,
+ size = "small"
+)
+
+java_test(
+ name = "StormStreamBuilderTest",
+ srcs = glob(["com/twitter/heron/eco/builder/storm/StormStreamBuilderTest.java"]),
deps = eco_test_deps,
size = "small"
)
java_test(
- name = "StreamBuilderTest",
- srcs = glob(["com/twitter/heron/eco/builder/StreamBuilderTest.java"]),
+ name = "HeronStreamBuilderTest",
+ srcs = glob(["com/twitter/heron/eco/builder/heron/HeronStreamBuilderTest.java"]),
deps = eco_test_deps,
size = "small"
)
diff --git a/eco/tests/java/com/twitter/heron/eco/EcoTest.java b/eco/tests/java/com/twitter/heron/eco/EcoTest.java
index cd082817ce..35c900677a 100644
--- a/eco/tests/java/com/twitter/heron/eco/EcoTest.java
+++ b/eco/tests/java/com/twitter/heron/eco/EcoTest.java
@@ -16,7 +16,6 @@
import java.io.FileInputStream;
import org.apache.storm.generated.StormTopology;
-import org.apache.storm.topology.TopologyBuilder;
import org.junit.After;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -27,9 +26,7 @@
import org.powermock.api.mockito.PowerMockito;
import com.twitter.heron.api.Config;
-import com.twitter.heron.eco.builder.EcoBuilder;
-import com.twitter.heron.eco.builder.ObjectBuilder;
-import com.twitter.heron.eco.definition.EcoExecutionContext;
+import com.twitter.heron.api.HeronTopology;
import com.twitter.heron.eco.definition.EcoTopologyDefinition;
import com.twitter.heron.eco.parser.EcoParser;
import com.twitter.heron.eco.submit.EcoSubmitter;
@@ -43,50 +40,59 @@
@RunWith(MockitoJUnitRunner.class)
public class EcoTest {
- @Mock
- private EcoBuilder mockEcoBuilder;
@Mock
private EcoParser mockEcoParser;
@Mock
- private TopologyBuilder mockTopologyBuilder;
- @Mock
private EcoSubmitter mockEcoSubmitter;
@InjectMocks
private Eco subject;
@After
public void ensureNoUnexpectedMockInteractions() {
- Mockito.verifyNoMoreInteractions(mockEcoBuilder,
- mockEcoParser,
- mockTopologyBuilder,
+ Mockito.verifyNoMoreInteractions(mockEcoParser,
mockEcoSubmitter);
}
@Test
- public void testSubmit_AllGood_BehavesAsExpected() throws Exception {
+ public void testSubmit_StormTopologyType_BehavesAsExpected() throws Exception {
FileInputStream mockStream = PowerMockito.mock(FileInputStream.class);
FileInputStream mockPropsStream = PowerMockito.mock(FileInputStream.class);
final String topologyName = "the name";
EcoTopologyDefinition topologyDefinition = new EcoTopologyDefinition();
topologyDefinition.setName(topologyName);
- Config config = new Config();
when(mockEcoParser.parseFromInputStream(eq(mockStream), eq(mockPropsStream), eq(false)))
.thenReturn(topologyDefinition);
- when(mockEcoBuilder.buildConfig(eq(topologyDefinition))).thenReturn(config);
- when(mockEcoBuilder.buildTopologyBuilder(any(EcoExecutionContext.class),
- any(ObjectBuilder.class))).thenReturn(mockTopologyBuilder);
subject.submit(mockStream, mockPropsStream, false);
verify(mockEcoParser).parseFromInputStream(same(mockStream),
same(mockPropsStream), eq(false));
- verify(mockEcoBuilder).buildConfig(same(topologyDefinition));
- verify(mockEcoBuilder).buildTopologyBuilder(any(EcoExecutionContext.class),
- any(ObjectBuilder.class));
- verify(mockTopologyBuilder).createTopology();
- verify(mockEcoSubmitter).submitTopology(any(String.class), any(Config.class),
+
+ verify(mockEcoSubmitter).submitStormTopology(any(String.class), any(Config.class),
any(StormTopology.class));
}
+
+ @Test
+ public void testSubmit_HeronTopologyType_BehavesAsExpected() throws Exception {
+ FileInputStream mockStream = PowerMockito.mock(FileInputStream.class);
+ FileInputStream mockPropsStream = PowerMockito.mock(FileInputStream.class);
+
+ final String topologyName = "the name";
+ EcoTopologyDefinition topologyDefinition = new EcoTopologyDefinition();
+ topologyDefinition.setName(topologyName);
+ topologyDefinition.setType("heron");
+
+ when(mockEcoParser.parseFromInputStream(eq(mockStream), eq(mockPropsStream), eq(false)))
+ .thenReturn(topologyDefinition);
+
+ subject.submit(mockStream, mockPropsStream, false);
+
+ verify(mockEcoParser).parseFromInputStream(same(mockStream),
+ same(mockPropsStream), eq(false));
+
+ verify(mockEcoSubmitter).submitHeronTopology(any(String.class), any(Config.class),
+ any(HeronTopology.class));
+ }
}
diff --git a/eco/tests/java/com/twitter/heron/eco/builder/heron/HeronEcoBuilderTest.java b/eco/tests/java/com/twitter/heron/eco/builder/heron/HeronEcoBuilderTest.java
new file mode 100644
index 0000000000..b360a3532e
--- /dev/null
+++ b/eco/tests/java/com/twitter/heron/eco/builder/heron/HeronEcoBuilderTest.java
@@ -0,0 +1,133 @@
+// Copyright 2017 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.eco.builder.heron;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import com.twitter.heron.api.Config;
+import com.twitter.heron.api.topology.TopologyBuilder;
+
+import com.twitter.heron.eco.builder.BoltBuilder;
+import com.twitter.heron.eco.builder.ComponentBuilder;
+import com.twitter.heron.eco.builder.ConfigBuilder;
+import com.twitter.heron.eco.builder.ObjectBuilder;
+
+import com.twitter.heron.eco.definition.EcoExecutionContext;
+import com.twitter.heron.eco.definition.EcoTopologyDefinition;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.CoreMatchers.sameInstance;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.same;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class HeronEcoBuilderTest {
+
+ @Mock
+ private SpoutBuilder mockSpoutBuilder;
+ @Mock
+ private BoltBuilder mockBoltBuilder;
+ @Mock
+ private StreamBuilder mockStreamBuilder;
+ @Mock
+ private ComponentBuilder mockComponentBuilder;
+ @Mock
+ private ConfigBuilder mockConfigBuilder;
+ @InjectMocks
+ private EcoBuilder subject;
+
+ private Map<String, Object> configMap;
+
+ private EcoTopologyDefinition ecoTopologyDefinition;
+
+ @Before
+ public void setUpForEachTestCase() {
+ configMap = new HashMap<>();
+ ecoTopologyDefinition = new EcoTopologyDefinition();
+ ecoTopologyDefinition.setConfig(configMap);
+ }
+
+ @After
+ public void ensureNoUnexpectedMockInteractions() {
+ verifyNoMoreInteractions(mockSpoutBuilder,
+ mockBoltBuilder,
+ mockStreamBuilder,
+ mockComponentBuilder,
+ mockConfigBuilder);
+ }
+
+ @Test
+ public void testBuild_EmptyConfigMap_ReturnsDefaultConfigs() throws Exception {
+
+ Config config = new Config();
+ when(mockConfigBuilder.buildConfig(eq(ecoTopologyDefinition))).thenReturn(config);
+
+ Config returnedConfig = subject.buildConfig(ecoTopologyDefinition);
+
+ verify(mockConfigBuilder).buildConfig(same(ecoTopologyDefinition));
+
+ assertThat(returnedConfig.get(Config.TOPOLOGY_DEBUG), is(nullValue()));
+ assertThat(config, sameInstance(returnedConfig));
+ }
+
+ @Test
+ public void testBuild_CustomConfigMap_ReturnsCorrectConfigs() throws Exception {
+ configMap.put(Config.TOPOLOGY_DEBUG, false);
+ final String environment = "dev";
+ final int spouts = 3;
+ configMap.put(Config.TOPOLOGY_ENVIRONMENT, environment);
+ configMap.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, spouts);
+
+ Config config = new Config();
+
+ when(mockConfigBuilder.buildConfig(eq(ecoTopologyDefinition))).thenReturn(config);
+
+ assertThat(subject.buildConfig(ecoTopologyDefinition), sameInstance(config));
+
+ verify(mockConfigBuilder).buildConfig(same(ecoTopologyDefinition));
+ }
+
+ @Test
+ public void testBuildTopologyBuilder_BuildsAsExpected()
+ throws IllegalAccessException, ClassNotFoundException,
+ InstantiationException, NoSuchFieldException, InvocationTargetException {
+ Config config = new Config();
+ EcoExecutionContext context = new EcoExecutionContext(ecoTopologyDefinition, config);
+ ObjectBuilder objectBuilder = new ObjectBuilder();
+ subject.buildTopologyBuilder(context, objectBuilder);
+
+ verify(mockSpoutBuilder).buildSpouts(same(context),
+ any(TopologyBuilder.class), same(objectBuilder));
+ verify(mockBoltBuilder).buildBolts(same(context), same(objectBuilder));
+ verify(mockStreamBuilder).buildStreams(same(context), any(TopologyBuilder.class),
+ same(objectBuilder));
+ verify(mockComponentBuilder).buildComponents(same(context), same(objectBuilder));
+ }
+}
diff --git a/eco/tests/java/com/twitter/heron/eco/builder/heron/HeronSpoutBuilderTest.java b/eco/tests/java/com/twitter/heron/eco/builder/heron/HeronSpoutBuilderTest.java
new file mode 100644
index 0000000000..bc57eaea8f
--- /dev/null
+++ b/eco/tests/java/com/twitter/heron/eco/builder/heron/HeronSpoutBuilderTest.java
@@ -0,0 +1,161 @@
+// Copyright 2018 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.eco.builder.heron;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import com.twitter.heron.api.spout.IRichSpout;
+import com.twitter.heron.api.spout.SpoutOutputCollector;
+import com.twitter.heron.api.topology.OutputFieldsDeclarer;
+import com.twitter.heron.api.topology.TopologyBuilder;
+import com.twitter.heron.api.topology.TopologyContext;
+
+import com.twitter.heron.eco.builder.ObjectBuilder;
+import com.twitter.heron.eco.definition.EcoExecutionContext;
+import com.twitter.heron.eco.definition.EcoTopologyDefinition;
+import com.twitter.heron.eco.definition.SpoutDefinition;
+
+import static org.mockito.Matchers.anyObject;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.same;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class HeronSpoutBuilderTest {
+
+ @Mock
+ private EcoExecutionContext mockContext;
+ @Mock
+ private TopologyBuilder mockTopologyBuilder;
+ @Mock
+ private ObjectBuilder mockObjectBuilder;
+
+ private SpoutBuilder subject;
+
+ @Before
+ public void setUpForEachTestCase() {
+ subject = new SpoutBuilder();
+ }
+
+ @After
+ public void ensureNoUnexpectedMockInteractions() {
+ Mockito.verifyNoMoreInteractions(mockContext,
+ mockTopologyBuilder,
+ mockObjectBuilder);
+ }
+
+ @Test
+ public void testBuildSpouts_AllGood_BehavesAsExpected() throws ClassNotFoundException,
+ InvocationTargetException, NoSuchFieldException,
+ InstantiationException, IllegalAccessException {
+ EcoTopologyDefinition topologyDefinition = new EcoTopologyDefinition();
+
+ SpoutDefinition spoutDefinition = new SpoutDefinition();
+ final String id = "id";
+ final int parallelism = 2;
+ spoutDefinition.setId(id);
+ spoutDefinition.setParallelism(parallelism);
+ SpoutDefinition spoutDefinition1 = new SpoutDefinition();
+ final String id1 = "id1";
+ final int parallelism1 = 3;
+ spoutDefinition1.setId(id1);
+ spoutDefinition1.setParallelism(parallelism1);
+ List<SpoutDefinition> spoutDefinitions = new ArrayList<>();
+ spoutDefinitions.add(spoutDefinition);
+ spoutDefinitions.add(spoutDefinition1);
+ topologyDefinition.setSpouts(spoutDefinitions);
+ MockSpout mockSpout = new MockSpout();
+ MockSpout mockSpout1 = new MockSpout();
+
+ when(mockObjectBuilder.buildObject(eq(spoutDefinition),
+ eq(mockContext))).thenReturn(mockSpout);
+ when(mockObjectBuilder.buildObject(eq(spoutDefinition1),
+ eq(mockContext))).thenReturn(mockSpout1);
+ when(mockContext.getTopologyDefinition()).thenReturn(topologyDefinition);
+
+ subject.buildSpouts(mockContext, mockTopologyBuilder, mockObjectBuilder);
+
+ verify(mockContext).getTopologyDefinition();
+ verify(mockObjectBuilder).buildObject(same(spoutDefinition), same(mockContext));
+ verify(mockObjectBuilder).buildObject(same(spoutDefinition1), same(mockContext));
+ verify(mockTopologyBuilder).setSpout(eq(id), eq(mockSpout), eq(parallelism));
+ verify(mockTopologyBuilder).setSpout(eq(id1), eq(mockSpout1), eq(parallelism1));
+ verify(mockContext).addSpout(eq(id), anyObject());
+ verify(mockContext).addSpout(eq(id1), anyObject());
+ }
+
+ @SuppressWarnings("serial")
+ private class MockSpout implements IRichSpout {
+
+ @Override
+ public void open(Map<String, Object> conf,
+ TopologyContext context, SpoutOutputCollector collector) {
+
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public void activate() {
+
+ }
+
+ @Override
+ public void deactivate() {
+
+ }
+
+ @Override
+ public void nextTuple() {
+
+ }
+
+ @Override
+ public void ack(Object msgId) {
+
+ }
+
+ @Override
+ public void fail(Object msgId) {
+
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+
+ }
+
+ @Override
+ public Map<String, Object> getComponentConfiguration() {
+ return null;
+ }
+ }
+
+
+}
diff --git a/eco/tests/java/com/twitter/heron/eco/builder/heron/HeronStreamBuilderTest.java b/eco/tests/java/com/twitter/heron/eco/builder/heron/HeronStreamBuilderTest.java
new file mode 100644
index 0000000000..1ac66918b3
--- /dev/null
+++ b/eco/tests/java/com/twitter/heron/eco/builder/heron/HeronStreamBuilderTest.java
@@ -0,0 +1,370 @@
+// Copyright 2018 Twitter. All rights reserved.
+//
+// Licensed 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.
+package com.twitter.heron.eco.builder.heron;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import com.twitter.heron.api.bolt.BasicOutputCollector;
+import com.twitter.heron.api.bolt.IBasicBolt;
+import com.twitter.heron.api.bolt.IRichBolt;
+import com.twitter.heron.api.bolt.IWindowedBolt;
+import com.twitter.heron.api.bolt.OutputCollector;
+import com.twitter.heron.api.grouping.CustomStreamGrouping;
+import com.twitter.heron.api.topology.BoltDeclarer;
+import com.twitter.heron.api.topology.OutputFieldsDeclarer;
+import com.twitter.heron.api.topology.TopologyBuilder;
+import com.twitter.heron.api.topology.TopologyContext;
+import com.twitter.heron.api.tuple.Fields;
+import com.twitter.heron.api.tuple.Tuple;
+import com.twitter.heron.api.windowing.TimestampExtractor;
+import com.twitter.heron.api.windowing.TupleWindow;
+
+import com.twitter.heron.eco.builder.ObjectBuilder;
+import com.twitter.heron.eco.definition.EcoExecutionContext;
+import com.twitter.heron.eco.definition.EcoTopologyDefinition;
+import com.twitter.heron.eco.definition.GroupingDefinition;
+import com.twitter.heron.eco.definition.ObjectDefinition;
+import com.twitter.heron.eco.definition.StreamDefinition;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyMap;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.same;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class HeronStreamBuilderTest {
+
+ @Mock
+ private EcoTopologyDefinition mockDefinition;
+ @Mock
+ private EcoExecutionContext mockContext;
+ @Mock
+ private TopologyBuilder mockTopologyBuilder;
+ @Mock
+ private ObjectBuilder mockObjectBuilder;
+ @Mock
+ private BoltDeclarer mockBoltDeclarer;
+
+ private StreamBuilder subject;
+
+ @Before
+ public void setUpForEachTestCase() {
+ subject = new StreamBuilder();
+ }
+
+ @After
+ public void ensureNoUnexpectedMockInteractions() {
+ Mockito.verifyNoMoreInteractions(mockDefinition,
+ mockContext,
+ mockTopologyBuilder,
+ mockObjectBuilder,
+ mockBoltDeclarer);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void buildStreams_SpoutToIRichBolt_ShuffleGrouping() throws ClassNotFoundException,
+ InvocationTargetException, NoSuchFieldException,
+ InstantiationException, IllegalAccessException {
+ final int iRichBoltParallelism = 1;
+ final String to = "to";
+ final String from = "from";
+ final String streamId = "id";
+ StreamDefinition streamDefinition = new StreamDefinition();
+ streamDefinition.setFrom(from);
+ streamDefinition.setTo(to);
+ streamDefinition.setId(streamId);
+ List<StreamDefinition> streams = new ArrayList<>();
+ streams.add(streamDefinition);
+ GroupingDefinition groupingDefinition = new GroupingDefinition();
+ groupingDefinition.setType(GroupingDefinition.Type.SHUFFLE);
+ groupingDefinition.setStreamId(streamId);
+ streamDefinition.setGrouping(groupingDefinition);
+ MockIRichBolt mockIRichBolt = new MockIRichBolt();
+
+ when(mockContext.getTopologyDefinition()).thenReturn(mockDefinition);
+ when(mockContext.getBolt(eq(to))).thenReturn(mockIRichBolt);
+ when(mockDefinition.getStreams()).thenReturn(streams);
+ when(mockDefinition.parallelismForBolt(eq(to))).thenReturn(iRichBoltParallelism);
+ when(mockTopologyBuilder.setBolt(eq(to),
+ eq(mockIRichBolt), eq(iRichBoltParallelism))).thenReturn(mockBoltDeclarer);
+
+ subject.buildStreams(mockContext, mockTopologyBuilder, mockObjectBuilder);
+
+ verify(mockContext).getTopologyDefinition();
+ verify(mockContext).getBolt(eq(to));
+ verify(mockDefinition).parallelismForBolt(eq(to));
+ verify(mockTopologyBuilder).setBolt(eq(to), eq(mockIRichBolt), eq(iRichBoltParallelism));
+ verify(mockBoltDeclarer).shuffleGrouping(eq(from), eq(streamId));
+ verify(mockContext).setStreams(anyMap());
+ verify(mockDefinition).getStreams();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void buildStreams_SpoutToIBasicBolt_FieldsGroupingWithArgs() throws
+ ClassNotFoundException,
+ InvocationTargetException, NoSuchFieldException,
+ InstantiationException, IllegalAccessException {
+ final int iRichBoltParallelism = 1;
+ final String to = "to";
+ final String from = "from";
+ final String streamId = "id";
+ StreamDefinition streamDefinition = new StreamDefinition();
+ streamDefinition.setFrom(from);
+ streamDefinition.setTo(to);
+ streamDefinition.setId(streamId);
+ List<StreamDefinition> streams = new ArrayList<>();
+ streams.add(streamDefinition);
+ GroupingDefinition groupingDefinition = new GroupingDefinition();
+ groupingDefinition.setType(GroupingDefinition.Type.FIELDS);
+ List<String> args = new ArrayList<>();
+ args.add("arg1");
+ groupingDefinition.setArgs(args);
+ groupingDefinition.setStreamId(streamId);
+ streamDefinition.setGrouping(groupingDefinition);
+ MockIBasicBolt mockIBasicBolt = new MockIBasicBolt();
+
+ when(mockContext.getTopologyDefinition()).thenReturn(mockDefinition);
+ when(mockContext.getBolt(eq(to))).thenReturn(mockIBasicBolt);
+ when(mockDefinition.getStreams()).thenReturn(streams);
+ when(mockDefinition.parallelismForBolt(eq(to))).thenReturn(iRichBoltParallelism);
+ when(mockTopologyBuilder.setBolt(eq(to),
+ eq(mockIBasicBolt), eq(iRichBoltParallelism))).thenReturn(mockBoltDeclarer);
+
+ subject.buildStreams(mockContext, mockTopologyBuilder, mockObjectBuilder);
+
+ verify(mockContext).getTopologyDefinition();
+ verify(mockContext).getBolt(eq(to));
+ verify(mockDefinition).parallelismForBolt(eq(to));
+ verify(mockTopologyBuilder).setBolt(eq(to), eq(mockIBasicBolt), eq(iRichBoltParallelism));
+ verify(mockBoltDeclarer).fieldsGrouping(eq(from), eq(streamId), any(Fields.class));
+ verify(mockContext).setStreams(anyMap());
+ verify(mockDefinition).getStreams();
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ @SuppressWarnings("unchecked")
+ public void buildStreams_SpoutToIBasicBolt_FieldsGroupingWithoutArgs_ExceptionThrown() throws
+ ClassNotFoundException,
+ InvocationTargetException, NoSuchFieldException,
+ InstantiationException, IllegalAccessException {
+ final int iRichBoltParallelism = 1;
+ final String to = "to";
+ final String from = "from";
+ final String streamId = "id";
+ StreamDefinition streamDefinition = new StreamDefinition();
+ streamDefinition.setFrom(from);
+ streamDefinition.setTo(to);
+ streamDefinition.setId(streamId);
+ List<StreamDefinition> streams = new ArrayList<>();
+ streams.add(streamDefinition);
+ GroupingDefinition groupingDefinition = new GroupingDefinition();
+ groupingDefinition.setType(GroupingDefinition.Type.FIELDS);
+
+ groupingDefinition.setStreamId(streamId);
+ streamDefinition.setGrouping(groupingDefinition);
+ MockIBasicBolt mockIBasicBolt = new MockIBasicBolt();
+ try {
+ when(mockContext.getTopologyDefinition()).thenReturn(mockDefinition);
+ when(mockContext.getBolt(eq(to))).thenReturn(mockIBasicBolt);
+ when(mockDefinition.getStreams()).thenReturn(streams);
+ when(mockDefinition.parallelismForBolt(eq(to))).thenReturn(iRichBoltParallelism);
+ when(mockTopologyBuilder.setBolt(eq(to),
+ eq(mockIBasicBolt), eq(iRichBoltParallelism))).thenReturn(mockBoltDeclarer);
+
+ subject.buildStreams(mockContext, mockTopologyBuilder, mockObjectBuilder);
+ } finally {
+
+ verify(mockContext).getTopologyDefinition();
+ verify(mockContext).getBolt(eq(to));
+ verify(mockDefinition).parallelismForBolt(eq(to));
+ verify(mockTopologyBuilder).setBolt(eq(to), eq(mockIBasicBolt), eq(iRichBoltParallelism));
+ verify(mockDefinition).getStreams();
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void buildStreams_SpoutToIWindowedBolt_CustomGrouping() throws ClassNotFoundException,
+ InvocationTargetException, NoSuchFieldException,
+ InstantiationException, IllegalAccessException {
+ final int iRichBoltParallelism = 1;
+ final String to = "to";
+ final String from = "from";
+ final String streamId = "id";
+ StreamDefinition streamDefinition = new StreamDefinition();
+ streamDefinition.setFrom(from);
+ streamDefinition.setTo(to);
+ streamDefinition.setId(streamId);
+ List<StreamDefinition> streams = new ArrayList<>();
+ streams.add(streamDefinition);
+ GroupingDefinition groupingDefinition = new GroupingDefinition();
+ groupingDefinition.setType(GroupingDefinition.Type.CUSTOM);
+ MockCustomObjectDefinition mockCustomObjectDefinition = new MockCustomObjectDefinition();
+
+ groupingDefinition.setCustomClass(mockCustomObjectDefinition);
+ List<String> args = new ArrayList<>();
+ args.add("arg1");
+ groupingDefinition.setArgs(args);
+ groupingDefinition.setStreamId(streamId);
+ streamDefinition.setGrouping(groupingDefinition);
+ MockIWindowedBolt mockIWindowedBolt = new MockIWindowedBolt();
+ MockCustomStreamGrouping mockCustomStreamGrouping = new MockCustomStreamGrouping();
+
+ when(mockContext.getTopologyDefinition()).thenReturn(mockDefinition);
+ when(mockContext.getBolt(eq(to))).thenReturn(mockIWindowedBolt);
+ when(mockDefinition.getStreams()).thenReturn(streams);
+ when(mockDefinition.parallelismForBolt(eq(to))).thenReturn(iRichBoltParallelism);
+ when(mockTopologyBuilder.setBolt(eq(to),
+ eq(mockIWindowedBolt), eq(iRichBoltParallelism))).thenReturn(mockBoltDeclarer);
+ when(mockObjectBuilder.buildObject(eq(mockCustomObjectDefinition),
+ eq(mockContext))).thenReturn(mockCustomStreamGrouping);
+
+ subject.buildStreams(mockContext, mockTopologyBuilder, mockObjectBuilder);
+
+ verify(mockContext).getTopologyDefinition();
+ verify(mockContext).getBolt(eq(to));
+ verify(mockDefinition).parallelismForBolt(eq(to));
+ verify(mockTopologyBuilder).setBolt(eq(to), eq(mockIWindowedBolt), eq(iRichBoltParallelism));
+ verify(mockBoltDeclarer).customGrouping(eq(from), eq(streamId), eq(mockCustomStreamGrouping));
+ verify(mockContext).setStreams(anyMap());
+ verify(mockDefinition).getStreams();
+ verify(mockObjectBuilder).buildObject(same(mockCustomObjectDefinition), same(mockContext));
+ }
+
+ private class MockCustomObjectDefinition extends ObjectDefinition {
+
+ }
+
+ @SuppressWarnings("serial")
+ private class MockCustomStreamGrouping implements CustomStreamGrouping {
+
+ @Override
+ public void prepare(TopologyContext context, String component,
+ String streamId, List<Integer> targetTasks) {
+
+ }
+
+ @Override
+ public List<Integer> chooseTasks(List<Object> values) {
+ return null;
+ }
+ }
+
+ @SuppressWarnings({"rawtypes", "unchecked", "serial"})
+ private class MockIRichBolt implements IRichBolt {
+
+ @Override
+ public void prepare(Map heronConf, TopologyContext context, OutputCollector collector) {
+
+ }
+
+ @Override
+ public void execute(Tuple input) {
+
+ }
+
+ @Override
+ public void cleanup() {
+
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+
+ }
+
+ @Override
+ public Map<String, Object> getComponentConfiguration() {
+ return null;
+ }
+ }
+
+ @SuppressWarnings({"rawtypes", "unchecked", "serial"})
+ private class MockIWindowedBolt implements IWindowedBolt {
+ @Override
+ public void prepare(Map<String, Object> topoConf,
+ TopologyContext context, OutputCollector collector) {
+
+ }
+
+ @Override
+ public void execute(TupleWindow inputWindow) {
+
+ }
+
+ @Override
+ public void cleanup() {
+
+ }
+
+ @Override
+ public TimestampExtractor getTimestampExtractor() {
+ return null;
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+
+ }
+
+ @Override
+ public Map<String, Object> getComponentConfiguration() {
+ return null;
+ }
+ }
+
+
+ @SuppressWarnings({"rawtypes", "unchecked", "serial"})
+ public class MockIBasicBolt implements IBasicBolt {
+ @Override
+ public void prepare(Map heronConf, TopologyContext context) {
+
+ }
+
+ @Override
+ public void execute(Tuple input, BasicOutputCollector collector) {
+
+ }
+
+ @Override
+ public void cleanup() {
+
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+
+ }
+
+ @Override
+ public Map<String, Object> getComponentConfiguration() {
+ return null;
+ }
+ }
+}
diff --git a/eco/tests/java/com/twitter/heron/eco/builder/EcoBuilderTest.java b/eco/tests/java/com/twitter/heron/eco/builder/storm/StormEcoBuilderTest.java
similarity index 93%
rename from eco/tests/java/com/twitter/heron/eco/builder/EcoBuilderTest.java
rename to eco/tests/java/com/twitter/heron/eco/builder/storm/StormEcoBuilderTest.java
index d43ecbcbea..0fffb6db03 100644
--- a/eco/tests/java/com/twitter/heron/eco/builder/EcoBuilderTest.java
+++ b/eco/tests/java/com/twitter/heron/eco/builder/storm/StormEcoBuilderTest.java
@@ -11,7 +11,7 @@
// 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.
-package com.twitter.heron.eco.builder;
+package com.twitter.heron.eco.builder.storm;
import java.lang.reflect.InvocationTargetException;
import java.util.HashMap;
@@ -27,6 +27,11 @@
import org.mockito.runners.MockitoJUnitRunner;
import com.twitter.heron.api.Config;
+import com.twitter.heron.eco.builder.BoltBuilder;
+import com.twitter.heron.eco.builder.ComponentBuilder;
+import com.twitter.heron.eco.builder.ConfigBuilder;
+import com.twitter.heron.eco.builder.ObjectBuilder;
+
import com.twitter.heron.eco.definition.EcoExecutionContext;
import com.twitter.heron.eco.definition.EcoTopologyDefinition;
@@ -42,7 +47,7 @@
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
-public class EcoBuilderTest {
+public class StormEcoBuilderTest {
@Mock
private SpoutBuilder mockSpoutBuilder;
diff --git a/eco/tests/java/com/twitter/heron/eco/builder/SpoutBuilderTest.java b/eco/tests/java/com/twitter/heron/eco/builder/storm/StormSpoutBuilderTest.java
similarity index 97%
rename from eco/tests/java/com/twitter/heron/eco/builder/SpoutBuilderTest.java
rename to eco/tests/java/com/twitter/heron/eco/builder/storm/StormSpoutBuilderTest.java
index 12abe93309..e659e16754 100644
--- a/eco/tests/java/com/twitter/heron/eco/builder/SpoutBuilderTest.java
+++ b/eco/tests/java/com/twitter/heron/eco/builder/storm/StormSpoutBuilderTest.java
@@ -11,7 +11,7 @@
// 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.
-package com.twitter.heron.eco.builder;
+package com.twitter.heron.eco.builder.storm;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
@@ -31,6 +31,8 @@
import org.mockito.Mockito;
import org.mockito.runners.MockitoJUnitRunner;
+import com.twitter.heron.eco.builder.ObjectBuilder;
+
import com.twitter.heron.eco.definition.EcoExecutionContext;
import com.twitter.heron.eco.definition.EcoTopologyDefinition;
import com.twitter.heron.eco.definition.SpoutDefinition;
@@ -42,7 +44,7 @@
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
-public class SpoutBuilderTest {
+public class StormSpoutBuilderTest {
@Mock
private EcoExecutionContext mockContext;
diff --git a/eco/tests/java/com/twitter/heron/eco/builder/StreamBuilderTest.java b/eco/tests/java/com/twitter/heron/eco/builder/storm/StormStreamBuilderTest.java
similarity index 98%
rename from eco/tests/java/com/twitter/heron/eco/builder/StreamBuilderTest.java
rename to eco/tests/java/com/twitter/heron/eco/builder/storm/StormStreamBuilderTest.java
index 9c9b868f73..abfe236d12 100644
--- a/eco/tests/java/com/twitter/heron/eco/builder/StreamBuilderTest.java
+++ b/eco/tests/java/com/twitter/heron/eco/builder/storm/StormStreamBuilderTest.java
@@ -11,7 +11,7 @@
// 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.
-package com.twitter.heron.eco.builder;
+package com.twitter.heron.eco.builder.storm;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
@@ -42,6 +42,8 @@
import org.mockito.Mockito;
import org.mockito.runners.MockitoJUnitRunner;
+import com.twitter.heron.eco.builder.ObjectBuilder;
+
import com.twitter.heron.eco.definition.EcoExecutionContext;
import com.twitter.heron.eco.definition.EcoTopologyDefinition;
import com.twitter.heron.eco.definition.GroupingDefinition;
@@ -56,7 +58,7 @@
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
-public class StreamBuilderTest {
+public class StormStreamBuilderTest {
@Mock
private EcoTopologyDefinition mockDefinition;
diff --git a/eco/tests/java/com/twitter/heron/eco/parser/EcoParserTest.java b/eco/tests/java/com/twitter/heron/eco/parser/EcoParserTest.java
index 5c5a756fec..d027c12454 100644
--- a/eco/tests/java/com/twitter/heron/eco/parser/EcoParserTest.java
+++ b/eco/tests/java/com/twitter/heron/eco/parser/EcoParserTest.java
@@ -172,6 +172,7 @@
+ "---\n"
+ "\n"
+ "name: \"kafka-topology\"\n"
+ + "type: \"heron\"\n"
+ "\n"
+ "# Components\n"
+ "# Components are analagous to Spring beans. They are meant to be used as constructor,\n"
@@ -349,6 +350,7 @@ public void testParseFromInputStream_VerifyComponents_MapsAsExpected() throws Ex
List<BeanDefinition> components = topologyDefinition.getComponents();
assertEquals("kafka-topology", topologyDefinition.getName());
+ assertEquals("heron", topologyDefinition.getType());
assertEquals(4, components.size());
BeanDefinition stringSchemeComponent = components.get(0);
diff --git a/eco/tests/java/com/twitter/heron/eco/submit/EcoSubmitterTest.java b/eco/tests/java/com/twitter/heron/eco/submit/EcoSubmitterTest.java
index 7929a944d1..25deb72c24 100644
--- a/eco/tests/java/com/twitter/heron/eco/submit/EcoSubmitterTest.java
+++ b/eco/tests/java/com/twitter/heron/eco/submit/EcoSubmitterTest.java
@@ -15,6 +15,7 @@
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.StormTopology;
+
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -23,13 +24,15 @@
import org.powermock.modules.junit4.PowerMockRunner;
import com.twitter.heron.api.Config;
+import com.twitter.heron.api.HeronSubmitter;
+import com.twitter.heron.api.HeronTopology;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.internal.verification.VerificationModeFactory.times;
@RunWith(PowerMockRunner.class)
-@PrepareForTest(StormSubmitter.class)
+@PrepareForTest({StormSubmitter.class, HeronSubmitter.class})
public class EcoSubmitterTest {
private EcoSubmitter subject;
@@ -40,7 +43,7 @@ public void setUpForEachTestCase() {
}
@Test
- public void submitTopology_AllGood_BehavesAsExpected()
+ public void submitStormTopology_AllGood_BehavesAsExpected()
throws Exception {
Config config = new Config();
StormTopology topology = new StormTopology();
@@ -48,9 +51,24 @@ public void submitTopology_AllGood_BehavesAsExpected()
PowerMockito.doNothing().when(StormSubmitter.class, "submitTopology",
any(String.class), any(Config.class), any(StormTopology.class));
- subject.submitTopology("name", config, topology);
+ subject.submitStormTopology("name", config, topology);
PowerMockito.verifyStatic(times(1));
StormSubmitter.submitTopology(anyString(), any(Config.class), any(StormTopology.class));
}
+
+ @Test
+ public void submitHeronTopology_AllGood_BehavesAsExpected()
+ throws Exception {
+ Config config = new Config();
+ HeronTopology topology = new HeronTopology(null);
+ PowerMockito.spy(HeronSubmitter.class);
+ PowerMockito.doNothing().when(HeronSubmitter.class, "submitTopology",
+ any(String.class), any(Config.class), any(HeronTopology.class));
+
+ subject.submitHeronTopology("name", config, topology);
+ PowerMockito.verifyStatic(times(1));
+ HeronSubmitter.submitTopology(anyString(), any(Config.class), any(HeronTopology.class));
+
+ }
}
diff --git a/scripts/get_all_heron_paths.sh b/scripts/get_all_heron_paths.sh
index 217f07b354..8df9b5207c 100755
--- a/scripts/get_all_heron_paths.sh
+++ b/scripts/get_all_heron_paths.sh
@@ -25,7 +25,7 @@ set +e
# Build everything
DIR=`dirname $0`
source ${DIR}/detect_os_type.sh
-bazel build --config=`platform` {heron,integration_test,tools/java,examples,heronpy,storm-compatibility,storm-compatibility-examples,eco,eco-examples}/...
+bazel build --config=`platform` {heron,integration_test,tools/java,examples,heronpy,storm-compatibility,storm-compatibility-examples,eco,eco-storm-examples,eco-heron-examples}/...
result=$?
if [ "${result}" -eq "0" ] ; then
echo "Bazel build successful!!"
@@ -67,7 +67,7 @@ function get_package_of() {
}
function get_heron_java_paths() {
- local java_paths=$(find {heron,heron/tools,tools,integration_test,storm-compatibility,contrib,eco,eco-examples} -name "*.java" | sed "s|/src/java/.*$|/src/java|"| sed "s|/java/src/.*$|/java/src|" | sed "s|/tests/java/.*$|/tests/java|" | sort -u | fgrep -v "heron/scheduler/" | fgrep -v "heron/scheduler/" )
+ local java_paths=$(find {heron,heron/tools,tools,integration_test,storm-compatibility,contrib,eco,eco-storm-examples,eco-heron-examples} -name "*.java" | sed "s|/src/java/.*$|/src/java|"| sed "s|/java/src/.*$|/java/src|" | sed "s|/tests/java/.*$|/tests/java|" | sort -u | fgrep -v "heron/scheduler/" | fgrep -v "heron/scheduler/" )
if [ "$(uname -s | tr 'A-Z' 'a-z')" != "darwin" ]; then
java_paths=$(echo "${java_paths}" | fgrep -v "/objc_tools/")
fi
diff --git a/scripts/travis/build.sh b/scripts/travis/build.sh
index e4739f88b9..e39d18dcea 100755
--- a/scripts/travis/build.sh
+++ b/scripts/travis/build.sh
@@ -66,7 +66,7 @@ start_timer "$T"
python ${UTILS}/save-logs.py "heron_build.txt" bazel\
--bazelrc=tools/travis/bazel.rc build --config=$PLATFORM heron/... \
heronpy/... examples/... storm-compatibility-examples/... \
- eco-examples/...
+ eco-storm-examples/... eco-heron-examples/...
end_timer "$T"
# run heron unit tests
@@ -77,7 +77,7 @@ python ${UTILS}/save-logs.py "heron_test_non_flaky.txt" bazel\
--test_summary=detailed --test_output=errors\
--config=$PLATFORM --test_tag_filters=-flaky heron/... \
heronpy/... examples/... storm-compatibility-examples/... \
- eco-examples/...
+ eco-storm-examples/... eco-heron-examples/...
end_timer "$T"
# flaky tests are often due to test port race conditions,
@@ -89,7 +89,7 @@ python ${UTILS}/save-logs.py "heron_test_flaky.txt" bazel\
--test_summary=detailed --test_output=errors\
--config=$PLATFORM --test_tag_filters=flaky --jobs=0 heron/... \
heronpy/... examples/... storm-compatibility-examples/... \
- eco-examples/...
+ eco-storm-examples/... eco-heron-examples/...
end_timer "$T"
# build packages
diff --git a/third_party/cereal/BUILD b/third_party/cereal/BUILD
new file mode 100644
index 0000000000..3f7e5ecc3b
--- /dev/null
+++ b/third_party/cereal/BUILD
@@ -0,0 +1,126 @@
+licenses(["notice"])
+
+package(default_visibility = ["//visibility:public"])
+
+package_name = "cereal"
+package_version = "1.2.1"
+
+package_file = package_name + "-" + package_version + ".tar.gz"
+package_dir = package_name + "-" + package_version
+
+file_list = [
+ "include/cereal/access.hpp",
+ "include/cereal/archives/adapters.hpp",
+ "include/cereal/archives/binary.hpp",
+ "include/cereal/archives/json.hpp",
+ "include/cereal/archives/portable_binary.hpp",
+ "include/cereal/archives/xml.hpp",
+ "include/cereal/cereal.hpp",
+ "include/cereal/details/helpers.hpp",
+ "include/cereal/details/polymorphic_impl.hpp",
+ "include/cereal/details/polymorphic_impl_fwd.hpp",
+ "include/cereal/details/static_object.hpp",
+ "include/cereal/details/traits.hpp",
+ "include/cereal/details/util.hpp",
+ "include/cereal/external/base64.hpp",
+ "include/cereal/external/rapidjson/allocators.h",
+ "include/cereal/external/rapidjson/document.h",
+ "include/cereal/external/rapidjson/encodedstream.h",
+ "include/cereal/external/rapidjson/encodings.h",
+ "include/cereal/external/rapidjson/error/en.h",
+ "include/cereal/external/rapidjson/error/error.h",
+ "include/cereal/external/rapidjson/filereadstream.h",
+ "include/cereal/external/rapidjson/filewritestream.h",
+ "include/cereal/external/rapidjson/fwd.h",
+ "include/cereal/external/rapidjson/internal/biginteger.h",
+ "include/cereal/external/rapidjson/internal/diyfp.h",
+ "include/cereal/external/rapidjson/internal/dtoa.h",
+ "include/cereal/external/rapidjson/internal/ieee754.h",
+ "include/cereal/external/rapidjson/internal/itoa.h",
+ "include/cereal/external/rapidjson/internal/meta.h",
+ "include/cereal/external/rapidjson/internal/pow10.h",
+ "include/cereal/external/rapidjson/internal/regex.h",
+ "include/cereal/external/rapidjson/internal/stack.h",
+ "include/cereal/external/rapidjson/internal/strfunc.h",
+ "include/cereal/external/rapidjson/internal/strtod.h",
+ "include/cereal/external/rapidjson/internal/swap.h",
+ "include/cereal/external/rapidjson/istreamwrapper.h",
+ "include/cereal/external/rapidjson/memorybuffer.h",
+ "include/cereal/external/rapidjson/memorystream.h",
+ "include/cereal/external/rapidjson/msinttypes/inttypes.h",
+ "include/cereal/external/rapidjson/msinttypes/stdint.h",
+ "include/cereal/external/rapidjson/ostreamwrapper.h",
+ "include/cereal/external/rapidjson/pointer.h",
+ "include/cereal/external/rapidjson/prettywriter.h",
+ "include/cereal/external/rapidjson/rapidjson.h",
+ "include/cereal/external/rapidjson/reader.h",
+ "include/cereal/external/rapidjson/schema.h",
+ "include/cereal/external/rapidjson/stream.h",
+ "include/cereal/external/rapidjson/stringbuffer.h",
+ "include/cereal/external/rapidjson/writer.h",
+ "include/cereal/external/rapidxml/rapidxml.hpp",
+ "include/cereal/external/rapidxml/rapidxml_iterators.hpp",
+ "include/cereal/external/rapidxml/rapidxml_print.hpp",
+ "include/cereal/external/rapidxml/rapidxml_utils.hpp",
+ "include/cereal/macros.hpp",
+ "include/cereal/types/array.hpp",
+ "include/cereal/types/base_class.hpp",
+ "include/cereal/types/bitset.hpp",
+ "include/cereal/types/boost_variant.hpp",
+ "include/cereal/types/chrono.hpp",
+ "include/cereal/types/common.hpp",
+ "include/cereal/types/complex.hpp",
+ "include/cereal/types/concepts/pair_associative_container.hpp",
+ "include/cereal/types/deque.hpp",
+ "include/cereal/types/forward_list.hpp",
+ "include/cereal/types/functional.hpp",
+ "include/cereal/types/list.hpp",
+ "include/cereal/types/map.hpp",
+ "include/cereal/types/memory.hpp",
+ "include/cereal/types/polymorphic.hpp",
+ "include/cereal/types/queue.hpp",
+ "include/cereal/types/set.hpp",
+ "include/cereal/types/stack.hpp",
+ "include/cereal/types/string.hpp",
+ "include/cereal/types/tuple.hpp",
+ "include/cereal/types/unordered_map.hpp",
+ "include/cereal/types/unordered_set.hpp",
+ "include/cereal/types/utility.hpp",
+ "include/cereal/types/valarray.hpp",
+ "include/cereal/types/vector.hpp",
+]
+
+genrule(
+ name = "cereal-srcs",
+ srcs = [
+ package_file,
+ ],
+ outs = file_list,
+ cmd = "\n".join([
+ "export WORKSPACE_ROOT=$$(pwd)",
+ "export INSTALL_DIR=$$(pwd)/$(@D)",
+ "export TMP_DIR=$$(mktemp -d -t cereal.XXXXX)",
+ "mkdir -p $$TMP_DIR",
+ "cp -R $(SRCS) $$TMP_DIR",
+ "cd $$TMP_DIR",
+ "tar xfz " + package_file,
+ "cd " + package_dir,
+ "$$WORKSPACE_ROOT/$(location //scripts/compile:env_exec) cmake -Wno-dev -DCMAKE_INSTALL_PREFIX:PATH=$$INSTALL_DIR .",
+ "$$WORKSPACE_ROOT/$(location //scripts/compile:env_exec) make install",
+ "rm -rf $$TMP_DIR",
+ ]),
+ tools = [
+ "//scripts/compile:env_exec",
+ ],
+)
+
+cc_library(
+ name = "cereal-cxx",
+ srcs = [
+ "empty.cc",
+ ] + file_list,
+ includes = [
+ "include",
+ ],
+ linkstatic = 1,
+)
diff --git a/third_party/cereal/cereal-1.2.1.tar.gz b/third_party/cereal/cereal-1.2.1.tar.gz
new file mode 100644
index 0000000000..7f469eed37
Binary files /dev/null and b/third_party/cereal/cereal-1.2.1.tar.gz differ
diff --git a/third_party/cereal/empty.cc b/third_party/cereal/empty.cc
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tools/rules/heron_examples.bzl b/tools/rules/heron_examples.bzl
index 2a411f321b..a6c069e90a 100644
--- a/tools/rules/heron_examples.bzl
+++ b/tools/rules/heron_examples.bzl
@@ -15,7 +15,8 @@ def heron_examples_conf_files():
def heron_examples_yaml_files():
return [
- "//eco-examples/src/java:heron-eco-examples-support",
+ "//eco-storm-examples/src/java:storm-eco-examples-support",
+ "//eco-heron-examples/src/java:heron-eco-examples-support",
]
def heron_examples_lib_files():
@@ -23,5 +24,6 @@ def heron_examples_lib_files():
"//examples/src/java:heron-api-examples",
"//examples/src/java:heron-streamlet-examples",
"//examples/src/scala:heron-streamlet-scala-examples",
- "//eco-examples/src/java:heron-eco-examples",
+ "//eco-storm-examples/src/java:storm-eco-examples",
+ "//eco-heron-examples/src/java:heron-eco-examples",
]
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services