You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by srdo <gi...@git.apache.org> on 2017/07/25 17:55:11 UTC
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
GitHub user srdo opened a pull request:
https://github.com/apache/storm/pull/2243
STORM-2658: Extract storm-kafka-client examples to storm-kafka-client…
…-examples, make storm-kafka-client-examples generate a jar with all dependencies.
See https://issues.apache.org/jira/browse/STORM-2658.
The changes to the Trident examples mainly have to do with the command line arguments not being consistently passed to the topologies, e.g. the broker url was passed to producing topologies but not the consuming topology. The extra parameters have been removed, I think the example code is fine with hard coded topic names.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/srdo/storm STORM-2658
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/2243.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2243
----
commit 8621b7071783a714099ca409e667d7e49fd88265
Author: Stig Rohde Døssing <sr...@apache.org>
Date: 2017-07-25T15:33:37Z
STORM-2658: Extract storm-kafka-client examples to storm-kafka-client-examples, make storm-kafka-client-examples generate a jar with all dependencies
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130438441
--- Diff: external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/builders/SingleTopicKafkaSpoutConfiguration.java ---
@@ -21,15 +21,10 @@
import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST;
import org.apache.kafka.clients.consumer.ConsumerConfig;
--- End diff --
Sure, will rename
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130425918
--- Diff: external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/builders/SingleTopicKafkaSpoutConfiguration.java ---
@@ -21,15 +21,10 @@
import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST;
import org.apache.kafka.clients.consumer.ConsumerConfig;
--- End diff --
I would call package where this class lives `config.builder` instead of `builders`, which is a bit misleading since this is really a configuration class.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130438296
--- Diff: examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountNamedTopics.java ---
@@ -42,72 +42,68 @@
import org.apache.storm.tuple.Values;
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<>(newKafkaSpoutConfig());
+ private KafkaTridentSpoutOpaque<String, String> newKafkaTridentSpoutOpaque(KafkaSpoutConfig<String, String> spoutConfig) {
+ return new KafkaTridentSpoutOpaque<>(spoutConfig);
}
private static Func<ConsumerRecord<String, String>, List<Object>> JUST_VALUE_FUNC = new JustValueFunc();
/**
- * Needs to be serializable
+ * Needs to be serializable.
*/
private static class JustValueFunc implements Func<ConsumerRecord<String, String>, List<Object>>, Serializable {
+
@Override
public List<Object> apply(ConsumerRecord<String, String> record) {
return new Values(record.value());
}
}
- protected KafkaSpoutConfig<String,String> newKafkaSpoutConfig() {
- return KafkaSpoutConfig.builder(KAFKA_LOCAL_BROKER, TOPIC_1, TOPIC_2)
- .setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup_" + System.nanoTime())
- .setProp(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 200)
- .setRecordTranslator(JUST_VALUE_FUNC, new Fields("str"))
- .setRetry(newRetryService())
- .setOffsetCommitPeriodMs(10_000)
- .setFirstPollOffsetStrategy(EARLIEST)
- .setMaxUncommittedOffsets(250)
- .build();
+ protected KafkaSpoutConfig<String, String> newKafkaSpoutConfig(String bootstrapServers) {
+ return KafkaSpoutConfig.builder(bootstrapServers, TOPIC_1, TOPIC_2)
+ .setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup_" + System.nanoTime())
+ .setProp(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 200)
+ .setRecordTranslator(JUST_VALUE_FUNC, new Fields("str"))
+ .setRetry(newRetryService())
+ .setOffsetCommitPeriodMs(10_000)
+ .setFirstPollOffsetStrategy(EARLIEST)
+ .setMaxUncommittedOffsets(250)
+ .build();
}
protected KafkaSpoutRetryService newRetryService() {
return new KafkaSpoutRetryExponentialBackoff(new TimeInterval(500L, TimeUnit.MICROSECONDS),
- TimeInterval.milliSeconds(2), Integer.MAX_VALUE, TimeInterval.seconds(10));
+ TimeInterval.milliSeconds(2), Integer.MAX_VALUE, TimeInterval.seconds(10));
}
public static void main(String[] args) throws Exception {
new TridentKafkaClientWordCountNamedTopics().run(args);
}
- protected void run(String[] args) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException, InterruptedException {
- 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 = new Config();
- tpConf.setDebug(true);
- tpConf.setMaxSpoutPending(5);
-
- // 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()));
-
- // Print results to console, which also causes the print filter in the consumer topology to print the results in the worker log
- Thread.sleep(2000);
- DrpcResultsPrinter.remoteClient().printResults(60, 1, TimeUnit.SECONDS);
- }
+ protected void run(String[] args) throws AlreadyAliveException, InvalidTopologyException,
--- End diff --
The topic name wasn't being passed to the consumer before, only the producers as far as I could tell, so if you used the parameters the example didn't work. Fixing it caused a conflict with the wildcard example, because I'd have to change the newKafkaSpoutConfig signature to take a list of topics. That won't work with the Pattern required by the wildcard example. It seemed easier to just remove the option.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130255074
--- Diff: examples/storm-kafka-client-examples/pom.xml ---
@@ -73,19 +75,27 @@
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka-client</artifactId>
<version>${project.version}</version>
+ <!-- You can reduce jar size by uncommenting this and putting dependencies in $STORM-HOME/extlib instead of including them in the jar
--- End diff --
Same here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130417840
--- Diff: examples/storm-kafka-client-examples/README.markdown ---
@@ -0,0 +1,10 @@
+## Usage
+This module contains example topologies demonstrating storm-kafka-client spout and Trident usage.
+
+The module is built by `mvn clean package`. This will generate the `target/storm-kafka-client-examples-VERSION.jar` file. The jar contains all dependencies and can be submitted to Storm via the Storm CLI. For example:
--- End diff --
... built running ... the Storm CLI, e.g.:
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2243: STORM-2658: Extract storm-kafka-client examples to storm-...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on the issue:
https://github.com/apache/storm/pull/2243
+1. Thanks @srdo
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130431738
--- Diff: examples/storm-kafka-client-examples/pom.xml ---
@@ -73,19 +75,27 @@
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka-client</artifactId>
<version>${project.version}</version>
+ <!-- You can reduce jar size by uncommenting this and providing the dependencies manually. See the README for details.
<scope>${provided.scope}</scope>
+ -->
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>${storm.kafka.artifact.id}</artifactId>
<version>${storm.kafka.client.version}</version>
+ <scope>compile</scope>
+ <!-- You can reduce jar size by uncommenting this and providing the dependencies manually. See the README for details.
<scope>${provided.scope}</scope>
+ -->
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${storm.kafka.client.version}</version>
+ <scope>compile</scope>
+ <!-- You can reduce jar size by uncommenting this and providing the dependencies manually. See the README for details.
--- End diff --
Isn't the goal of ${provided.scope} to handle the proper scope according to the profile, e.g. just like it is done with the Intellij profile. I am not quite following why the user has to comment/uncomment the scope configuration.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130255082
--- Diff: examples/storm-kafka-client-examples/pom.xml ---
@@ -73,19 +75,27 @@
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka-client</artifactId>
<version>${project.version}</version>
+ <!-- You can reduce jar size by uncommenting this and putting dependencies in $STORM-HOME/extlib instead of including them in the jar
<scope>${provided.scope}</scope>
+ -->
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>${storm.kafka.artifact.id}</artifactId>
<version>${storm.kafka.client.version}</version>
+ <scope>compile</scope>
+ <!-- You can reduce jar size by uncommenting this and putting dependencies in $STORM-HOME/extlib instead of including them in the jar
--- End diff --
Same here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2243: STORM-2658: Extract storm-kafka-client examples to storm-...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on the issue:
https://github.com/apache/storm/pull/2243
@srdo
I have no idea which branches I need to apply this patch, so please go on merging yourself. Thanks for the patch. :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130422627
--- Diff: examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/trident/TridentKafkaClientWordCountNamedTopics.java ---
@@ -42,72 +42,68 @@
import org.apache.storm.tuple.Values;
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<>(newKafkaSpoutConfig());
+ private KafkaTridentSpoutOpaque<String, String> newKafkaTridentSpoutOpaque(KafkaSpoutConfig<String, String> spoutConfig) {
+ return new KafkaTridentSpoutOpaque<>(spoutConfig);
}
private static Func<ConsumerRecord<String, String>, List<Object>> JUST_VALUE_FUNC = new JustValueFunc();
/**
- * Needs to be serializable
+ * Needs to be serializable.
*/
private static class JustValueFunc implements Func<ConsumerRecord<String, String>, List<Object>>, Serializable {
+
@Override
public List<Object> apply(ConsumerRecord<String, String> record) {
return new Values(record.value());
}
}
- protected KafkaSpoutConfig<String,String> newKafkaSpoutConfig() {
- return KafkaSpoutConfig.builder(KAFKA_LOCAL_BROKER, TOPIC_1, TOPIC_2)
- .setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup_" + System.nanoTime())
- .setProp(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 200)
- .setRecordTranslator(JUST_VALUE_FUNC, new Fields("str"))
- .setRetry(newRetryService())
- .setOffsetCommitPeriodMs(10_000)
- .setFirstPollOffsetStrategy(EARLIEST)
- .setMaxUncommittedOffsets(250)
- .build();
+ protected KafkaSpoutConfig<String, String> newKafkaSpoutConfig(String bootstrapServers) {
+ return KafkaSpoutConfig.builder(bootstrapServers, TOPIC_1, TOPIC_2)
+ .setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup_" + System.nanoTime())
+ .setProp(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 200)
+ .setRecordTranslator(JUST_VALUE_FUNC, new Fields("str"))
+ .setRetry(newRetryService())
+ .setOffsetCommitPeriodMs(10_000)
+ .setFirstPollOffsetStrategy(EARLIEST)
+ .setMaxUncommittedOffsets(250)
+ .build();
}
protected KafkaSpoutRetryService newRetryService() {
return new KafkaSpoutRetryExponentialBackoff(new TimeInterval(500L, TimeUnit.MICROSECONDS),
- TimeInterval.milliSeconds(2), Integer.MAX_VALUE, TimeInterval.seconds(10));
+ TimeInterval.milliSeconds(2), Integer.MAX_VALUE, TimeInterval.seconds(10));
}
public static void main(String[] args) throws Exception {
new TridentKafkaClientWordCountNamedTopics().run(args);
}
- protected void run(String[] args) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException, InterruptedException {
- 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 = new Config();
- tpConf.setDebug(true);
- tpConf.setMaxSpoutPending(5);
-
- // 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()));
-
- // Print results to console, which also causes the print filter in the consumer topology to print the results in the worker log
- Thread.sleep(2000);
- DrpcResultsPrinter.remoteClient().printResults(60, 1, TimeUnit.SECONDS);
- }
+ protected void run(String[] args) throws AlreadyAliveException, InvalidTopologyException,
--- End diff --
why remove the ability to specify the topic name from the command line ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2243: STORM-2658: Extract storm-kafka-client examples to storm-...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2243
Will do, thanks @HeartSaVioR :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130435628
--- Diff: examples/storm-kafka-client-examples/README.markdown ---
@@ -0,0 +1,10 @@
+## Usage
+This module contains example topologies demonstrating storm-kafka-client spout and Trident usage.
+
+The module is built by `mvn clean package`. This will generate the `target/storm-kafka-client-examples-VERSION.jar` file. The jar contains all dependencies and can be submitted to Storm via the Storm CLI. For example:
--- End diff --
Will fix
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2243: STORM-2658: Extract storm-kafka-client examples to storm-...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on the issue:
https://github.com/apache/storm/pull/2243
+1
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130418867
--- Diff: examples/storm-kafka-client-examples/pom.xml ---
@@ -42,7 +42,9 @@
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka</artifactId>
<version>${project.version}</version>
+ <!-- You can reduce jar size by uncommenting this and providing the dependencies manually. See the README for details.
<scope>${provided.scope}</scope>
--- End diff --
Isn't the goal of ${provided.scope} to handle the proper scope according to the profile, e.g. just like it is done with the Intellij profile. I am not quite following why the user has to comment/uncomment the scope configuration.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130429805
--- Diff: external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/builders/SingleTopicKafkaSpoutConfiguration.java ---
@@ -21,15 +21,10 @@
import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST;
import org.apache.kafka.clients.consumer.ConsumerConfig;
--- End diff --
I also would call the two getXyz methods in this class createXyz, as they are static factory methods. I know that the name was already like that, but since we are changing it, we should just make it more conventional.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130419415
--- Diff: examples/storm-kafka-client-examples/pom.xml ---
@@ -73,19 +75,27 @@
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka-client</artifactId>
<version>${project.version}</version>
+ <!-- You can reduce jar size by uncommenting this and providing the dependencies manually. See the README for details.
<scope>${provided.scope}</scope>
--- End diff --
Isn't the goal of ${provided.scope} to handle the proper scope according to the profile, e.g. just like it is done with the Intellij profile. I am not quite following why the user has to comment/uncomment the scope configuration.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/2243
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130254961
--- Diff: examples/storm-kafka-client-examples/README.markdown ---
@@ -0,0 +1,10 @@
+## Usage
+This module contains example topologies demonstrating storm-kafka-client spout and Trident usage.
+
+The module is built by `mvn clean package`. This will generate the `target/storm-kafka-client-examples-VERSION.jar` file. The jar contains all dependencies and can be submitted to Storm via the Storm CLI. For example:
+```
+storm jar storm-kafka-client-examples-2.0.0-SNAPSHOT.jar org.apache.storm.kafka.spout.test.KafkaSpoutTopologyMainNamedTopics
+```
+will submit the topologies set up by KafkaSpoutTopologyMainNamedTopics to Storm.
+
+Note that this example produces a jar containing all dependencies for ease of use. In a production environment you may want to reduce the jar size by extracting some dependencies (e.g. org.apache.kafka:kafka-clients) from the jar. You can do this by setting the dependencies you don't want to include in the jars to `provided` scope, and then manually copying the dependencies to your Storm extlib directory.
--- End diff --
Instead of copying dependencies to the extlib, you can achieve the same thing (or more) via using `--artifacts` to add dependencies for specific topology while submitting. I think this is simpler and topology-wide, so would love to guide both, or only `--artifacts`. (We already replaced the guide for how to add dependencies from Storm SQL.)
Please refer https://github.com/apache/storm/blob/master/docs/Command-line-client.md#jar for details.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130428532
--- Diff: examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/spout/test/KafkaSpoutTopologyMainNamedTopics.java ---
@@ -0,0 +1,103 @@
+/*
+ * 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.spout.test;
+
+import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.storm.Config;
+import org.apache.storm.StormSubmitter;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.kafka.spout.ByTopicRecordTranslator;
+import org.apache.storm.kafka.spout.KafkaSpout;
+import org.apache.storm.kafka.spout.KafkaSpoutConfig;
+import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff;
+import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff.TimeInterval;
+import org.apache.storm.kafka.spout.KafkaSpoutRetryService;
+import org.apache.storm.kafka.trident.KafkaProducerTopology;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+public class KafkaSpoutTopologyMainNamedTopics {
+
+ private static final String TOPIC_2_STREAM = "test_2_stream";
+ private static final String TOPIC_0_1_STREAM = "test_0_1_stream";
+ private static final String KAFKA_LOCAL_BROKER = "localhost:9092";
+ private static final String TOPIC_0 = "kafka-spout-test";
+ private static final String TOPIC_1 = "kafka-spout-test-1";
+ private static final String TOPIC_2 = "kafka-spout-test-2";
+
+ public static void main(String[] args) throws Exception {
+ new KafkaSpoutTopologyMainNamedTopics().runMain(args);
+ }
+
+ protected void runMain(String[] args) throws Exception {
--- End diff --
Isn't this change removing the ability to run this code in LocalCluster mode? I think it is very useful. For example, I use it all the time to run these simple test examples from IntelliJ.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130435817
--- Diff: examples/storm-kafka-client-examples/pom.xml ---
@@ -42,7 +42,9 @@
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka</artifactId>
<version>${project.version}</version>
+ <!-- You can reduce jar size by uncommenting this and providing the dependencies manually. See the README for details.
<scope>${provided.scope}</scope>
--- End diff --
Yes, I somehow missed the dollar. I'll try reverting this and update the readme to set the scope to compile
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130431776
--- Diff: examples/storm-kafka-client-examples/pom.xml ---
@@ -73,19 +75,27 @@
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka-client</artifactId>
<version>${project.version}</version>
+ <!-- You can reduce jar size by uncommenting this and providing the dependencies manually. See the README for details.
<scope>${provided.scope}</scope>
+ -->
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>${storm.kafka.artifact.id}</artifactId>
<version>${storm.kafka.client.version}</version>
+ <scope>compile</scope>
+ <!-- You can reduce jar size by uncommenting this and providing the dependencies manually. See the README for details.
--- End diff --
Isn't the goal of ${provided.scope} to handle the proper scope according to the profile, e.g. just like it is done with the Intellij profile. I am not quite following why the user has to comment/uncomment the scope configuration.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130449657
--- Diff: examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/spout/test/KafkaSpoutTopologyMainNamedTopics.java ---
@@ -0,0 +1,103 @@
+/*
+ * 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.spout.test;
+
+import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.storm.Config;
+import org.apache.storm.StormSubmitter;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.kafka.spout.ByTopicRecordTranslator;
+import org.apache.storm.kafka.spout.KafkaSpout;
+import org.apache.storm.kafka.spout.KafkaSpoutConfig;
+import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff;
+import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff.TimeInterval;
+import org.apache.storm.kafka.spout.KafkaSpoutRetryService;
+import org.apache.storm.kafka.trident.KafkaProducerTopology;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+public class KafkaSpoutTopologyMainNamedTopics {
+
+ private static final String TOPIC_2_STREAM = "test_2_stream";
+ private static final String TOPIC_0_1_STREAM = "test_0_1_stream";
+ private static final String KAFKA_LOCAL_BROKER = "localhost:9092";
+ private static final String TOPIC_0 = "kafka-spout-test";
+ private static final String TOPIC_1 = "kafka-spout-test-1";
+ private static final String TOPIC_2 = "kafka-spout-test-2";
+
+ public static void main(String[] args) throws Exception {
+ new KafkaSpoutTopologyMainNamedTopics().runMain(args);
+ }
+
+ protected void runMain(String[] args) throws Exception {
--- End diff --
I remembered why I removed it. LocalCluster is in the storm-server jar, which isn't included by the example projects. I think including it would cause conflict when the jar is deployed to a real cluster. How about I move the ability to run this from a local cluster to a test class? That should still leave people able to run on a local cluster from an IDE, but doesn't interfere with the generated jar.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130255089
--- Diff: examples/storm-kafka-client-examples/pom.xml ---
@@ -73,19 +75,27 @@
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka-client</artifactId>
<version>${project.version}</version>
+ <!-- You can reduce jar size by uncommenting this and putting dependencies in $STORM-HOME/extlib instead of including them in the jar
<scope>${provided.scope}</scope>
+ -->
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>${storm.kafka.artifact.id}</artifactId>
<version>${storm.kafka.client.version}</version>
+ <scope>compile</scope>
+ <!-- You can reduce jar size by uncommenting this and putting dependencies in $STORM-HOME/extlib instead of including them in the jar
<scope>${provided.scope}</scope>
+ -->
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${storm.kafka.client.version}</version>
+ <scope>compile</scope>
+ <!-- You can reduce jar size by uncommenting this and putting dependencies in $STORM-HOME/extlib instead of including them in the jar
--- End diff --
Same here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130438812
--- Diff: examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/spout/test/KafkaSpoutTopologyMainNamedTopics.java ---
@@ -0,0 +1,103 @@
+/*
+ * 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.spout.test;
+
+import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.storm.Config;
+import org.apache.storm.StormSubmitter;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.kafka.spout.ByTopicRecordTranslator;
+import org.apache.storm.kafka.spout.KafkaSpout;
+import org.apache.storm.kafka.spout.KafkaSpoutConfig;
+import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff;
+import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff.TimeInterval;
+import org.apache.storm.kafka.spout.KafkaSpoutRetryService;
+import org.apache.storm.kafka.trident.KafkaProducerTopology;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+public class KafkaSpoutTopologyMainNamedTopics {
+
+ private static final String TOPIC_2_STREAM = "test_2_stream";
+ private static final String TOPIC_0_1_STREAM = "test_0_1_stream";
+ private static final String KAFKA_LOCAL_BROKER = "localhost:9092";
+ private static final String TOPIC_0 = "kafka-spout-test";
+ private static final String TOPIC_1 = "kafka-spout-test-1";
+ private static final String TOPIC_2 = "kafka-spout-test-2";
+
+ public static void main(String[] args) throws Exception {
+ new KafkaSpoutTopologyMainNamedTopics().runMain(args);
+ }
+
+ protected void runMain(String[] args) throws Exception {
--- End diff --
Yes. I'll restore that bit
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130352763
--- Diff: examples/storm-kafka-client-examples/README.markdown ---
@@ -0,0 +1,10 @@
+## Usage
+This module contains example topologies demonstrating storm-kafka-client spout and Trident usage.
+
+The module is built by `mvn clean package`. This will generate the `target/storm-kafka-client-examples-VERSION.jar` file. The jar contains all dependencies and can be submitted to Storm via the Storm CLI. For example:
+```
+storm jar storm-kafka-client-examples-2.0.0-SNAPSHOT.jar org.apache.storm.kafka.spout.test.KafkaSpoutTopologyMainNamedTopics
+```
+will submit the topologies set up by KafkaSpoutTopologyMainNamedTopics to Storm.
+
+Note that this example produces a jar containing all dependencies for ease of use. In a production environment you may want to reduce the jar size by extracting some dependencies (e.g. org.apache.kafka:kafka-clients) from the jar. You can do this by setting the dependencies you don't want to include in the jars to `provided` scope, and then manually copying the dependencies to your Storm extlib directory.
--- End diff --
Thanks, I didn't know about this flag. It's much better, will replace references to extlib.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2243: STORM-2658: Extract storm-kafka-client examples to...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on a diff in the pull request:
https://github.com/apache/storm/pull/2243#discussion_r130255066
--- Diff: examples/storm-kafka-client-examples/pom.xml ---
@@ -42,7 +42,9 @@
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka</artifactId>
<version>${project.version}</version>
+ <!-- You can reduce jar size by uncommenting this and putting dependencies in $STORM-HOME/extlib instead of including them in the jar
--- End diff --
Same above. Maybe we can let users choose how they provide dependencies, change the sentence to `...uncommenting this and providing dependencies manually.` or so. My intention is that we don't recommend putting dependencies to extlib directory, unless they know what they're doing (affecting whole topologies' dependencies)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2243: STORM-2658: Extract storm-kafka-client examples to storm-...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2243
@hmcl I think I addressed everything. Please look again. Thanks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---