STORM-2236: Kafka Spout with manual partition management. #1835

liurenjie1024 · 2016-12-19T10:16:23Z

As the title says.

srdo

This looks good to me, barring a few nits and the caught Exception in KafkaSpout. Good work :)

srdo · 2016-12-19T19:42:14Z

...nal/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaPartitionReaders.java

+import java.util.HashSet;
+
+/**
+ * Created by liurenjie on 19/12/2016.


Please remove these.

srdo · 2016-12-19T19:48:44Z

external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java

@@ -120,6 +124,8 @@ public void open(Map conf, TopologyContext context, SpoutOutputCollector collect
            commitTimer = new Timer(500, kafkaSpoutConfig.getOffsetsCommitPeriodMs(), TimeUnit.MILLISECONDS);
        }

+        // Manual partition assignment


This doesn't seem like it belongs here anymore

srdo · 2016-12-19T19:48:57Z

external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java

@@ -75,6 +78,7 @@
    private transient KafkaSpoutRetryService retryService;              // Class that has the logic to handle tuple failure
    private transient Timer commitTimer;                                // timer == null for auto commit mode
    private transient boolean initialized;                              // Flag indicating that the spout is still undergoing initialization process.
+    private transient KafkaRecordsFetcher<K, V> recordsFetcher;         // Class that encapsulate the log of partition management


I don't understand the comment

Sorry, that's a typo. I've fixed in the new patch.

srdo · 2016-12-19T19:49:42Z

external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java

+                try {
+                    setWaitingToEmit(pollKafkaBroker());
+                } catch (Exception e) {
+                    LOG.error("Failed to poll from kafka.", e);


Could you share what exception you're seeing here? We haven't seen the spout be unstable due to exceptions from here.

We have encountered many exceptions: KafkaFetchException, LeaderNoFoundException, etc. These are caused by temporarily kafka partition migration. I don't think we should restart the process in these cases.

Right, that makes sense. We're not getting them because I think automatic assignment avoids them. I think it's fine to catch them, but you should replace the Exception with the specific exception classes you're seeing.

As kafka client's doc says, I've changed the exception to KafkaException.

srdo · 2016-12-19T19:58:48Z

...t/src/main/java/org/apache/storm/kafka/spout/internal/fetcher/ManualKafkaRecordsFetcher.java

+        doRefreshMyPartitions();
+    }
+
+    private void refreshMyPartitionsIfNeed() {


Nit: Should be IfNeeded

srdo · 2016-12-19T20:04:10Z

...t/src/main/java/org/apache/storm/kafka/spout/internal/fetcher/ManualKafkaRecordsFetcher.java

+    }
+
+    public static PartitionAssignmentChangeListener listenerOf(final ConsumerRebalanceListener consumerRebalanceListener) {
+        return new PartitionAssignmentChangeListener() {


You don't have to change it, but on master we're targeting Java 8, so lambdas are an option for this kind of thing :)

Cool, I also love lambda.

srdo · 2016-12-19T20:05:17Z

...rnal/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaPartitionReader.java

+
+import java.util.List;
+
+public interface KafkaPartitionReader {


Nit: Some of these classes could probably go in the internal package, or a new subpackage of the internal package.

Moved to another package.

srdo · 2016-12-19T20:15:03Z

@hmcl You might want to take a look at this too

srdo · 2016-12-20T18:19:07Z

external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java

-                setWaitingToEmit(pollKafkaBroker());
+                try {
+                    setWaitingToEmit(pollKafkaBroker());
+                } catch (KafkaException e) {


I think you should catch https://kafka.apache.org/0100/javadoc/org/apache/kafka/common/errors/RetriableException.html instead, it seems to cover all the exceptions that are expected to be transient (leader change for example, through InvalidMetadataException). Catching KafkaException will break interrupt handling (see #1821) because InterruptException is an instance of KafkaException.

Yes, I've fixed it.

@srdo good point, but the main point is also that the code that is handling the interrupt exceptions is very fragile. Right now you were able to catch this, but it's very difficult for other devs to code accounting for that.

@liurenjie1024 we are basically catching and logging the exception. Is that really the intent? Is there a scenario that we should not catch the exception, in which we would rather let the process die, and then restart ? The way the code currently works, if no new records are polled (i.e. an exception occurred) waitingToEmit will always be false.

hmcl · 2016-12-20T19:00:01Z

@liurenjie1024 once the review is complete and it incorporates the feedback from all the reviewers (I am going through the review now), please squash the commits into one commit, and make sure that this commit has the title of the JIRA it is addressing, as well as a summary of its contents.

harshach · 2016-12-20T20:06:23Z

...rc/main/java/org/apache/storm/kafka/spout/internal/fetcher/AutomaticKafkaRecordsFetcher.java

+import java.util.List;
+import java.util.regex.Pattern;
+
+/**


please remove author names and add a apache license on top of the file

harshach · 2016-12-20T21:16:57Z

...client/src/main/java/org/apache/storm/kafka/spout/internal/fetcher/KafkaRecordsFetchers.java

+                                                                         KafkaConsumer<K, V> consumer,
+                                                                         TopologyContext context,
+                                                                         ConsumerRebalanceListener rebalanceListener) {
+        if (kafkaSpoutConfig.getManualPartitionAssignment()) {


Do we need to check these flags. i.e you are already implementing an interface KafkaRecordsFetcher and any code related to Manual or Automatic should go into respective classes. User can configure one or the other . Instead of checking if its manual or not here we should be doing with in the record fetchers .

I think just a flag is easier to use since the user has to write code to configure it otherwise, especially in the case of using flux to configure storm topology. Though your way is more flexible but I don't think we need the flexibility here since it's an internal implementation and the user do not need to know much about that. Let the user assign the implementation class should be used in cases like tuple builder.

@liurenjie1024 internal details or not we will keep adding one implementation or another so it's better to make this cleaner when we have an opportunity to do so.

@liurenjie1024 irrespective of if it's an internal detail or not we should go with the right approach. We have interfaces for a reason. If we keep adding implementations on how the records are being fetched we need to keep adding these if conditions. Given we are already making changes lets go with the right approach.

I think this is a factory method. Perhaps you could call this method create(...) or newInstance(...), or something along those lines.

The if conditions are wrapped in a factory method so adding more cases will not pollute other components.

The construction of instances can only happen in the open method since it needs topology context.

Interface is used to hide the client from implementation details and I don't think our approach here violated the rule.

srdo · 2016-12-21T07:36:05Z

+1 once the merge conflict is resolved. It would probably be good to add some unit tests of the new classes, but I think this can be merged without them.

liurenjie1024 · 2016-12-21T08:40:28Z

@srdo Do I have the access to resolve the conflicts? Or it should be resolved by those who merge the code?

srdo · 2016-12-21T17:37:57Z

@liurenjie1024 You have access. You need to open your branch in the Git client you usually use and pull the latest commits from master on this repo (https://github.com/apache/storm.git). If you like you can use git pull --rebase to avoid a merge commit in the history. Once your branch is up to date, push it back to your fork. The PR will update automatically.

Here's an example of how it can be done from command line (I have a remote called "upstream" that points to this repository, if you don't have one like it use git remote add upstream https://github.com/apache/storm.git to add it).

git checkout STORM-2236
git pull upstream master --rebase //you'll get a merge conflict
git mergetool //resolve merge conflict (you can also use whatever git GUI you usually use)
git add <the files you resolved conflicts for>
git rebase --continue
//do the above 3 lines until the rebase is over
//At this point you should verify that the branch diff to master looks reasonable. If it looks fine, push to your branch with the following.
git push --force

harshach · 2016-12-21T19:13:26Z

@liurenjie1024 @srdo lets wait till all the pending comments are resolved.

hmcl · 2016-12-22T01:40:53Z

...t/src/main/java/org/apache/storm/kafka/spout/internal/fetcher/ManualKafkaRecordsFetcher.java

+import org.apache.storm.kafka.spout.KafkaSpout;
+import org.apache.storm.kafka.spout.TopicPartitionComparator;
+
+import java.util.*;


Please remove wildcard imports

hmcl · 2016-12-22T01:57:13Z

external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java

+        return partitionRefreshPeriodMs;
+    }
+
+    public boolean getManualPartitionAssignment() {


isManualPartitionAssignment()

liurenjie1024 · 2016-12-22T02:07:00Z

external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java

@@ -61,6 +64,7 @@

    // Storm
    protected SpoutOutputCollector collector;
+    private TopologyContext topologyContext;


Yes, it's used to create records fetcher.

hmcl · 2016-12-22T02:35:24Z

external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java

@@ -591,7 +589,7 @@ public String toString() {

    // =========== Timer ===========

-    private class Timer {
+    public static class Timer {


I think that we should make Timer it's own class, rather than a static inner class, since it's used in other objects as well.

hmcl · 2016-12-22T02:39:36Z

...t/src/main/java/org/apache/storm/kafka/spout/internal/fetcher/ManualKafkaRecordsFetcher.java

+
+    private void doRefreshMyPartitions() {
+        List<TopicPartition> topicPartitions = partitionReader.readPartitions(consumer);
+        Collections.sort(topicPartitions, KAFKA_TOPIC_PARTITION_COMPARATOR);


Why is this sort necessary ?

This sort ensures that all instances of KafkaSpout see the same ordered list so that the calculation is correct.

hmcl · 2016-12-22T02:44:20Z

...ent/src/main/java/org/apache/storm/kafka/spout/internal/partition/KafkaPartitionReaders.java

+import java.util.HashSet;
+
+public final class KafkaPartitionReaders {
+    public static KafkaPartitionReader partitionReaderOf(KafkaSpoutStreams kafkaSpoutStreams) {


these names are very misleading. Can you please call this create or newInstance?

srdo · 2016-12-22T17:07:01Z

external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/Timer.java

+import java.util.concurrent.TimeUnit;
+
+/**
+ * Created by liurenjie on 22/12/2016.


You should really just edit the template in your IDE :)

I'm making the same change here https://github.com/apache/storm/pull/1832/files#diff-73306e32282def8d045e0bcd2d33f59b btw.

Sorry for that. I've fixed it. I'll merge your change later after you submitted it.

liurenjie1024 · 2016-12-23T03:47:58Z

Hi, all:
I've resolved the comments and when this patch can be merged into master? I've read trident source code and it seems that there exists some bugs there, and the bug fix may rely on codes in this branch, so I hope it can be merge ASAP.

srdo · 2016-12-23T07:03:12Z

Please copy the Apache license into all the new files. You can just grab it from the top of KafkaSpout.

srdo · 2016-12-23T16:22:08Z

...client/src/main/java/org/apache/storm/kafka/spout/internal/fetcher/KafkaRecordsFetchers.java

+                kafkaSpoutConfig.getPartitionRefreshPeriodMs(), TimeUnit.MILLISECONDS);
+
+            ManualKafkaRecordsFetcher.PartitionAssignmentChangeListener partitionAssignmentChangeListener = null;
+            if (rebalanceListener != null) {


Won't this cause an NPE in ManualKafkaRecordsFetcher if the null case is hit here? Why does it make sense to call create without a rebalance listener?

I can't figure out a case for a null rebalance listener now, so this maybe an overdesign here.

liurenjie1024 · 2016-12-29T03:29:02Z

@harshach @srdo @hmcl Any updates?

srdo · 2016-12-29T08:39:21Z

LGTM. +1

harshach · 2017-01-03T08:59:43Z

@liurenjie1024 thanks for the updates. Overall looks good to me. Can you squash your commits into one commit. +1 after that.

hmcl · 2016-12-22T02:44:32Z

...ent/src/main/java/org/apache/storm/kafka/spout/internal/partition/KafkaPartitionReaders.java

+        }
+    }
+
+    public static TopicPartition convert(PartitionInfo partitionInfo) {


call this toTopicPartition

hmcl · 2017-01-03T21:42:38Z

+1. Just one minor nit on a name. can you please squash the commits and put a little summary of the patch in the commit message. Thanks.

liurenjie1024 · 2017-01-05T02:13:22Z

Hi, all:
All commits have been squashed into one.

liurenjie1024 · 2017-01-13T09:06:02Z

Any update about the merge process of this patch?

harshach · 2017-01-14T05:16:34Z

@liurenjie1024 thanks for you patience. Merged into master. I would like to merge this into 1.x-branch as well. You are using FunctionalInterface and 1.x branch we are still on java 7 . Do you want to change that open another PR for 1.x-branch?

liurenjie1024 · 2017-01-17T04:36:39Z

@harshach OK I'll submit a PR for that later.

liurenjie1024 mentioned this pull request Dec 19, 2016

STORM-2236 storm kafka client support manual partition management. #1825

Closed

srdo reviewed Dec 19, 2016

View reviewed changes

srdo reviewed Dec 20, 2016

View reviewed changes

harshach reviewed Dec 20, 2016

View reviewed changes

hmcl reviewed Dec 22, 2016

View reviewed changes

liurenjie1024 commented Dec 22, 2016

View reviewed changes

hmcl reviewed Dec 22, 2016

View reviewed changes

liurenjie1024 force-pushed the STORM-2236 branch from 107a33a to 69bfe43 Compare December 22, 2016 03:07

srdo reviewed Dec 22, 2016

View reviewed changes

srdo reviewed Dec 23, 2016

View reviewed changes

hmcl reviewed Jan 3, 2017

View reviewed changes

STORM-2236 Add manual partition management support for kafka support.

00f5408

liurenjie1024 force-pushed the STORM-2236 branch from 320f2e9 to 00f5408 Compare January 4, 2017 04:00

asfgit merged commit 00f5408 into apache:master Jan 14, 2017

srdo mentioned this pull request Jan 18, 2017

STORM-2225: change spout config to be simpler. #1808

Merged


		import java.util.List;

		public interface KafkaPartitionReader {

STORM-2236: Kafka Spout with manual partition management. #1835

STORM-2236: Kafka Spout with manual partition management. #1835

Conversation

liurenjie1024 commented Dec 19, 2016

srdo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

srdo commented Dec 19, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hmcl Dec 22, 2016 • edited Loading

Choose a reason for hiding this comment

hmcl commented Dec 20, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hmcl Dec 22, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

srdo commented Dec 21, 2016

liurenjie1024 commented Dec 21, 2016

srdo commented Dec 21, 2016

harshach commented Dec 21, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

liurenjie1024 commented Dec 23, 2016

srdo commented Dec 23, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

liurenjie1024 commented Dec 29, 2016

srdo commented Dec 29, 2016

harshach commented Jan 3, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hmcl commented Jan 3, 2017

liurenjie1024 commented Jan 5, 2017

liurenjie1024 commented Jan 13, 2017

harshach commented Jan 14, 2017

liurenjie1024 commented Jan 17, 2017

hmcl Dec 22, 2016 •

edited

Loading

hmcl Dec 22, 2016 •

edited

Loading