You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2015/08/17 13:28:15 UTC
[1/6] storm git commit: Storm-Kafka trident topology example
Repository: storm
Updated Branches:
refs/heads/master 87fc2982e -> 285d3288c
Storm-Kafka trident topology example
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/287718c8
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/287718c8
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/287718c8
Branch: refs/heads/master
Commit: 287718c8bc0f8b51ca89de4f62fcb6710d525992
Parents: f72beb0
Author: Arun Mahadevan <ai...@hortonworks.com>
Authored: Tue Aug 4 14:06:45 2015 +0530
Committer: Arun Mahadevan <ai...@hortonworks.com>
Committed: Tue Aug 4 14:06:45 2015 +0530
----------------------------------------------------------------------
examples/storm-starter/pom.xml | 6 +
.../starter/trident/TridentKafkaWordCount.java | 208 +++++++++++++++++++
2 files changed, 214 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/287718c8/examples/storm-starter/pom.xml
----------------------------------------------------------------------
diff --git a/examples/storm-starter/pom.xml b/examples/storm-starter/pom.xml
index db69d74..36f7b65 100644
--- a/examples/storm-starter/pom.xml
+++ b/examples/storm-starter/pom.xml
@@ -96,6 +96,12 @@
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-kafka</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
</dependencies>
<build>
http://git-wip-us.apache.org/repos/asf/storm/blob/287718c8/examples/storm-starter/src/jvm/storm/starter/trident/TridentKafkaWordCount.java
----------------------------------------------------------------------
diff --git a/examples/storm-starter/src/jvm/storm/starter/trident/TridentKafkaWordCount.java b/examples/storm-starter/src/jvm/storm/starter/trident/TridentKafkaWordCount.java
new file mode 100644
index 0000000..11a9111
--- /dev/null
+++ b/examples/storm-starter/src/jvm/storm/starter/trident/TridentKafkaWordCount.java
@@ -0,0 +1,208 @@
+package storm.starter.trident;
+
+
+import backtype.storm.Config;
+import backtype.storm.LocalCluster;
+import backtype.storm.LocalDRPC;
+import backtype.storm.generated.StormTopology;
+import backtype.storm.spout.SchemeAsMultiScheme;
+import backtype.storm.topology.TopologyBuilder;
+import backtype.storm.tuple.Fields;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import storm.kafka.StringScheme;
+import storm.kafka.ZkHosts;
+import storm.kafka.bolt.KafkaBolt;
+import storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaMapper;
+import storm.kafka.bolt.selector.DefaultTopicSelector;
+import storm.kafka.trident.TransactionalTridentKafkaSpout;
+import storm.kafka.trident.TridentKafkaConfig;
+import storm.starter.spout.RandomSentenceSpout;
+import storm.trident.Stream;
+import storm.trident.TridentState;
+import storm.trident.TridentTopology;
+import storm.trident.operation.builtin.Count;
+import storm.trident.operation.builtin.FilterNull;
+import storm.trident.operation.builtin.MapGet;
+import storm.trident.testing.MemoryMapState;
+import storm.trident.testing.Split;
+
+import java.util.Properties;
+
+/**
+ * A sample word count trident topology using transactional kafka spout that has the following components.
+ * <ol>
+ * <li> {@link KafkaBolt}
+ * that receives random sentences from {@link RandomSentenceSpout} and
+ * publishes the sentences to a kafka "test" topic.
+ * </li>
+ * <li> {@link TransactionalTridentKafkaSpout}
+ * that consumes sentences from the "test" topic, splits it into words, aggregates
+ * and stores the word count in a {@link MemoryMapState}.
+ * </li>
+ * <li> DRPC query
+ * that returns the word counts by querying the trident state (MemoryMapState).
+ * </li>
+ * </ol>
+ * <p>
+ * For more background read the <a href="https://storm.apache.org/documentation/Trident-tutorial.html">trident tutorial</a>,
+ * <a href="https://storm.apache.org/documentation/Trident-state">trident state</a> and
+ * <a href="https://github.com/apache/storm/tree/master/external/storm-kafka"> Storm Kafka </a>.
+ * </p>
+ */
+public class TridentKafkaWordCount {
+
+ private String zkUrl;
+ private String brokerUrl;
+
+ TridentKafkaWordCount(String zkUrl, String brokerUrl) {
+ this.zkUrl = zkUrl;
+ this.brokerUrl = brokerUrl;
+ }
+
+ /**
+ * Creates a transactional kafka spout that consumes any new data published to "test" topic.
+ * <p/>
+ * For more info on transactional spouts
+ * see "Transactional spouts" section in
+ * <a href="https://storm.apache.org/documentation/Trident-state"> Trident state</a> doc.
+ *
+ * @return a transactional trident kafka spout.
+ */
+ private TransactionalTridentKafkaSpout createKafkaSpout() {
+ ZkHosts hosts = new ZkHosts(zkUrl);
+ TridentKafkaConfig config = new TridentKafkaConfig(hosts, "test");
+ config.scheme = new SchemeAsMultiScheme(new StringScheme());
+
+ // Consume new data from the topic
+ config.startOffsetTime = kafka.api.OffsetRequest.LatestTime();
+ return new TransactionalTridentKafkaSpout(config);
+ }
+
+
+ private Stream addDRPCStream(TridentTopology tridentTopology, TridentState state, LocalDRPC drpc) {
+ return tridentTopology.newDRPCStream("words", drpc)
+ .each(new Fields("args"), new Split(), new Fields("word"))
+ .groupBy(new Fields("word"))
+ .stateQuery(state, new Fields("word"), new MapGet(), new Fields("count"))
+ .each(new Fields("count"), new FilterNull())
+ .project(new Fields("word", "count"));
+ }
+
+ private TridentState addTridentState(TridentTopology tridentTopology) {
+ return tridentTopology.newStream("spout1", createKafkaSpout()).parallelismHint(1)
+ .each(new Fields("str"), new Split(), new Fields("word"))
+ .groupBy(new Fields("word"))
+ .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
+ .parallelismHint(1);
+ }
+
+ /**
+ * Creates a trident topology that consumes sentences from the kafka "test" topic using a
+ * {@link TransactionalTridentKafkaSpout} computes the word count and stores it in a {@link MemoryMapState}.
+ * A DRPC stream is then created to query the word counts.
+ * @param drpc
+ * @return
+ */
+ public StormTopology buildConsumerTopology(LocalDRPC drpc) {
+ TridentTopology tridentTopology = new TridentTopology();
+ addDRPCStream(tridentTopology, addTridentState(tridentTopology), drpc);
+ return tridentTopology.build();
+ }
+
+ /**
+ * Return the consumer topology config.
+ *
+ * @return the topology config
+ */
+ public Config getConsumerConfig() {
+ Config conf = new Config();
+ conf.setMaxSpoutPending(20);
+ // conf.setDebug(true);
+ return conf;
+ }
+
+ /**
+ * A topology that produces random sentences using {@link RandomSentenceSpout} and
+ * publishes the sentences using a KafkaBolt to kafka "test" topic.
+ *
+ * @return the storm topology
+ */
+ public StormTopology buildProducerTopology() {
+ TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("spout", new RandomSentenceSpout(), 2);
+ /**
+ * The output field of the RandomSentenceSpout ("word") is provided as the boltMessageField
+ * so that this gets written out as the message in the kafka topic.
+ */
+ KafkaBolt bolt = new KafkaBolt()
+ .withTopicSelector(new DefaultTopicSelector("test"))
+ .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper("key", "word"));
+ builder.setBolt("forwardToKafka", bolt, 1).shuffleGrouping("spout");
+ return builder.createTopology();
+ }
+
+ /**
+ * Returns the storm config for the topology that publishes sentences to kafka "test" topic using a kafka bolt.
+ * The KAFKA_BROKER_PROPERTIES is needed for the KafkaBolt.
+ *
+ * @return the topology config
+ */
+ public Config getProducerConfig() {
+ Config conf = new Config();
+ conf.setMaxSpoutPending(20);
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+ props.put(ProducerConfig.CLIENT_ID_CONFIG, "storm-kafka-producer");
+ conf.put(KafkaBolt.KAFKA_BROKER_PROPERTIES, props);
+ return conf;
+ }
+
+ /**
+ * <p>
+ * To run this topology ensure you have a kafka broker running.
+ * </p>
+ * Create a topic test with command line,
+ * kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partition 1 --topic test
+ */
+ public static void main(String[] args) throws Exception {
+
+ String zkUrl = "localhost:2181"; // the defaults.
+ String brokerUrl = "localhost:9092";
+
+ if (args.length > 2 || (args.length == 1 && args[0].matches("^-h|--help$"))) {
+ System.out.println("Usage: TridentKafkaWordCount [kafka zookeeper url] [kafka broker url]");
+ System.out.println(" E.g TridentKafkaWordCount [" + zkUrl + "]" + " [" + brokerUrl + "]");
+ System.exit(1);
+ } else if (args.length == 1) {
+ zkUrl = args[0];
+ } else if (args.length == 2) {
+ zkUrl = args[0];
+ brokerUrl = args[1];
+ }
+
+ System.out.println("Using Kafka zookeeper url: " + zkUrl + " broker url: " + brokerUrl);
+
+ TridentKafkaWordCount wordCount = new TridentKafkaWordCount(zkUrl, brokerUrl);
+
+ LocalDRPC drpc = new LocalDRPC();
+ LocalCluster cluster = new LocalCluster();
+
+ // submit the consumer topology.
+ cluster.submitTopology("wordCounter", wordCount.getConsumerConfig(), wordCount.buildConsumerTopology(drpc));
+
+ // submit the producer topology.
+ cluster.submitTopology("kafkaBolt", wordCount.getProducerConfig(), wordCount.buildProducerTopology());
+
+ // keep querying the word counts for a minute.
+ for (int i = 0; i < 60; i++) {
+ System.out.println("DRPC RESULT: " + drpc.execute("words", "the and apple snow jumped"));
+ Thread.sleep(1000);
+ }
+
+ cluster.killTopology("kafkaBolt");
+ cluster.killTopology("wordCounter");
+ cluster.shutdown();
+ }
+}
\ No newline at end of file
[5/6] storm git commit: Merge branch 'storm-kafka-example' of
https://github.com/arunmahadevan/storm into STORM-975
Posted by ka...@apache.org.
Merge branch 'storm-kafka-example' of https://github.com/arunmahadevan/storm into STORM-975
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/2295ac2d
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/2295ac2d
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/2295ac2d
Branch: refs/heads/master
Commit: 2295ac2da711eb81e9c4bb34c118926d62a3b2ec
Parents: 87fc298 0fa672f
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Mon Aug 17 18:27:08 2015 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Mon Aug 17 18:27:08 2015 +0900
----------------------------------------------------------------------
examples/storm-starter/pom.xml | 29 +++
.../starter/trident/TridentKafkaWordCount.java | 230 +++++++++++++++++++
pom.xml | 2 +-
3 files changed, 260 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
[6/6] storm git commit: add STORM-975 to CHANGELOG.md
Posted by ka...@apache.org.
add STORM-975 to CHANGELOG.md
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/285d3288
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/285d3288
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/285d3288
Branch: refs/heads/master
Commit: 285d3288c4bdaf3793621fdac135f84234d979e6
Parents: 2295ac2
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Mon Aug 17 20:27:53 2015 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Mon Aug 17 20:27:53 2015 +0900
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/285d3288/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f2e2f16..a2ad498 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -31,6 +31,7 @@
* STORM-845: Storm ElasticSearch connector
* STORM-944: storm-hive pom.xml has a dependency conflict with calcite
* STORM-988: supervisor.slots.ports should not allow duplicate element values
+ * STORM-975: Storm-Kafka trident topology example
## 0.10.0-beta2
* STORM-843: [storm-redis] Add Javadoc to storm-redis
[3/6] storm git commit: merging upstream/master
Posted by ka...@apache.org.
merging upstream/master
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/c8e337d1
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/c8e337d1
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/c8e337d1
Branch: refs/heads/master
Commit: c8e337d13521e6ca5f83639b9a68120618750332
Parents: 62fa05b 544e55c
Author: Arun Mahadevan <ai...@hortonworks.com>
Authored: Tue Aug 11 10:27:48 2015 +0530
Committer: Arun Mahadevan <ai...@hortonworks.com>
Committed: Tue Aug 11 10:27:48 2015 +0530
----------------------------------------------------------------------
CHANGELOG.md | 16 ++-
DEVELOPER.md | 7 +-
README.markdown | 1 +
external/storm-elasticsearch/README.md | 72 ++++++++++
external/storm-elasticsearch/pom.xml | 95 +++++++++++++
.../elasticsearch/bolt/AbstractEsBolt.java | 81 +++++++++++
.../storm/elasticsearch/bolt/EsIndexBolt.java | 68 ++++++++++
.../elasticsearch/bolt/EsPercolateBolt.java | 78 +++++++++++
.../storm/elasticsearch/common/EsConfig.java | 54 ++++++++
.../storm/elasticsearch/trident/EsState.java | 117 ++++++++++++++++
.../elasticsearch/trident/EsStateFactory.java | 50 +++++++
.../storm/elasticsearch/trident/EsUpdater.java | 31 +++++
.../elasticsearch/bolt/AbstractEsBoltTest.java | 81 +++++++++++
.../elasticsearch/bolt/EsIndexBoltTest.java | 68 ++++++++++
.../elasticsearch/bolt/EsIndexTopology.java | 120 +++++++++++++++++
.../elasticsearch/bolt/EsPercolateBoltTest.java | 65 +++++++++
.../storm/elasticsearch/common/EsConstants.java | 22 +++
.../storm/elasticsearch/common/EsTestUtil.java | 70 ++++++++++
.../trident/TridentEsTopology.java | 135 +++++++++++++++++++
external/storm-hive/README.md | 1 +
.../org/apache/storm/hive/bolt/HiveBolt.java | 44 ++++--
.../apache/storm/hive/common/HiveWriter.java | 5 +-
.../apache/storm/hive/bolt/TestHiveBolt.java | 100 ++++++++++++--
.../src/jvm/storm/kafka/ZkCoordinator.java | 2 +-
external/storm-redis/README.md | 1 +
log4j2/cluster.xml | 6 +-
log4j2/worker.xml | 6 +-
pom.xml | 1 +
storm-core/src/clj/backtype/storm/cluster.clj | 2 +-
.../src/clj/backtype/storm/daemon/task.clj | 1 -
storm-core/src/clj/backtype/storm/tuple.clj | 6 +-
storm-core/src/clj/backtype/storm/zookeeper.clj | 26 ++--
storm-core/src/jvm/backtype/storm/Config.java | 2 +-
.../jvm/backtype/storm/ConfigValidation.java | 15 +--
.../src/jvm/backtype/storm/drpc/DRPCSpout.java | 6 +-
.../storm/grouping/PartialKeyGrouping.java | 27 +++-
.../storm/testing/TestWordBytesCounter.java | 27 ++++
.../backtype/storm/testing/TestWordCounter.java | 6 +-
.../src/jvm/storm/trident/TridentTopology.java | 13 +-
.../test/clj/backtype/storm/config_test.clj | 19 ++-
.../test/clj/backtype/storm/grouping_test.clj | 43 +++---
.../test/clj/backtype/storm/nimbus_test.clj | 29 ++++
storm-dist/binary/src/main/assembly/binary.xml | 14 ++
43 files changed, 1531 insertions(+), 102 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/c8e337d1/pom.xml
----------------------------------------------------------------------
diff --cc pom.xml
index 646181d,cb00783..a79f5ec
--- a/pom.xml
+++ b/pom.xml
@@@ -169,7 -170,7 +169,8 @@@
<module>external/storm-redis</module>
<module>external/storm-eventhubs</module>
<module>external/flux</module>
+ <module>examples/storm-starter</module>
+ <module>external/storm-elasticsearch</module>
</modules>
<scm>
[2/6] storm git commit: added transitive dependencies for storm kafka
example to build; moved storm-starter as the last module in the main pom.xml
Posted by ka...@apache.org.
added transitive dependencies for storm kafka example to build; moved storm-starter as the last module in the main pom.xml
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/62fa05b9
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/62fa05b9
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/62fa05b9
Branch: refs/heads/master
Commit: 62fa05b97cb7e1c0dc3eca8e2ce05fd78a8b88c8
Parents: 287718c
Author: Arun Mahadevan <ai...@hortonworks.com>
Authored: Tue Aug 4 15:25:49 2015 +0530
Committer: Arun Mahadevan <ai...@hortonworks.com>
Committed: Tue Aug 4 15:25:49 2015 +0530
----------------------------------------------------------------------
examples/storm-starter/pom.xml | 23 +++++++++++++++++++++++
pom.xml | 2 +-
2 files changed, 24 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/62fa05b9/examples/storm-starter/pom.xml
----------------------------------------------------------------------
diff --git a/examples/storm-starter/pom.xml b/examples/storm-starter/pom.xml
index 36f7b65..b24c436 100644
--- a/examples/storm-starter/pom.xml
+++ b/examples/storm-starter/pom.xml
@@ -102,6 +102,29 @@
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_2.10</artifactId>
+ <version>0.8.2.1</version>
+ <!-- use provided scope, so users can pull in whichever scala version they choose -->
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.zookeeper</groupId>
+ <artifactId>zookeeper</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>0.8.2.1</version>
+ <scope>provided</scope>
+ </dependency>
</dependencies>
<build>
http://git-wip-us.apache.org/repos/asf/storm/blob/62fa05b9/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index bd65f04..646181d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -161,7 +161,6 @@
<module>storm-multilang/python</module>
<module>storm-multilang/ruby</module>
<module>storm-core</module>
- <module>examples/storm-starter</module>
<module>external/storm-kafka</module>
<module>external/storm-hdfs</module>
<module>external/storm-hbase</module>
@@ -170,6 +169,7 @@
<module>external/storm-redis</module>
<module>external/storm-eventhubs</module>
<module>external/flux</module>
+ <module>examples/storm-starter</module>
</modules>
<scm>
[4/6] storm git commit: added apache license
Posted by ka...@apache.org.
added apache license
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/0fa672f4
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/0fa672f4
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/0fa672f4
Branch: refs/heads/master
Commit: 0fa672f43672a01c842c7f542731b08e1e523feb
Parents: c8e337d
Author: Arun Mahadevan <ai...@hortonworks.com>
Authored: Sat Aug 15 09:42:43 2015 +0530
Committer: Arun Mahadevan <ai...@hortonworks.com>
Committed: Sat Aug 15 09:42:43 2015 +0530
----------------------------------------------------------------------
.../starter/trident/TridentKafkaWordCount.java | 24 +++++++++++++++++++-
1 file changed, 23 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/0fa672f4/examples/storm-starter/src/jvm/storm/starter/trident/TridentKafkaWordCount.java
----------------------------------------------------------------------
diff --git a/examples/storm-starter/src/jvm/storm/starter/trident/TridentKafkaWordCount.java b/examples/storm-starter/src/jvm/storm/starter/trident/TridentKafkaWordCount.java
index 11a9111..813841a 100644
--- a/examples/storm-starter/src/jvm/storm/starter/trident/TridentKafkaWordCount.java
+++ b/examples/storm-starter/src/jvm/storm/starter/trident/TridentKafkaWordCount.java
@@ -1,3 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * Contains some contributions under the Thrift Software License.
+ * Please see doc/old-thrift-license.txt in the Thrift distribution for
+ * details.
+ */
package storm.starter.trident;
@@ -205,4 +227,4 @@ public class TridentKafkaWordCount {
cluster.killTopology("wordCounter");
cluster.shutdown();
}
-}
\ No newline at end of file
+}