You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by sr...@apache.org on 2016/11/15 06:35:35 UTC
[2/4] storm git commit: STORM-2182: Refactor Storm Kafka Examples
Into Own Modules - Created modules - storm-kafka-examples -
storm-kafka-client-examples
STORM-2182: Refactor Storm Kafka Examples Into Own Modules
- Created modules
- storm-kafka-examples
- storm-kafka-client-examples
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/4d22a68a
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/4d22a68a
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/4d22a68a
Branch: refs/heads/master
Commit: 4d22a68abb94a0f2a4bcd5cdc4d512ae0a9a0590
Parents: 76d6e7e
Author: Hugo Louro <hm...@gmail.com>
Authored: Tue Oct 25 15:48:21 2016 -0700
Committer: Hugo Louro <hm...@gmail.com>
Committed: Tue Nov 1 10:10:04 2016 -0700
----------------------------------------------------------------------
examples/storm-kafka-client-examples/pom.xml | 134 +++++++++
.../TridentKafkaClientWordCountNamedTopics.java | 156 ++++++++++
...identKafkaClientWordCountWildcardTopics.java | 46 +++
examples/storm-kafka-examples/pom.xml | 18 ++
.../storm/kafka/TridentKafkaTopology.java | 91 ------
.../kafka/trident/KafkaProducerTopology.java | 67 +++++
.../storm/kafka/trident/LocalSubmitter.java | 91 ++++++
.../trident/TridentKafkaConsumerTopology.java | 94 +++++++
.../kafka/trident/TridentKafkaTopology.java | 89 ++++++
.../kafka/trident/TridentKafkaWordCount.java | 140 +++++++++
examples/storm-starter/pom.xml | 22 --
.../TridentKafkaClientWordCountNamedTopics.java | 122 --------
...identKafkaClientWordCountWildcardTopics.java | 51 ----
.../starter/trident/TridentKafkaWordCount.java | 282 -------------------
.../kafka/trident/OpaqueTridentKafkaSpout.java | 5 +-
pom.xml | 1 +
16 files changed, 838 insertions(+), 571 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-kafka-client-examples/pom.xml
----------------------------------------------------------------------
diff --git a/examples/storm-kafka-client-examples/pom.xml b/examples/storm-kafka-client-examples/pom.xml
new file mode 100644
index 0000000..cb6db94
--- /dev/null
+++ b/examples/storm-kafka-client-examples/pom.xml
@@ -0,0 +1,134 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>storm</artifactId>
+ <groupId>org.apache.storm</groupId>
+ <version>2.0.0-SNAPSHOT</version>
+ <relativePath>../../pom.xml</relativePath>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>storm-kafka-client-examples</artifactId>
+
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <provided.scope>provided</provided.scope>
+ </properties>
+
+ <profiles>
+ <profile>
+ <id>intellij</id>
+ <properties>
+ <provided.scope>compile</provided.scope>
+ </properties>
+ </profile>
+ </profiles>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-core</artifactId>
+ <version>${project.version}</version>
+ <scope>${provided.scope}</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-kafka</artifactId>
+ <version>${project.version}</version>
+ <scope>${provided.scope}</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>${storm.kafka.artifact.id}</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-kafka-examples</artifactId>
+ <version>${project.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>${storm.kafka.artifact.id}</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-kafka-client</artifactId>
+ <version>${project.version}</version>
+ <scope>${provided.scope}</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>${storm.kafka.artifact.id}</artifactId>
+ <version>${storm.kafka.client.version}</version>
+ <scope>${provided.scope}</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${storm.kafka.client.version}</version>
+ <scope>${provided.scope}</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <configuration>
+ <createDependencyReducedPom>true</createDependencyReducedPom>
+ </configuration>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass>org.apache.storm.kafka.trident.TridentKafkaClientWordCountNamedTopics</mainClass>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountNamedTopics.java
----------------------------------------------------------------------
diff --git a/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountNamedTopics.java b/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountNamedTopics.java
new file mode 100644
index 0000000..2482eae
--- /dev/null
+++ b/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountNamedTopics.java
@@ -0,0 +1,156 @@
+/*
+ * 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.
+ */
+
+package org.apache.storm.kafka.trident;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.storm.Config;
+import org.apache.storm.StormSubmitter;
+import org.apache.storm.generated.AlreadyAliveException;
+import org.apache.storm.generated.AuthorizationException;
+import org.apache.storm.generated.InvalidTopologyException;
+import org.apache.storm.kafka.spout.KafkaSpoutConfig;
+import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff;
+import org.apache.storm.kafka.spout.KafkaSpoutRetryService;
+import org.apache.storm.kafka.spout.KafkaSpoutStreams;
+import org.apache.storm.kafka.spout.KafkaSpoutStreamsNamedTopics;
+import org.apache.storm.kafka.spout.KafkaSpoutTupleBuilder;
+import org.apache.storm.kafka.spout.KafkaSpoutTuplesBuilder;
+import org.apache.storm.kafka.spout.KafkaSpoutTuplesBuilderNamedTopics;
+import org.apache.storm.kafka.spout.trident.KafkaTridentSpoutManager;
+import org.apache.storm.kafka.spout.trident.KafkaTridentSpoutOpaque;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST;
+
+public class TridentKafkaClientWordCountNamedTopics {
+ private static final String TOPIC_1 = "test-trident";
+ private static final String TOPIC_2 = "test-trident-1";
+ private static final String KAFKA_LOCAL_BROKER = "localhost:9092";
+
+ private KafkaTridentSpoutOpaque<String, String> newKafkaTridentSpoutOpaque() {
+ return new KafkaTridentSpoutOpaque<>(new KafkaTridentSpoutManager<>(
+ newKafkaSpoutConfig(
+ newKafkaSpoutStreams())));
+ }
+
+ private KafkaSpoutConfig<String,String> newKafkaSpoutConfig(KafkaSpoutStreams kafkaSpoutStreams) {
+ return new KafkaSpoutConfig.Builder<>(newKafkaConsumerProps(),
+ kafkaSpoutStreams, newTuplesBuilder(), newRetryService())
+ .setOffsetCommitPeriodMs(10_000)
+ .setFirstPollOffsetStrategy(EARLIEST)
+ .setMaxUncommittedOffsets(250)
+ .build();
+ }
+
+ protected Map<String,Object> newKafkaConsumerProps() {
+ Map<String, Object> props = new HashMap<>();
+ props.put(KafkaSpoutConfig.Consumer.BOOTSTRAP_SERVERS, "127.0.0.1:9092");
+ props.put(KafkaSpoutConfig.Consumer.GROUP_ID, "kafkaSpoutTestGroup");
+ props.put(KafkaSpoutConfig.Consumer.KEY_DESERIALIZER, "org.apache.kafka.common.serialization.StringDeserializer");
+ props.put(KafkaSpoutConfig.Consumer.VALUE_DESERIALIZER, "org.apache.kafka.common.serialization.StringDeserializer");
+ props.put("max.partition.fetch.bytes", 200);
+ return props;
+ }
+
+ protected KafkaSpoutTuplesBuilder<String, String> newTuplesBuilder() {
+ return new KafkaSpoutTuplesBuilderNamedTopics.Builder<>(
+ new TopicsTupleBuilder<String, String>(TOPIC_1, TOPIC_2))
+ .build();
+ }
+
+ protected KafkaSpoutRetryService newRetryService() {
+ return new KafkaSpoutRetryExponentialBackoff(new KafkaSpoutRetryExponentialBackoff.TimeInterval(500L, TimeUnit.MICROSECONDS),
+ KafkaSpoutRetryExponentialBackoff.TimeInterval.milliSeconds(2),
+ Integer.MAX_VALUE, KafkaSpoutRetryExponentialBackoff.TimeInterval.seconds(10));
+ }
+
+ protected KafkaSpoutStreams newKafkaSpoutStreams() {
+ return new KafkaSpoutStreamsNamedTopics.Builder(new Fields("str"), new String[]{"test-trident","test-trident-1"}).build();
+ }
+
+ protected static class TopicsTupleBuilder<K, V> extends KafkaSpoutTupleBuilder<K,V> {
+ public TopicsTupleBuilder(String... topics) {
+ super(topics);
+ }
+ @Override
+ public List<Object> buildTuple(ConsumerRecord<K, V> consumerRecord) {
+ return new Values(consumerRecord.value());
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ new TridentKafkaClientWordCountNamedTopics().run(args);
+ }
+
+ protected void run(String[] args) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException {
+ if (args.length > 0 && Arrays.stream(args).anyMatch(option -> option.equals("-h"))) {
+ System.out.printf("Usage: java %s [%s] [%s] [%s] [%s]\n", getClass().getName(),
+ "broker_host:broker_port", "topic1", "topic2", "topology_name");
+ } else {
+ final String brokerUrl = args.length > 0 ? args[0] : KAFKA_LOCAL_BROKER;
+ final String topic1 = args.length > 1 ? args[1] : TOPIC_1;
+ final String topic2 = args.length > 2 ? args[2] : TOPIC_2;
+
+ System.out.printf("Running with broker_url: [%s], topics: [%s, %s]\n", brokerUrl, topic1, topic2);
+
+ Config tpConf = LocalSubmitter.defaultConfig();
+
+ if (args.length == 4) { //Submit Remote
+ // Producers
+ StormSubmitter.submitTopology(topic1 + "-producer", tpConf, KafkaProducerTopology.newTopology(brokerUrl, topic1));
+ StormSubmitter.submitTopology(topic2 + "-producer", tpConf, KafkaProducerTopology.newTopology(brokerUrl, topic2));
+ // Consumer
+ StormSubmitter.submitTopology("topics-consumer", tpConf, TridentKafkaConsumerTopology.newTopology(newKafkaTridentSpoutOpaque()));
+ } else { //Submit Local
+
+ final LocalSubmitter localSubmitter = LocalSubmitter.newInstance();
+ final String topic1Tp = "topic1-producer";
+ final String topic2Tp = "topic2-producer";
+ final String consTpName = "topics-consumer";
+
+ try {
+ // Producers
+ localSubmitter.submit(topic1Tp, tpConf, KafkaProducerTopology.newTopology(brokerUrl, topic1));
+ localSubmitter.submit(topic2Tp, tpConf, KafkaProducerTopology.newTopology(brokerUrl, topic2));
+ // Consumer
+ localSubmitter.submit(consTpName, tpConf, TridentKafkaConsumerTopology.newTopology(
+ localSubmitter.getDrpc(), newKafkaTridentSpoutOpaque()));
+
+ // print
+ localSubmitter.printResults(15, 1, TimeUnit.SECONDS);
+ } finally {
+ // kill
+ localSubmitter.kill(topic1Tp);
+ localSubmitter.kill(topic2Tp);
+ localSubmitter.kill(consTpName);
+ // shutdown
+ localSubmitter.shutdown();
+ }
+ }
+ }
+ System.exit(0); // Kill all the non daemon threads
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountWildcardTopics.java
----------------------------------------------------------------------
diff --git a/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountWildcardTopics.java b/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountWildcardTopics.java
new file mode 100644
index 0000000..e33f19d
--- /dev/null
+++ b/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountWildcardTopics.java
@@ -0,0 +1,46 @@
+/*
+ * 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.
+ */
+
+package org.apache.storm.kafka.trident;
+
+import org.apache.storm.kafka.spout.KafkaSpoutStream;
+import org.apache.storm.kafka.spout.KafkaSpoutStreams;
+import org.apache.storm.kafka.spout.KafkaSpoutStreamsWildcardTopics;
+import org.apache.storm.kafka.spout.KafkaSpoutTuplesBuilder;
+import org.apache.storm.kafka.spout.KafkaSpoutTuplesBuilderWildcardTopics;
+import org.apache.storm.tuple.Fields;
+
+import java.util.regex.Pattern;
+
+public class TridentKafkaClientWordCountWildcardTopics extends TridentKafkaClientWordCountNamedTopics {
+ private static final String TOPIC_WILDCARD_PATTERN = "test-trident(-1)?";
+
+ protected KafkaSpoutTuplesBuilder<String, String> newTuplesBuilder() {
+ return new KafkaSpoutTuplesBuilderWildcardTopics<>(new TopicsTupleBuilder<>(TOPIC_WILDCARD_PATTERN));
+ }
+
+ protected KafkaSpoutStreams newKafkaSpoutStreams() {
+ final Fields outputFields = new Fields("str");
+ final KafkaSpoutStream kafkaSpoutStream = new KafkaSpoutStream(outputFields, Pattern.compile(TOPIC_WILDCARD_PATTERN));
+ return new KafkaSpoutStreamsWildcardTopics(kafkaSpoutStream);
+ }
+
+ public static void main(String[] args) throws Exception {
+ new TridentKafkaClientWordCountWildcardTopics().run(args);
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-kafka-examples/pom.xml
----------------------------------------------------------------------
diff --git a/examples/storm-kafka-examples/pom.xml b/examples/storm-kafka-examples/pom.xml
index 05d0363..d809157 100644
--- a/examples/storm-kafka-examples/pom.xml
+++ b/examples/storm-kafka-examples/pom.xml
@@ -44,6 +44,11 @@
<dependencies>
<dependency>
<groupId>org.apache.storm</groupId>
+ <artifactId>storm-starter</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>${project.version}</version>
<scope>${provided.scope}</scope>
@@ -52,6 +57,19 @@
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka</artifactId>
<version>${project.version}</version>
+ <scope>${provided.scope}</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>${storm.kafka.artifact.id}</artifactId>
+ <version>${storm.kafka.version}</version>
+ <scope>${provided.scope}</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${storm.kafka.version}</version>
+ <scope>${provided.scope}</scope>
</dependency>
</dependencies>
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/TridentKafkaTopology.java
----------------------------------------------------------------------
diff --git a/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/TridentKafkaTopology.java b/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/TridentKafkaTopology.java
deleted file mode 100644
index fdc6752..0000000
--- a/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/TridentKafkaTopology.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/**
- * 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.
- */
-package org.apache.storm.kafka;
-
-import org.apache.storm.Config;
-import org.apache.storm.LocalCluster;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Values;
-import com.google.common.collect.ImmutableMap;
-import org.apache.storm.kafka.trident.TridentKafkaStateFactory;
-import org.apache.storm.kafka.trident.TridentKafkaUpdater;
-import org.apache.storm.kafka.trident.mapper.FieldNameBasedTupleToKafkaMapper;
-import org.apache.storm.kafka.trident.selector.DefaultTopicSelector;
-import org.apache.storm.trident.Stream;
-import org.apache.storm.trident.TridentTopology;
-import org.apache.storm.trident.testing.FixedBatchSpout;
-
-import java.util.Properties;
-
-public class TridentKafkaTopology {
-
- private static StormTopology buildTopology(String brokerConnectionString) {
- Fields fields = new Fields("word", "count");
- FixedBatchSpout spout = new FixedBatchSpout(fields, 4,
- new Values("storm", "1"),
- new Values("trident", "1"),
- new Values("needs", "1"),
- new Values("javadoc", "1")
- );
- spout.setCycle(true);
-
- TridentTopology topology = new TridentTopology();
- Stream stream = topology.newStream("spout1", spout);
-
- Properties props = new Properties();
- props.put("bootstrap.servers", brokerConnectionString);
- props.put("acks", "1");
- props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
- props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
-
- TridentKafkaStateFactory stateFactory = new TridentKafkaStateFactory()
- .withProducerProperties(props)
- .withKafkaTopicSelector(new DefaultTopicSelector("test"))
- .withTridentTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper("word", "count"));
- stream.partitionPersist(stateFactory, fields, new TridentKafkaUpdater(), new Fields());
-
- return topology.build();
- }
-
- /**
- * To run this topology ensure you have a kafka broker running and provide connection string to broker as argument.
- * Create a topic test with command line,
- * kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partition 1 --topic test
- *
- * run this program and run the kafka consumer:
- * kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
- *
- * you should see the messages flowing through.
- *
- * @param args
- * @throws Exception
- */
- public static void main(String[] args) throws Exception {
- if(args.length < 1) {
- System.out.println("Please provide kafka broker url ,e.g. localhost:9092");
- }
-
- LocalCluster cluster = new LocalCluster();
- cluster.submitTopology("wordCounter", new Config(), buildTopology(args[0]));
- Thread.sleep(60 * 1000);
- cluster.killTopology("wordCounter");
-
- cluster.shutdown();
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/KafkaProducerTopology.java
----------------------------------------------------------------------
diff --git a/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/KafkaProducerTopology.java b/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/KafkaProducerTopology.java
new file mode 100644
index 0000000..56bfb83
--- /dev/null
+++ b/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/KafkaProducerTopology.java
@@ -0,0 +1,67 @@
+/*
+ * 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.
+ */
+
+package org.apache.storm.kafka.trident;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.kafka.bolt.KafkaBolt;
+import org.apache.storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaMapper;
+import org.apache.storm.kafka.bolt.selector.DefaultTopicSelector;
+import org.apache.storm.starter.spout.RandomSentenceSpout;
+import org.apache.storm.topology.TopologyBuilder;
+
+import java.util.Properties;
+
+public class KafkaProducerTopology {
+ /**
+ * @param brokerUrl Kafka broker URL
+ * @param topicName Topic to which publish sentences
+ * @return A Storm topology that produces random sentences using {@link RandomSentenceSpout} and uses a {@link KafkaBolt} to
+ * publish the sentences to the kafka topic specified
+ */
+ public static StormTopology newTopology(String brokerUrl, String topicName) {
+ final TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("spout", new RandomSentenceSpout.TimeStamped(""), 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. */
+ final KafkaBolt<String, String> bolt = new KafkaBolt<String, String>()
+ .withProducerProperties(newProps(brokerUrl, topicName))
+ .withTopicSelector(new DefaultTopicSelector(topicName))
+ .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper<>("key", "word"));
+
+ builder.setBolt("forwardToKafka", bolt, 1).shuffleGrouping("spout");
+
+ return builder.createTopology();
+ }
+
+ /**
+ * @return the Storm config for the topology that publishes sentences to kafka using a kafka bolt.
+ */
+ private static Properties newProps(final String brokerUrl, final String topicName) {
+ return new Properties() {{
+ put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
+ put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+ put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+ put(ProducerConfig.CLIENT_ID_CONFIG, topicName);
+ }};
+ }
+}
+
+
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/LocalSubmitter.java
----------------------------------------------------------------------
diff --git a/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/LocalSubmitter.java b/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/LocalSubmitter.java
new file mode 100644
index 0000000..a47c67e
--- /dev/null
+++ b/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/LocalSubmitter.java
@@ -0,0 +1,91 @@
+/*
+ * 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.
+ */
+
+package org.apache.storm.kafka.trident;
+
+import org.apache.storm.Config;
+import org.apache.storm.LocalCluster;
+import org.apache.storm.LocalDRPC;
+import org.apache.storm.generated.StormTopology;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.TimeUnit;
+
+public class LocalSubmitter {
+ protected static final Logger LOG = LoggerFactory.getLogger(LocalSubmitter.class);
+
+ private LocalDRPC drpc;
+ private LocalCluster cluster;
+
+ public LocalSubmitter(LocalDRPC drpc, LocalCluster cluster) {
+ this.drpc = drpc;
+ this.cluster = cluster;
+ }
+
+ public static LocalSubmitter newInstance() {
+ return new LocalSubmitter(new LocalDRPC(), new LocalCluster());
+ }
+
+ public static Config defaultConfig() {
+ final Config conf = new Config();
+ conf.setMaxSpoutPending(20);
+ conf.setMaxTaskParallelism(1);
+ conf.setNumWorkers(1);
+ return conf;
+ }
+
+ public LocalSubmitter(StormTopology topology, LocalDRPC drpc, LocalCluster cluster, String name) {
+ this(drpc, cluster);
+ }
+
+ public void submit(String name, Config config, StormTopology topology) {
+ cluster.submitTopology(name, config, topology);
+ }
+
+ /**
+ * Prints the DRPC results for the amount of time specified
+ */
+ public void printResults(int num, int time, TimeUnit unit) {
+ for (int i = 0; i < num; i++) {
+ try {
+ LOG.info("--- DRPC RESULT: " + drpc.execute("words", "the and apple snow jumped"));
+ System.out.println();
+ Thread.sleep(unit.toMillis(time));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ public void kill (String name) {
+ cluster.killTopology(name);
+ }
+
+ public void shutdown() {
+ cluster.shutdown();
+ }
+
+ public LocalDRPC getDrpc() {
+ return drpc;
+ }
+
+ public LocalCluster getCluster() {
+ return cluster;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaConsumerTopology.java
----------------------------------------------------------------------
diff --git a/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaConsumerTopology.java b/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaConsumerTopology.java
new file mode 100644
index 0000000..b59f973
--- /dev/null
+++ b/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaConsumerTopology.java
@@ -0,0 +1,94 @@
+/*
+ * 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.
+ */
+
+package org.apache.storm.kafka.trident;
+
+import org.apache.storm.Config;
+import org.apache.storm.LocalDRPC;
+import org.apache.storm.StormSubmitter;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.starter.trident.DebugMemoryMapState;
+import org.apache.storm.trident.Stream;
+import org.apache.storm.trident.TridentState;
+import org.apache.storm.trident.TridentTopology;
+import org.apache.storm.trident.operation.builtin.Count;
+import org.apache.storm.trident.operation.builtin.Debug;
+import org.apache.storm.trident.operation.builtin.FilterNull;
+import org.apache.storm.trident.operation.builtin.MapGet;
+import org.apache.storm.trident.spout.ITridentDataSource;
+import org.apache.storm.trident.testing.MemoryMapState;
+import org.apache.storm.trident.testing.Split;
+import org.apache.storm.tuple.Fields;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TridentKafkaConsumerTopology {
+ protected static final Logger LOG = LoggerFactory.getLogger(TridentKafkaConsumerTopology.class);
+
+ public static void submitRemote(String name, ITridentDataSource tridentSpout) {
+ try {
+ StormSubmitter.submitTopology(name, newTpConfig(), newTopology(null, tridentSpout));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * See {@link TridentKafkaConsumerTopology#newTopology(LocalDRPC, ITridentDataSource)}
+ */
+ public static StormTopology newTopology(ITridentDataSource tridentSpout) {
+ return newTopology(null, tridentSpout);
+ }
+
+ /**
+ * @param drpc The DRPC stream to be used in querying the word counts. Can be null in distributed mode
+ * @return a trident topology that consumes sentences from the kafka topic specified using a
+ * {@link TransactionalTridentKafkaSpout} computes the word count and stores it in a {@link MemoryMapState}.
+ */
+ public static StormTopology newTopology(LocalDRPC drpc, ITridentDataSource tridentSpout) {
+ final TridentTopology tridentTopology = new TridentTopology();
+ addDRPCStream(tridentTopology, addTridentState(tridentTopology, tridentSpout), drpc);
+ return tridentTopology.build();
+ }
+
+ private static 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 static TridentState addTridentState(TridentTopology tridentTopology, ITridentDataSource tridentSpout) {
+ final Stream spoutStream = tridentTopology.newStream("spout1", tridentSpout).parallelismHint(1);
+
+ return spoutStream.each(spoutStream.getOutputFields(), new Debug(true))
+ .each(new Fields("str"), new Split(), new Fields("word"))
+ .groupBy(new Fields("word"))
+ .persistentAggregate(new DebugMemoryMapState.Factory(), new Count(), new Fields("count"));
+ }
+
+ private static Config newTpConfig() {
+ Config conf = new Config();
+ conf.setMaxSpoutPending(20);
+ conf.setMaxTaskParallelism(1);
+ return conf;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaTopology.java
----------------------------------------------------------------------
diff --git a/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaTopology.java b/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaTopology.java
new file mode 100644
index 0000000..babb502
--- /dev/null
+++ b/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaTopology.java
@@ -0,0 +1,89 @@
+/*
+ * 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.
+ */
+
+package org.apache.storm.kafka.trident;
+
+import org.apache.storm.Config;
+import org.apache.storm.LocalCluster;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.kafka.trident.mapper.FieldNameBasedTupleToKafkaMapper;
+import org.apache.storm.kafka.trident.selector.DefaultTopicSelector;
+import org.apache.storm.trident.Stream;
+import org.apache.storm.trident.TridentTopology;
+import org.apache.storm.trident.testing.FixedBatchSpout;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+import java.util.Properties;
+
+public class TridentKafkaTopology {
+
+ private static StormTopology buildTopology(String brokerConnectionString) {
+ Fields fields = new Fields("word", "count");
+ FixedBatchSpout spout = new FixedBatchSpout(fields, 4,
+ new Values("storm", "1"),
+ new Values("trident", "1"),
+ new Values("needs", "1"),
+ new Values("javadoc", "1")
+ );
+ spout.setCycle(true);
+
+ TridentTopology topology = new TridentTopology();
+ Stream stream = topology.newStream("spout1", spout);
+
+ Properties props = new Properties();
+ props.put("bootstrap.servers", brokerConnectionString);
+ props.put("acks", "1");
+ props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+ props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+
+ TridentKafkaStateFactory stateFactory = new TridentKafkaStateFactory()
+ .withProducerProperties(props)
+ .withKafkaTopicSelector(new DefaultTopicSelector("test"))
+ .withTridentTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper("word", "count"));
+ stream.partitionPersist(stateFactory, fields, new TridentKafkaUpdater(), new Fields());
+
+ return topology.build();
+ }
+
+ /**
+ * To run this topology ensure you have a kafka broker running and provide connection string to broker as argument.
+ * Create a topic test with command line,
+ * kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partition 1 --topic test
+ *
+ * run this program and run the kafka consumer:
+ * kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
+ *
+ * you should see the messages flowing through.
+ *
+ * @param args
+ * @throws Exception
+ */
+ public static void main(String[] args) throws Exception {
+ if(args.length < 1) {
+ System.out.println("Please provide kafka broker url ,e.g. localhost:9092");
+ }
+
+ LocalCluster cluster = new LocalCluster();
+ cluster.submitTopology("wordCounter", new Config(), buildTopology(args[0]));
+ Thread.sleep(60 * 1000);
+ cluster.killTopology("wordCounter");
+
+ cluster.shutdown();
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaWordCount.java
----------------------------------------------------------------------
diff --git a/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaWordCount.java b/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaWordCount.java
new file mode 100644
index 0000000..0247cce
--- /dev/null
+++ b/examples/storm-kafka-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaWordCount.java
@@ -0,0 +1,140 @@
+/*
+ * 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.
+ */
+package org.apache.storm.kafka.trident;
+
+
+import org.apache.storm.Config;
+import org.apache.storm.StormSubmitter;
+import org.apache.storm.kafka.StringScheme;
+import org.apache.storm.kafka.ZkHosts;
+import org.apache.storm.kafka.bolt.KafkaBolt;
+import org.apache.storm.spout.SchemeAsMultiScheme;
+import org.apache.storm.trident.testing.MemoryMapState;
+
+import java.io.Serializable;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 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 implements Serializable {
+ /**
+ * <p>
+ * To run this topology it is required that you have a kafka broker running.
+ * </p>
+ * Create a topic 'test' with command line,
+ * <pre>
+ * kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partition 1 --topic test
+ * </pre>
+ * To run in local mode,
+ * <pre>
+ * storm jar storm-starter-topologies-{version}.jar org.apache.storm.starter.trident.TridentKafkaWordCount
+ * </pre>
+ * This will also run a local DRPC query and print the word counts.
+ * <p>
+ * To run in distributed mode, run it with a topology name. You will also need to start a drpc server and
+ * specify the drpc server details storm.yaml before submitting the topology.
+ * </p>
+ * <pre>
+ * storm jar storm-starter-topologies-{version}.jar org.apache.storm.starter.trident.TridentKafkaWordCount zkhost:port broker:port wordcount
+ * </pre>
+ * This will submit two topologies, one for the producer and another for the consumer. You can query the results
+ * (word counts) by running an external drpc query against the drpc server.
+ */
+ public static void main(String[] args) throws Exception {
+ final String[] zkBrokerUrl = parseUrl(args);
+ final String topicName = "test";
+ Config tpConf = LocalSubmitter.defaultConfig();
+
+ if (args.length == 3) { //Submit Remote
+ // Producer
+ StormSubmitter.submitTopology(args[2] + "-producer", tpConf, KafkaProducerTopology.newTopology(zkBrokerUrl[1], topicName));
+ // Consumer
+ StormSubmitter.submitTopology(args[2] + "-consumer", tpConf, TridentKafkaConsumerTopology.newTopology(
+ new TransactionalTridentKafkaSpout(newTridentKafkaConfig(zkBrokerUrl[0]))));
+ } else { //Submit Local
+ final LocalSubmitter localSubmitter = LocalSubmitter.newInstance();
+ final String prodTpName = "kafkaBolt";
+ final String consTpName = "wordCounter";
+
+ try {
+ // Producer
+ localSubmitter.submit(prodTpName, tpConf, KafkaProducerTopology.newTopology(zkBrokerUrl[1], topicName));
+ // Consumer
+ localSubmitter.submit(consTpName, tpConf, TridentKafkaConsumerTopology.newTopology(localSubmitter.getDrpc(),
+ new TransactionalTridentKafkaSpout(newTridentKafkaConfig(zkBrokerUrl[0]))));
+
+ // print
+ localSubmitter.printResults(60, 1, TimeUnit.SECONDS);
+ } finally {
+ // kill
+ localSubmitter.kill(prodTpName);
+ localSubmitter.kill(consTpName);
+ // shutdown
+ localSubmitter.shutdown();
+ }
+ }
+ }
+
+ private static String[] parseUrl(String[] args) {
+ String zkUrl = "localhost:2181"; // the defaults.
+ String brokerUrl = "localhost:9092";
+
+ if (args.length > 3 || (args.length == 1 && args[0].matches("^-h|--help$"))) {
+ System.out.println("Usage: TridentKafkaWordCount [kafka zookeeper url] [kafka broker url] [topology name]");
+ System.out.println(" E.g TridentKafkaWordCount [" + zkUrl + "]" + " [" + brokerUrl + "] [wordcount]");
+ 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 uHrl: " + zkUrl + " broker url: " + brokerUrl);
+ return new String[]{zkUrl, brokerUrl};
+ }
+
+ private static TridentKafkaConfig newTridentKafkaConfig(String zkUrl) {
+ 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 config;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-starter/pom.xml
----------------------------------------------------------------------
diff --git a/examples/storm-starter/pom.xml b/examples/storm-starter/pom.xml
index 0ee02c4..df443de 100644
--- a/examples/storm-starter/pom.xml
+++ b/examples/storm-starter/pom.xml
@@ -24,7 +24,6 @@
<relativePath>../../pom.xml</relativePath>
</parent>
- <groupId>org.apache.storm</groupId>
<artifactId>storm-starter</artifactId>
<packaging>jar</packaging>
@@ -142,18 +141,6 @@
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
- <artifactId>storm-kafka</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.storm</groupId>
- <artifactId>storm-kafka-client</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.storm</groupId>
<artifactId>storm-hdfs</artifactId>
<version>${project.version}</version>
</dependency>
@@ -163,15 +150,6 @@
<version>${project.version}</version>
</dependency>
<dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>${storm.kafka.artifact.id}</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka-clients</artifactId>
- <version>${storm.kafka.client.version}</version>
- </dependency>
- <dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-redis</artifactId>
<version>${project.version}</version>
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaClientWordCountNamedTopics.java
----------------------------------------------------------------------
diff --git a/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaClientWordCountNamedTopics.java b/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaClientWordCountNamedTopics.java
deleted file mode 100644
index 44af184..0000000
--- a/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaClientWordCountNamedTopics.java
+++ /dev/null
@@ -1,122 +0,0 @@
-/*
- * 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.
- */
-
-package org.apache.storm.starter.trident;
-
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.storm.kafka.spout.KafkaSpoutConfig;
-import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff;
-import org.apache.storm.kafka.spout.KafkaSpoutRetryService;
-import org.apache.storm.kafka.spout.KafkaSpoutStreams;
-import org.apache.storm.kafka.spout.KafkaSpoutStreamsNamedTopics;
-import org.apache.storm.kafka.spout.KafkaSpoutTupleBuilder;
-import org.apache.storm.kafka.spout.KafkaSpoutTuplesBuilder;
-import org.apache.storm.kafka.spout.KafkaSpoutTuplesBuilderNamedTopics;
-import org.apache.storm.kafka.spout.trident.KafkaTridentSpoutManager;
-import org.apache.storm.kafka.spout.trident.KafkaTridentSpoutOpaque;
-import org.apache.storm.trident.Stream;
-import org.apache.storm.trident.TridentState;
-import org.apache.storm.trident.TridentTopology;
-import org.apache.storm.trident.operation.builtin.Count;
-import org.apache.storm.trident.operation.builtin.Debug;
-import org.apache.storm.trident.testing.Split;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Values;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
-import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST;
-
-public class TridentKafkaClientWordCountNamedTopics extends TridentKafkaWordCount {
- public TridentKafkaClientWordCountNamedTopics(String zkUrl, String brokerUrl) {
- super(zkUrl, brokerUrl);
- }
-
- protected TridentState addTridentState(TridentTopology tridentTopology) {
- final Stream spoutStream = tridentTopology.newStream("spout1", createOpaqueKafkaSpoutNew()).parallelismHint(1);
-
- return spoutStream.each(spoutStream.getOutputFields(), new Debug(true))
- .each(new Fields("str"), new Split(), new Fields("word"))
- .groupBy(new Fields("word"))
- .persistentAggregate(new DebugMemoryMapState.Factory(), new Count(), new Fields("count"));
- }
-
- private KafkaTridentSpoutOpaque<String, String> createOpaqueKafkaSpoutNew() {
- return new KafkaTridentSpoutOpaque<String, String>(getKafkaTridentManager());
- }
-
- private KafkaTridentSpoutManager<String, String> getKafkaTridentManager() {
- return new KafkaTridentSpoutManager<>(getKafkaSpoutConfig(getKafkaSpoutStreams()));
- }
-
- private KafkaSpoutConfig<String,String> getKafkaSpoutConfig(KafkaSpoutStreams kafkaSpoutStreams) {
- return new KafkaSpoutConfig.Builder<String, String>(getKafkaConsumerProps(), kafkaSpoutStreams, getTuplesBuilder(), getRetryService())
- .setOffsetCommitPeriodMs(10_000)
- .setFirstPollOffsetStrategy(EARLIEST)
- .setMaxUncommittedOffsets(250)
- .build();
- }
-
- protected Map<String,Object> getKafkaConsumerProps() {
- Map<String, Object> props = new HashMap<>();
- props.put(KafkaSpoutConfig.Consumer.BOOTSTRAP_SERVERS, "127.0.0.1:9092");
- props.put(KafkaSpoutConfig.Consumer.GROUP_ID, "kafkaSpoutTestGroup");
- props.put(KafkaSpoutConfig.Consumer.KEY_DESERIALIZER, "org.apache.kafka.common.serialization.StringDeserializer");
- props.put(KafkaSpoutConfig.Consumer.VALUE_DESERIALIZER, "org.apache.kafka.common.serialization.StringDeserializer");
- props.put("max.partition.fetch.bytes", 200);
- return props;
- }
-
- protected KafkaSpoutTuplesBuilder<String, String> getTuplesBuilder() {
- return new KafkaSpoutTuplesBuilderNamedTopics.Builder<>(
- new TopicsTupleBuilder<String, String>("test-trident","test-trident-1"))
- .build();
- }
-
- protected KafkaSpoutRetryService getRetryService() {
- return new KafkaSpoutRetryExponentialBackoff(getTimeInterval(500, TimeUnit.MICROSECONDS),
- KafkaSpoutRetryExponentialBackoff.TimeInterval.milliSeconds(2), Integer.MAX_VALUE, KafkaSpoutRetryExponentialBackoff.TimeInterval.seconds(10));
- }
-
- private static KafkaSpoutRetryExponentialBackoff.TimeInterval getTimeInterval(long delay, TimeUnit timeUnit) {
- return new KafkaSpoutRetryExponentialBackoff.TimeInterval(delay, timeUnit);
- }
-
- protected KafkaSpoutStreams getKafkaSpoutStreams() {
- final Fields outputFields = new Fields("str");
- return new KafkaSpoutStreamsNamedTopics.Builder(outputFields, new String[]{"test-trident","test-trident-1"}).build();
- }
-
- protected static class TopicsTupleBuilder<K, V> extends KafkaSpoutTupleBuilder<K,V> {
- public TopicsTupleBuilder(String... topics) {
- super(topics);
- }
- @Override
- public List<Object> buildTuple(ConsumerRecord<K, V> consumerRecord) {
- return new Values(consumerRecord.value());
- }
- }
-
- public static void main(String[] args) throws Exception {
- final String[] zkBrokerUrl = parseUrl(args);
- runMain(args, new TridentKafkaClientWordCountNamedTopics(zkBrokerUrl[0], zkBrokerUrl[1]));
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaClientWordCountWildcardTopics.java
----------------------------------------------------------------------
diff --git a/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaClientWordCountWildcardTopics.java b/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaClientWordCountWildcardTopics.java
deleted file mode 100644
index 018b38d..0000000
--- a/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaClientWordCountWildcardTopics.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * 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.
- */
-
-package org.apache.storm.starter.trident;
-
-import org.apache.storm.kafka.spout.KafkaSpoutStream;
-import org.apache.storm.kafka.spout.KafkaSpoutStreams;
-import org.apache.storm.kafka.spout.KafkaSpoutStreamsWildcardTopics;
-import org.apache.storm.kafka.spout.KafkaSpoutTuplesBuilder;
-import org.apache.storm.kafka.spout.KafkaSpoutTuplesBuilderWildcardTopics;
-import org.apache.storm.tuple.Fields;
-
-import java.util.regex.Pattern;
-
-public class TridentKafkaClientWordCountWildcardTopics extends TridentKafkaClientWordCountNamedTopics {
- private static final String TOPIC_WILDCARD_PATTERN = "test-trident(-1)?";
-
- public TridentKafkaClientWordCountWildcardTopics(String zkUrl, String brokerUrl) {
- super(zkUrl, brokerUrl);
- }
-
- public static void main(String[] args) throws Exception {
- final String[] zkBrokerUrl = parseUrl(args);
- runMain(args, new TridentKafkaClientWordCountWildcardTopics(zkBrokerUrl[0], zkBrokerUrl[1]));
- }
-
- protected KafkaSpoutTuplesBuilder<String, String> getTuplesBuilder() {
- return new KafkaSpoutTuplesBuilderWildcardTopics<>(new TopicsTupleBuilder<>(TOPIC_WILDCARD_PATTERN));
- }
-
- protected KafkaSpoutStreams getKafkaSpoutStreams() {
- final Fields outputFields = new Fields("str");
- final KafkaSpoutStream kafkaSpoutStream = new KafkaSpoutStream(outputFields, Pattern.compile(TOPIC_WILDCARD_PATTERN));
- return new KafkaSpoutStreamsWildcardTopics(kafkaSpoutStream);
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaWordCount.java
----------------------------------------------------------------------
diff --git a/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaWordCount.java b/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaWordCount.java
deleted file mode 100644
index 6b35d1b..0000000
--- a/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaWordCount.java
+++ /dev/null
@@ -1,282 +0,0 @@
-/*
- * 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 org.apache.storm.starter.trident;
-
-
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.storm.Config;
-import org.apache.storm.LocalCluster;
-import org.apache.storm.LocalDRPC;
-import org.apache.storm.StormSubmitter;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.kafka.StringScheme;
-import org.apache.storm.kafka.ZkHosts;
-import org.apache.storm.kafka.bolt.KafkaBolt;
-import org.apache.storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaMapper;
-import org.apache.storm.kafka.bolt.selector.DefaultTopicSelector;
-import org.apache.storm.kafka.trident.OpaqueTridentKafkaSpout;
-import org.apache.storm.kafka.trident.TransactionalTridentKafkaSpout;
-import org.apache.storm.kafka.trident.TridentKafkaConfig;
-import org.apache.storm.spout.SchemeAsMultiScheme;
-import org.apache.storm.starter.spout.RandomSentenceSpout;
-import org.apache.storm.topology.TopologyBuilder;
-import org.apache.storm.trident.Stream;
-import org.apache.storm.trident.TridentState;
-import org.apache.storm.trident.TridentTopology;
-import org.apache.storm.trident.operation.builtin.Count;
-import org.apache.storm.trident.operation.builtin.Debug;
-import org.apache.storm.trident.operation.builtin.FilterNull;
-import org.apache.storm.trident.operation.builtin.MapGet;
-import org.apache.storm.trident.testing.MemoryMapState;
-import org.apache.storm.trident.testing.Split;
-import org.apache.storm.tuple.Fields;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.Serializable;
-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 implements Serializable {
- private static final Logger LOG = LoggerFactory.getLogger(TridentKafkaWordCount.class);
-
- 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 createTransactionalKafkaSpout() {
- 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 OpaqueTridentKafkaSpout createOpaqueKafkaSpout() {
- 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 OpaqueTridentKafkaSpout(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"));
- }
-
- protected TridentState addTridentState(TridentTopology tridentTopology) {
-// final Stream spoutStream = tridentTopology.newStream("spout1", createTransactionalKafkaSpout()).parallelismHint(1);
- final Stream spoutStream = tridentTopology.newStream("spout1", createOpaqueKafkaSpout()).parallelismHint(1);
-
- return spoutStream.each(spoutStream.getOutputFields(), new Debug(true))
- .each(new Fields("str"), new Split(), new Fields("word"))
- .groupBy(new Fields("word"))
- .persistentAggregate(new DebugMemoryMapState.Factory(), new Count(), new Fields("count"));
- }
-
- /**
- * 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.setMaxTaskParallelism(1);
- // 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(Properties prop) {
- TopologyBuilder builder = new TopologyBuilder();
- builder.setSpout("spout", new RandomSentenceSpout.TimeStamped(""), 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().withProducerProperties(prop)
- .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 Properties getProducerConfig() {
- 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");
- return props;
- }
-
- /**
- * <p>
- * To run this topology ensure you have a kafka broker running.
- * </p>
- * Create a topic 'test' with command line,
- * <pre>
- * kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partition 1 --topic test
- * </pre>
- * To run in local mode,
- * <pre>
- * storm jar storm-starter-topologies-{version}.jar org.apache.storm.starter.trident.TridentKafkaWordCount
- * </pre>
- * This will also run a local DRPC query and print the word counts.
- * <p>
- * To run in distributed mode, run it with a topology name. You will also need to start a drpc server and
- * specify the drpc server details storm.yaml before submitting the topology.
- * </p>
- * <pre>
- * storm jar storm-starter-topologies-{version}.jar org.apache.storm.starter.trident.TridentKafkaWordCount zkhost:port broker:port wordcount
- * </pre>
- * This will submit two topologies, one for the producer and another for the consumer. You can query the results
- * (word counts) by running an external drpc query against the drpc server.
- */
- public static void main(String[] args) throws Exception {
- final String[] zkBrokerUrl = parseUrl(args);
- runMain(args, new TridentKafkaWordCount(zkBrokerUrl[0], zkBrokerUrl[1]));
- }
-
- protected static String[] parseUrl(String[] args) {
- String zkUrl = "localhost:2181"; // the defaults.
- String brokerUrl = "localhost:9092";
-
- if (args.length > 3 || (args.length == 1 && args[0].matches("^-h|--help$"))) {
- System.out.println("Usage: TridentKafkaWordCount [kafka zookeeper url] [kafka broker url] [topology name]");
- System.out.println(" E.g TridentKafkaWordCount [" + zkUrl + "]" + " [" + brokerUrl + "] [wordcount]");
- 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);
- return new String[]{zkUrl, brokerUrl};
- }
-
- protected static void runMain(String[] args, TridentKafkaWordCount wordCount) throws Exception {
- if (args.length == 3) {
- Config conf = new Config();
- conf.setMaxSpoutPending(20);
- conf.setNumWorkers(1);
- // submit the CONSUMER topology.
- StormSubmitter.submitTopology(args[2] + "-consumer", conf, wordCount.buildConsumerTopology(null));
- // submit the producer topology.
- StormSubmitter.submitTopology(args[2] + "-producer", conf, wordCount.buildProducerTopology(wordCount.getProducerConfig()));
- } else {
- LocalDRPC drpc = new LocalDRPC();
- LocalCluster cluster = new LocalCluster();
-
- // submit the CONSUMER topology.
- cluster.submitTopology("wordCounter", wordCount.getConsumerConfig(), wordCount.buildConsumerTopology(drpc));
-
- // submit the PRODUCER topology.
- Config conf = new Config();
- conf.setMaxSpoutPending(20);
-// conf.setDebug(true);
- cluster.submitTopology("kafkaBolt", conf, wordCount.buildProducerTopology(wordCount.getProducerConfig()));
-
- // keep querying the word counts for a minute.
- for (int i = 0; i < 60; i++) {
- LOG.info("--- DRPC RESULT: " + drpc.execute("words", "the and apple snow jumped"));
- System.out.println();
- Thread.sleep(1000);
- }
-
- cluster.killTopology("kafkaBolt");
- cluster.killTopology("wordCounter");
- cluster.shutdown();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/OpaqueTridentKafkaSpout.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/OpaqueTridentKafkaSpout.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/OpaqueTridentKafkaSpout.java
index f540c87..71324f7 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/OpaqueTridentKafkaSpout.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/OpaqueTridentKafkaSpout.java
@@ -17,14 +17,13 @@
*/
package org.apache.storm.kafka.trident;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.tuple.Fields;
import org.apache.storm.kafka.Partition;
+import org.apache.storm.task.TopologyContext;
import org.apache.storm.trident.spout.IOpaquePartitionedTridentSpout;
+import org.apache.storm.tuple.Fields;
import java.util.List;
import java.util.Map;
-import java.util.UUID;
public class OpaqueTridentKafkaSpout implements IOpaquePartitionedTridentSpout<List<GlobalPartitionInformation>, Partition, Map> {
http://git-wip-us.apache.org/repos/asf/storm/blob/4d22a68a/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index c7fc3a3..0a6b83a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -332,6 +332,7 @@
<module>examples/storm-opentsdb-examples</module>
<module>examples/storm-solr-examples</module>
<module>examples/storm-kafka-examples</module>
+ <module>examples/storm-kafka-client-examples</module>
<module>examples/storm-jdbc-examples</module>
<module>examples/storm-hdfs-examples</module>
<module>examples/storm-hbase-examples</module>