[GitHub] storm issue #2502: new PR for STORM-2306 : Messaging subsystem redesign
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2502 Congrats @roshannaik great effort and perseverance to get this in and thanks to @revans2 for reviewing in great detail. ---
[GitHub] storm issue #2241: STORM-2306 : Messaging subsystem redesign.
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2241 @revans2 here is the initial test cases we are looking to run against master & STORM-2306. Let us know if you would like to any further cases https://docs.google.com/document/d/1trXXK9IfQ1c_Ptq4DoglNhkTvnJK01uoiUutmhTO6CA/edit?usp=sharing --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2241: STORM-2306 : Messaging subsystem redesign.
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2241 @revans2 I am trying to reproduce the worst-case in your last chart. Running TVL topology with 4 spout, 10 splitters, 4 counters, 2 ackers. Here is the code https://gist.github.com/harshach/73dae347c178ac5dd8651cb0e7902412 Running it via following command against Master and STORM-2306 `/bin/storm jar /tmp/storm-starter-2.0.0-SNAPSHOT.jar org.apache.storm.starter.ThroughputVsLatency 500 1 -c topology.workers=1 -c topology.max.spout.pending=500 -c topology.acker.executors=2` You can look at my results here https://docs.google.com/spreadsheets/d/1wPpC3YXp-vTIelRTUVoLxuxIYxUekiIHUpZ2ZEysC4Y/edit#gid=1239810430 in **sheet 2** What I see not much difference between Master and STORM-2306. Let me know if I am missing something in running this test. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2241: STORM-2306 : Messaging subsystem redesign.
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2241 @revans2 I am trying to reproduce the worst-case in your last chart. Running TVL topology with 4 spout, 10 splitters, 4 counters, 2 ackers. Here is the code https://gist.github.com/harshach/73dae347c178ac5dd8651cb0e7902412 Running it via following command against Master and STORM-2306 `/bin/storm jar /tmp/storm-starter-2.0.0-SNAPSHOT.jar org.apache.storm.starter.ThroughputVsLatency 500 1 -c topology.workers=1 -c topology.max.spout.pending=500 -c topology.acker.executors=2` You can look at my results here https://docs.google.com/spreadsheets/d/1wPpC3YXp-vTIelRTUVoLxuxIYxUekiIHUpZ2ZEysC4Y/edit#gid=1239810430 in **sheet 2** What I see not much difference between Master and STORM-2306. Let me know if I am missing something in running this test. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2241: STORM-2306 : Messaging subsystem redesign.
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2241 @HeartSaVioR Its not 12 executors per worker. If you don't pass a command-line argument, it sets parallelism variable here to 4 https://github.com/apache/storm/blob/master/examples/storm-starter/src/jvm/org/apache/storm/starter/ThroughputVsLatency.java#L277 and multiplys with 4 here again https://github.com/apache/storm/blob/master/examples/storm-starter/src/jvm/org/apache/storm/starter/ThroughputVsLatency.java#L359 . So setting a parallelism unit 16 per component. This is nothing to do with how many workers you've. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2241: STORM-2306 : Messaging subsystem redesign.
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2241 @revans2 @HeartSaVioR Here are my findings https://docs.google.com/spreadsheets/d/1wPpC3YXp-vTIelRTUVoLxuxIYxUekiIHUpZ2ZEysC4Y/edit#gid=1644511... 1. Looking at ThroughputvsLatency I found some issues: - By default it adds 51 total threads , that IMO is incorrect when benchmarking in a 4-core machine. - Also it adds two bolts for logging/measurements which might be impacting the numbers https://github.com/apache/storm/blob/master/examples/storm-starter/src/jvm/org/apache/storm/starter/... https://github.com/apache/storm/blob/master/examples/storm-starter/src/jvm/org/apache/storm/starter/... - It also throttles the spout https://github.com/apache/storm/blob/master/examples/storm-starter/src/jvm/org/apache/storm/starter/... I did the following changes: - Disable the HTTP and Logging bolts - Disable throttling spout, we want spout to run as fast as it can - reduced the executor counts If you see lines from 78 - 102. Apache Master clearly couldn't handle the faster spout and starts timing out. Perf degrades considerably and very quickly. Where as STORM-2306 not only was able to handle the faster spout and delivered stable and processing at more start out being 10x faster then improves to 35x faster compared to master. 2. Also ran storm-perf topologies ConstSpoutIdNullBoltIdTopo and ConstSpoutNullBoltTopo. These topologies are trying to see whats the message throughput and latency when there are only 2 components involved without including any external dependencies. Essentially testing the messaging system. From line 3-45 you can see with this patch we are getting under 10ms (depends on the topology) compare to an avg of 250ms+. (with batchSize=1) 3. Also ran storm-examples ThroughputVsLatency with 2 workers. Here there is clearly a bug which is prevent inter-worker communication so don't have comparative numbers. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2241: STORM-2306 : Messaging subsystem redesign.
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2241 @HeartSaVioR I don't mind breaking this into multiple PRs if it helps reviewing and merging in. Its up to @roshannaik . --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2241: STORM-2306 : Messaging subsystem redesign.
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2241 @HeartSaVioR lets keep this discussion to reviews. This is not forum to discuss what one should tweet or not that's up to individuals. Nobody is trying to promote something that's not feasible lets not try to be a moral authority here to suggest what one can do or not. Regarding breaking this into multiple PRs addressing different subsystems, that's a reasonable ask. But lets wait before we go down that path we need to look into the issues raised here and reproduce the case. I am running few tests myself and I'll report my findings. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2241: STORM-2306 : Messaging subsystem redesign.
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2241 @revans2 Do you mind posting your storm.yaml or are you running with defaults. We will try to see if we can reproduce this same behavior on our side. If there are any bugs we will work to fix it and but its shows great potential on the perf improvements. Regarding posting to twitter , Yes we are very excited about the patch and definitely want to share the results with the community. Not sure why you are getting upset about it. Its important that we make these perf improvements and also let the community know that there are continuous improvements in Storm. If you found a bug thats great thats why we've PR and review process in place. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2151: STORM-2542: Remove KafkaConsumer.subscribe API option, ma...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2151 @srdo are we not planning on pushing this into 1.x-branch? --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2208: [STORM-2627] The annotation of storm.zookeeper.topology.a...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2208 @HeartSaVioR wouldn't that be an issue incase of non-secure cluster if we are defaulting to "digest"? --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2147: STORM-2538: New kafka spout emits duplicate tuples
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2147 @hmcl @srdo I don't think we need this given this PR https://github.com/apache/storm/pull/2151 makes manual assignment as default. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2151: STORM-2542: Remove KafkaConsumer.subscribe API option, ma...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2151 +1. Thanks @srdo this looks great. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2155: STORM-2548: Simplify KafkaSpoutConfig to avoid duplicatin...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2155 still +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2217: [1.x-branch] [STORM-2505] OffsetManager should account fo...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2217 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2215: STORM-2548: Simplify KafkaSpoutConfig (1.x)
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2215 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm pull request #2215: STORM-2548: Simplify KafkaSpoutConfig (1.x)
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/2215#discussion_r127821694
--- Diff:
external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/SerializableDeserializer.java
---
@@ -16,10 +16,14 @@
package org.apache.storm.kafka.spout;
import java.io.Serializable;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Deserializer;
/**
* @param The type this deserializer deserializes to.
+ * @deprecated Avoid using this class. Use {@link
KafkaSpoutConfig.Builder#setProp(java.lang.String, java.lang.Object) } with
+ * {@link ConsumerConfig#KEY_DESERIALIZER_CLASS_CONFIG} and {@link
ConsumerConfig#VALUE_DESERIALIZER_CLASS_CONFIG} instead
*/
-public interface SerializableDeserializer extends Deserializer,
Serializable {
+@Deprecated
--- End diff --
why are we adding this back?
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2208: [STORM-2627] The annotation of storm.zookeeper.topology.a...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2208 @liu-zhaokun I think the comment there meant to say by default it will be "No Authentication". I.e Its users responsibility to set to digest in a secure clusters. But since the default settings for non-secure the comment looks ok to me. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2209: [STORM-2622] Add owner resource summary on storm UI
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2209 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2165: STORM-2558: Port storm.sh to Powershell and remove outdat...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2165 +1. Tried on windows 10 looks good to me. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm pull request #2157: STORM-2517 storm-hdfs writers can't be subclassed
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/2157#discussion_r124087558
--- Diff:
external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/AbstractHdfsBolt.java
---
@@ -297,9 +296,21 @@ protected Path getBasePathForNextFile(Tuple tuple) {
abstract protected String getWriterKey(Tuple tuple);
-abstract protected AbstractHDFSWriter makeNewWriter(Path path, Tuple
tuple) throws IOException;
+abstract protected Writer makeNewWriter(Path path, Tuple tuple) throws
IOException;
-static class WritersMap extends LinkedHashMap<String,
AbstractHDFSWriter> {
+public interface Writer {
--- End diff --
can you move this into this its own file rather than putting it in HDFSBolt
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2166: [STORM-2559] There are three configurations in defaults.y...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2166 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2173: STORM-2597: Don't parse passed in class paths
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2173 @revans2 this means the users exported STORM_EXT_CLASSPATH must contain a wildcard "*" This could result in issues with current users who are just passing the dir and not adding a wildcard and if they have multiple jars in that dir because of this change it will break right --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2155: STORM-2548: Simplify KafkaSpoutConfig to avoid duplicatin...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2155 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2176: STORM-2598 Add proxy server option for dependency resolve...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2176 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm pull request #2150: STORM-2541: Fix storm-kafka-client manual subscrip...
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/2150#discussion_r123881560
--- Diff:
external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/PatternTopicFilter.java
---
@@ -0,0 +1,59 @@
+/*
+ * Copyright 2017 The Apache Software Foundation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.storm.kafka.spout;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+
+/**
+ * Filter that returns all partitions for topics matching the given {@link
Pattern}.
+ */
+public class PatternTopicFilter implements TopicFilter {
+
+private final Pattern pattern;
+
+/**
+ * Creates filter based on a Pattern. Only topic names matching the
Pattern are passed by the filter.
+ * @param pattern The Pattern to use.
+ */
+public PatternTopicFilter(Pattern pattern) {
+this.pattern = pattern;
+}
+
+@Override
+public List
getFilteredTopicPartitions(KafkaConsumer consumer) {
+List allPartitions = new ArrayList<>();
+for (Map.Entry<String, List> entry:
consumer.listTopics().entrySet()) {
+if (pattern.matcher(entry.getKey()).matches()) {
+for (PartitionInfo partitionInfo: entry.getValue()) {
+allPartitions.add(new
TopicPartition(partitionInfo.topic(), partitionInfo.partition()));
+}
+}
+}
+return allPartitions;
+}
+
+@Override
+public String getTopicsString() {
+return pattern.pattern();
--- End diff --
@priyank5485 can you answer the storm-kafka-monitor question 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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2117: STORM-2515: Fix most checkstyle violations in storm-kafka...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2117 @srdo thought I sent and replied your earlier emails as well. Looks like some issue with gmail they are showing as sent but didn't reached the mailing list. I sent them again. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2117: STORM-2515: Fix most checkstyle violations in storm-kafka...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2117 @srdo Sorry not meant as negative for the PR. But want to get a better exposure to everyone on changes that we make and for users/devs who might not be able to follow dev list day-in/day-out and its easy to get buried in the new emails in mailing list. I started discussion around having KIP (Kafka) proposal docs, so that there is a central we are documenting the critical fixes and changes. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2117: STORM-2515: Fix most checkstyle violations in storm-kafka...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2117 @srdo @HeartSaVioR @erikdw I understand this PR is merged. But we should be extremely careful when we break the backward incompatibility , if it justifies better implementation of a connector yes but I don't agree on doing this for check style issues. I think we need to propose a better way to get the changes in. We can adopt the KIP style proposals in Storm community that can be discussed and voted upon before we agree. We can skip small changes/bug fixes out of this process and use it bigger changes and back-ward incompatible changes. I started discussion thread we can continue there. 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2121: (1.x) STORM-2518 Handles empty name for "USER type" ACL w...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2121 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2120: STORM-2518 Handles empty name for "USER type" ACL when no...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2120 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2122: STORM-2519: Modify AbstractAutoCreds to look for configKe...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2122 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2112: [STORM-2510] update checkstyle configuration to lower vio...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2112 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2114: STORM-2511: Submitting a topology with name containing un...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2114 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2115: STORM-2512: Make constructor public and add one more buil...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2115 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2116: STORM-2512: Make constructor public and add one more buil...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2116 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2104: [STORM-2505] Spout to support topic compaction
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2104 @vivekmittal I think you need to open another PR against 1.x-branch. Don't think this can be cherry-picked onto 1.x-branch. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2104: [STORM-2505] Spout to support topic compaction
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2104 +1. Thanks @vivekmittal --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2104: [STORM-2505] Spout to support topic compaction
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2104 @vivekmittal over LGTM. I am +1 once the method name is addressed. Thanks for finding & addressing the bug. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm pull request #2104: [STORM-2505] Spout to support topic compaction
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/2104#discussion_r115551459
--- Diff:
external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java
---
@@ -49,10 +51,14 @@ public OffsetManager(TopicPartition tp, long
initialFetchOffset) {
LOG.debug("Instantiated {}", this);
}
-public void add(KafkaSpoutMessageId msgId) { // O(Log N)
+public void ack(KafkaSpoutMessageId msgId) { // O(Log N)
--- End diff --
@vivekmittal I think "addAck" is better than just "ack" as its part of the
storm apis might confuse the method name. Probably better name would be
"addToAckMsgs" and "addToEmitMsgs"
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2058: [STORM-2466] The example of jaas.conf in jaas_kerberos.co...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2058 @liu-zhaokun I am can help merge this. But my comment is not addressed. I think its better to break this into two files no? 1. storm_jaas.conf which contains storm related sections only StormServer, StormClient, Client 2. zookeeeper_jaas.conf which contains Server having all of them in one will confuse users as having Server section in storm jaas is not required. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2071: STORM-1858: KafkaBolt: sharing a single producer instance...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2071 @vesense also if we make this static all the internal producer state becomes shared and this could result in unexpected behavior as per user. Since producer doesn't call flush to broker until the batch.size is met , in a shared state this could mean multiple instances of bolt can quickly fill up the batch.size vs previously where each has their own state. Again we can run into some other unknows as well. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2071: STORM-1858: KafkaBolt: sharing a single producer instance...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2071 @vesense not sure any benefit doing this static and making one instance per JVM. This actually adds complexity in code without giving any benefit. For the most part , when users configures parallelism it means they are trying to spawn the bolt across the workers and possibly across the machines. So any particular that we want to introduce this complexity. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2098: STORM-2499: Add Serialization plugin for EventHub System ...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2098 @rban1 looks good. Can you add the new config to README and also squash the commits into 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2104: [STORM-2505] Spout to support topic compaction
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2104 @vivekmittal can you squash your commits into singe one. "Topology stopped processing (or died) & topic got compacted (cleanup.policy=compact) leaving offset voids in the topic. Topology stopped processing (or died) & Topic got cleaned up (cleanup.policy=delete) and the offset." In both of these cases are we not getting OffsetOutofRange exception? --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm pull request #2104: [STORM-2505] Spout to support topic compaction
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/2104#discussion_r115412857
--- Diff:
external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java
---
@@ -68,13 +74,34 @@ public OffsetAndMetadata findNextCommitOffset() {
KafkaSpoutMessageId nextCommitMsg = null; // this is a
convenience variable to make it faster to create OffsetAndMetadata
for (KafkaSpoutMessageId currAckedMsg : ackedMsgs) { //
complexity is that of a linear scan on a TreeMap
-if ((currOffset = currAckedMsg.offset()) == nextCommitOffset +
1) {// found the next offset to commit
+currOffset = currAckedMsg.offset();
+if (currOffset == nextCommitOffset + 1) {// found
the next offset to commit
found = true;
nextCommitMsg = currAckedMsg;
nextCommitOffset = currOffset;
-} else if (currAckedMsg.offset() > nextCommitOffset + 1) {
// offset found is not continuous to the offsets listed to go in the next
commit, so stop search
-LOG.debug("topic-partition [{}] has non-continuous offset
[{}]. It will be processed in a subsequent batch.", tp, currOffset);
-break;
+} else if (currOffset > nextCommitOffset + 1) {
+if (emittedOffsets.contains(nextCommitOffset + 1)) {
+LOG.debug("topic-partition [{}] has non-continuous
offset [{}]. It will be processed in a subsequent batch.", tp, currOffset);
+break;
+} else {
+/*
+This case will arise in case of non contiguous
offset being processed.
+So, if the queue doesn't contain offset =
committedOffset + 1 (possible
+if the queue is compacted or deleted), the
consumer should jump to
--- End diff --
Minor nit: "queue" => "topic"
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2102: STORM-2496 Dependency artifacts should be uploaded to blo...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2102 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2100: STORM-2503: Fix lgtm.com alerts on equality and compariso...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2100 Thanks @adityasharad this looks good. +1. Can you please squash your commits into single one. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2097: [STORM-2482] Refactor the Storm auto credential plugins t...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2097 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2081: [STORM-2482] Refactor the Storm auto credential plugins t...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2081 @arunmahadevan we need a PR for master --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #1924: STORM-2343: New Kafka spout can stop emitting tuples if m...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/1924 Thanks @srdo for your patience. Merged into 1.x & master. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2081: [STORM-2482] Refactor the Storm auto credential plugins t...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2081 still +1 after the above comments. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2083: STORM-2421: support lists of childopts in DaemonConfig.
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2083 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2082: expose Tuple for node.js
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2082 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2086: STORM-2491: Adding extra Cassandra configuration paramete...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2086 overall LGTM. +1 . @tandrup would like to see these configs documented here https://github.com/apache/storm/blob/master/external/storm-cassandra/README.md --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2086: STORM-2491: Adding extra Cassandra configuration paramete...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2086 @tandrup for small docs changes etc. we don't file JIRAs but this one had quite few changes good to have that in JIRA and subsequently in CHANGELOG. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2089: STORM-2490: Lambda support
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2089 @vesense this looks good. +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2066: [STORM-2472] kafkaspout should work normally in kerberos ...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2066 @liu-zhaokun following upon my previous comment , do not use stom-kafka-client from 1.0.x-branch as there are lot of bug-fixes went into storm-kafka-client in Storm 1.1 release --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2066: [STORM-2472] kafkaspout should work normally in kerberos ...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2066 @liu-zhaokun I am still not sure why are we adding this code. One shouldn't be using 0.9 kafka clients and this storm-kafka-client will not work with 0.9 as the interface changed in 0.10 and we made changes to storm-kafka-client to start working only from 0.10 kafka clients only. As I said before 0.9 kafka clients shouldn't be used as they are alpha quality. If you look the kafka client docs https://kafka.apache.org/documentation/ , you'll see "**sasl.jaas.config"** this is consumer config one can pass and the storm-kafka-client will allow you to pass this config. So there is necessity to add this code. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2074: Storm 1290:port backtype.storm.local-state-test to java
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2074 +1. once the commits gets squashed. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2074: Storm 1290:port backtype.storm.local-state-test to java
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2074 @kamleshbhatt can you please squash commits into single commit. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2088: [STORM-2486] Prevent cd from printing target directory to...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2088 +1. Thanks @erikdw --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2090: STORM-2489: Overlap and data loss on WindowedBolt based o...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2090 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2086: Adding extra Cassandra configuration parameters
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2086 @tandrup can you please a file a JIRA https://issues.apache.org/jira/browse/ under STORM project. Also update the title of the JIRA and squash the commits in this PR. More details https://github.com/apache/storm/blob/master/DEVELOPER.md#contribute-code https://github.com/apache/storm/blob/master/DEVELOPER.md#pull-requests --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2084: STORM-2488: The UI user Must be HTTP
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2084 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2024: STORM-2349: Add one RocketMQ plugin for the Apache Storm
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2024 LGTM @vesense . +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2029: STORM-2379: update for Elasticsearch 2.
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2029 Thanks @hmcc merged into master. @HeartSaVioR agree we can keep this in master and look at releasing 2.0 instead of back-porting. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2080: STORM-2481 Upgrade Aether version to resolve Aether bug B...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2080 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111462620
--- Diff:
external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java
---
@@ -0,0 +1,189 @@
+/**
+ * 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.rocketmq.spout;
+
+import org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+import
org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import
org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
+import
org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.storm.Config;
+import org.apache.storm.rocketmq.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.ObjectReader;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * RocketMQSpout uses MQPushConsumer as the default implementation.
+ * PushConsumer is a high level consumer API, wrapping the pulling details
+ * Looks like broker push messages to consumer
+ */
+public class RocketMQSpout implements IRichSpout {
+// TODO add metrics
+
+private static MQPushConsumer consumer;
+private SpoutOutputCollector collector;
+private BlockingQueue queue;
+
+private Properties properties;
+private MessageRetryManager messageRetryManager;
+
+public RocketMQSpout(Properties properties) {
+this.properties = properties;
+}
+
+@Override
+public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
+Validate.notEmpty(properties, "Consumer properties can not be
empty");
+boolean ordered = getBoolean(properties,
RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE,
ObjectReader.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+queue = new LinkedBlockingQueue<>(queueSize);
+
+// Since RocketMQ Consumer is thread-safe, RocketMQSpout uses a
single
+// consumer instance across threads to improve the performance.
+synchronized (RocketMQSpout.class) {
+if (consumer == null) {
+consumer = new DefaultMQPushConsumer();
+RocketMQConfig.buildConsumerConfigs(properties,
(DefaultMQPushConsumer)consumer);
+
+if (ordered) {
+consumer.registerMessageListener(new
MessageListenerOrderly() {
--- End diff --
is this a push model from server instead of spout polling?
---
If your project is set up for it, you can reply to this email and have your
reply appea
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111524719
--- Diff:
external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java
---
@@ -0,0 +1,189 @@
+/**
+ * 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.rocketmq.spout;
+
+import org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+import
org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import
org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
+import
org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.storm.Config;
+import org.apache.storm.rocketmq.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.ObjectReader;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * RocketMQSpout uses MQPushConsumer as the default implementation.
+ * PushConsumer is a high level consumer API, wrapping the pulling details
+ * Looks like broker push messages to consumer
+ */
+public class RocketMQSpout implements IRichSpout {
+// TODO add metrics
+
+private static MQPushConsumer consumer;
+private SpoutOutputCollector collector;
+private BlockingQueue queue;
+
+private Properties properties;
+private MessageRetryManager messageRetryManager;
+
+public RocketMQSpout(Properties properties) {
+this.properties = properties;
+}
+
+@Override
+public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
+Validate.notEmpty(properties, "Consumer properties can not be
empty");
+boolean ordered = getBoolean(properties,
RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE,
ObjectReader.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+queue = new LinkedBlockingQueue<>(queueSize);
+
+// Since RocketMQ Consumer is thread-safe, RocketMQSpout uses a
single
+// consumer instance across threads to improve the performance.
+synchronized (RocketMQSpout.class) {
+if (consumer == null) {
+consumer = new DefaultMQPushConsumer();
+RocketMQConfig.buildConsumerConfigs(properties,
(DefaultMQPushConsumer)consumer);
+
+if (ordered) {
+consumer.registerMessageListener(new
MessageListenerOrderly() {
+@Override
+public ConsumeOrderlyStatus
consumeMessage(List msgs,
+
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111462177
--- Diff:
external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java
---
@@ -0,0 +1,189 @@
+/**
+ * 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.rocketmq.spout;
+
+import org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+import
org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import
org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
+import
org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.storm.Config;
+import org.apache.storm.rocketmq.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.ObjectReader;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * RocketMQSpout uses MQPushConsumer as the default implementation.
+ * PushConsumer is a high level consumer API, wrapping the pulling details
+ * Looks like broker push messages to consumer
+ */
+public class RocketMQSpout implements IRichSpout {
+// TODO add metrics
+
+private static MQPushConsumer consumer;
+private SpoutOutputCollector collector;
+private BlockingQueue queue;
+
+private Properties properties;
+private MessageRetryManager messageRetryManager;
+
+public RocketMQSpout(Properties properties) {
+this.properties = properties;
+}
+
+@Override
+public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
+Validate.notEmpty(properties, "Consumer properties can not be
empty");
+boolean ordered = getBoolean(properties,
RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE,
ObjectReader.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+queue = new LinkedBlockingQueue<>(queueSize);
+
+// Since RocketMQ Consumer is thread-safe, RocketMQSpout uses a
single
+// consumer instance across threads to improve the performance.
+synchronized (RocketMQSpout.class) {
--- End diff --
even if its thread-safe shouldn't we consider making per spout instance its
own consumer. That way it will more performant instead of one consumer making a
call to the rocketmq-servers?
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2058: [STORM-2466] The example of jaas.conf in jaas_kerberos.co...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2058 @liu-zhaokun why are we including Server section intended for Zookeeper server in this file? --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2066: [STORM-2472] kafkaspout should work normally in kerberos ...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2066 @liu-zhaokun you can set it via topology.worker.childopts or worker.childopts. Lets is not put work-around for this when there is viable option to set it. Also 0.9 consumer API is for Alpha and its not been used by anyone. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2062: [STORM-2470] kafkaspout should support kerberos
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2062 @liu-zhaokun KafkaSpoutConfig offers this method to add security-related configs https://github.com/apache/storm/blob/master/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java#L227 you can pass in consumer side properties by passing in "sasl.jaas.config" via setProps/setProp method. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2029: STORM-2379: update for Elasticsearch 2.
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2029 @hmcc can you please up-merge your patch. I'll merge it into master & 1.x-branch --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2054: STORM-2462 Adding regex mapper to KerberosPrincipalToLoca...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2054 @ambud we should add the auth_to_local rules an make that as part of storm.yaml config option . Adding regex will not be helpful 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2062: [STORM-2470] kafkaspout should support kerberos
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2062 @liu-zhaokun More details are documented here http://kafka.apache.org/documentation.html#security_sasl . We don't need these changes on storm-kafka-client side --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2062: [STORM-2470] kafkaspout should support kerberos
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2062 @liu-zhaokun with the latest kafka consumer APIs one can pass a keytab and principal via consumer or producer properties. Even before that they can pass the jaas config via JVM param and set security.protocol to SASL_PLAINTEXT. with the new kafka client APIs we are trying hard to remain with configs supported by them. Let's not introduce to new configs on storm side to make it confusing for the user. I am -1 on this. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2053: [STORM-2455] Expose the window start and end timestamp in...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2053 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2026: STORM-2371: Replace existing AMQP eventhub client with th...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2026 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2029: STORM-2379: update for Elasticsearch 2.
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2029 Thanks @hmcc. +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2056: [STORM-2464] update storm-mongodb.md
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2056 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2032: [STORM-2093] Fix permissions in multi-tenant, secure mode
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2032 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2053: [STORM-2455] Expose the window start and end timestamp in...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2053 @arunmahadevan this looks like backward incompatible change? --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2004: STORM-2413: Make new Kafka spout respect tuple retry limi...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2004 @srdo can you squash some of those commits. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2028: Fix headers
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2028 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm pull request #2029: STORM-2379: update for Elasticsearch 2.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/2029#discussion_r110038799
--- Diff: examples/storm-elasticsearch-examples/pom.xml ---
@@ -53,6 +54,11 @@
storm-elasticsearch
${project.version}
+
+org.elasticsearch
+elasticsearch
+${elasticsearch.test.version}
--- End diff --
can you add the scope to provided. This will allow users to pick their own
version of elasticsearch2 while packaging the topology
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2031: Update HdfsSpout.java
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2031 @megha10 thanks for the patch. Pleas follow the guide lines here https://github.com/apache/storm/blob/master/DEVELOPER.md#contribute-code. 1. Open a STORM jira https://issues.apache.org/jira/browse/STORM/?selectedTab=com.atlassian.jira.jira-projects-plugin:summary-panel with a proper title include your propose change. 2. Update the PR title reflect the JIRA title (for examples look at the other PRs) 3. Update your commit to reflect the JIRA title as well. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2042: STORM-2453 Move non-connectors into the top directory
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2042 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2044: [STORM-2454] the default returned value of this method wh...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2044 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2026: Eventhub3
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2026 @rban1 can you change the PR title to reflect the JIRA title. You can look other PRs for example --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2026: Eventhub3
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2026 @rban1 you are keep re-opening the PRs. We should keep only one PR. If you want to address the comments and update the PR , all you need to do is to work on the same git branch and push changes into the same branch that will update the PR you opened. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #1998: Eventhub2
Github user harshach commented on the issue: https://github.com/apache/storm/pull/1998 @rban1 there are still some unaddressed comments and also merge conflicts. Make sure you squashed your commits for the PR as well. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2027: STORM-2432: Storm-Kafka-Client Trident Spout Seeks Incorr...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2027 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #1999: STORM-2409: Storm-Kafka-Client KafkaSpout Support for Fai...
Github user harshach commented on the issue: https://github.com/apache/storm/pull/1999 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm issue #2020: STORM-2425: Storm Hive Bolt not closing open transactions
Github user harshach commented on the issue: https://github.com/apache/storm/pull/2020 +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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm pull request #1998: Eventhub2
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1998#discussion_r106768060
--- Diff:
external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/EventHubBolt.java
---
@@ -84,11 +85,47 @@ public void prepare(Map config, TopologyContext context,
@Override
public void execute(Tuple tuple) {
try {
-
sender.send(boltConfig.getEventDataFormat().serialize(tuple));
+ EventData sendEvent = new
EventData(boltConfig.getEventDataFormat().serialize(tuple));
+ if(boltConfig.getPartitionMode() && sender!=null)
+ sender.sendSync(sendEvent);
+ else if(boltConfig.getPartitionMode() && sender==null)
+ throw new EventHubException("Sender is null");
+ else if(!boltConfig.getPartitionMode() &&
ehClient!=null)
+ ehClient.sendSync(sendEvent);
+ else if(!boltConfig.getPartitionMode() &&
ehClient==null)
+ throw new EventHubException("ehclient is null");
collector.ack(tuple);
- } catch (EventHubException ex) {
+ } catch (EventHubException ex ) {
collector.reportError(ex);
collector.fail(tuple);
+ }catch (ServiceBusException e){
+ collector.reportError(e);
+ collector.fail(tuple);
+ }
+ }
+
+ @Override
+ public void cleanup() {
+ if(sender != null) {
+ try {
+ sender.close().whenComplete((voidargs,error)->{
--- End diff --
FYI, 1.x-branch still is on JDK 1.7 make sure you open another PR for
1.x-branch with JDK7 changes.
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1998: Eventhub2
Github user harshach commented on a diff in the pull request: https://github.com/apache/storm/pull/1998#discussion_r106768022 --- Diff: external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/EventHubBolt.java --- @@ -41,7 +42,8 @@ .getLogger(EventHubBolt.class); protected OutputCollector collector; - protected EventHubSender sender; + protected PartitionSender sender=null; --- End diff -- Don't need to assign null --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] storm pull request #1998: Eventhub2
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1998#discussion_r106768166
--- Diff:
external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubReceiverImpl.java
---
@@ -65,77 +57,81 @@ public EventHubReceiverImpl(EventHubSpoutConfig config,
String partitionId) {
}
@Override
- public void open(IEventHubFilter filter) throws EventHubException {
-logger.info("creating eventhub receiver: partitionId=" + partitionId +
- ", filterString=" + filter.getFilterString());
+ public void open(String offset) throws EventHubException {
+logger.info("creating eventhub receiver: partitionId=" + partitionId +
+", offset=" + offset);
long start = System.currentTimeMillis();
-receiver = new ResilientEventHubReceiver(connectionString, entityName,
- partitionId, consumerGroupName, defaultCredits, filter);
-receiver.initialize();
-
+try {
+ ehClient =
EventHubClient.createFromConnectionStringSync(connectionString);
+ receiver = ehClient.createEpochReceiverSync(
+ consumerGroupName,
+ partitionId,
+ offset,
+ false,
+ 1);
+}catch (Exception e){
+ logger.info("Exception in creating EventhubClient"+e.toString());
+}
long end = System.currentTimeMillis();
logger.info("created eventhub receiver, time taken(ms): " +
(end-start));
}
@Override
- public void close() {
+ public void close(){
if(receiver != null) {
- receiver.close();
+ try {
+receiver.close().whenComplete((voidargs,error)->{
+ try{
+if(error!=null){
+ logger.error("Exception during receiver close
phase"+error.toString());
+}
+ehClient.closeSync();
+ }catch (Exception e){
+logger.error("Exception during ehclient close
phase"+e.toString());
+ }
+}).get();
+ }catch (InterruptedException e){
+logger.error("Exception occured during close phase"+e.toString());
+ }catch (ExecutionException e){
--- 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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1998: Eventhub2
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1998#discussion_r106768041
--- Diff:
external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/EventHubBolt.java
---
@@ -84,11 +85,47 @@ public void prepare(Map config, TopologyContext context,
@Override
public void execute(Tuple tuple) {
try {
-
sender.send(boltConfig.getEventDataFormat().serialize(tuple));
+ EventData sendEvent = new
EventData(boltConfig.getEventDataFormat().serialize(tuple));
+ if(boltConfig.getPartitionMode() && sender!=null)
--- End diff --
can you please add braces around if conditions. This goes to all of the
code.
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
