STORM-2541: Fix storm-kafka-client manual subscription not being able… #2150

srdo · 2017-06-05T13:38:54Z

… to start consuming

I also refactored the manual subscriptions a bit to deduplicate the code. I think there's no real issue with backporting this to 1.x, because clearly no one has been using these subscriptions so the API breakage shouldn't bother anyone.

With regard to testing the manual subscription, I'd like to do that in https://issues.apache.org/jira/browse/STORM-2542 instead, since it'll make the current tests use the manual subscription.

hmcl

@srdo it looks good in general by I suggest that you incorporate the changes I mention in the comments. The PR #2174 addressing STORM-2554 already incorporates these suggestions. You can just do a diff to see what I mean.

hmcl · 2017-06-24T02:34:30Z

...orm-kafka-client/src/main/java/org/apache/storm/kafka/spout/ManualPartitionSubscription.java

            currentAssignment = newAssignment;
-            consumer.assign(currentAssignment);


@srdo lines 62 and 63 should be switched. We want to call the listener only once all is set.

hmcl · 2017-06-24T02:43:07Z

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

+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+
+public interface TopicFilter {


must implement Serializable. Otherwise code didn't work for me. The trident changes are already do this.

hmcl · 2017-06-24T02:43:53Z

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

+
+    @Override
+    public String getTopicsString() {
+        return pattern.pattern();


This implementation should be for a method called getTopicsPattern. Please make a diff with my slight modification of your changes. I suggest that you incorporate that diff in this patch.

I'm wondering if this works? storm-kafka-monitor seems to be getting the subscribed topics out of the getComponentConfiguration method, but is that called more than once? If it isn't being called occasionally then the list this method returns is going to be empty for pattern subscriptions.

edit: What I'm getting at is that the pattern based topic string should be returning all existing topics that match the pattern in order to work with storm-kafka-monitor, but that list isn't known until the spout is running and has requested its subscription. I'm not sure the configuration storm-kafka-monitor is looking at is updated after the topology is built.

@priyank5485 can you answer the storm-kafka-monitor question here.

@srdo Isn't the kafka monitor a runtime thing? If so, it won't have any values until the topology is running, right ? I believe that the approach I am suggesting in the Trident PR will solve what both you and I have in mind. Please take a look at the following:

https://github.com/hmcl/storm-apache/blob/ca420a4a960c9667262833add74e231fe8d5518b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/PatternTopicFilter.java#L60-L67

https://github.com/hmcl/storm-apache/blob/ca420a4a960c9667262833add74e231fe8d5518b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/PatternTopicFilter.java#L35

https://github.com/hmcl/storm-apache/blob/ca420a4a960c9667262833add74e231fe8d5518b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/PatternTopicFilter.java#L53

@hmcl I went and tried out storm-kafka-monitor from the #2174 branch, and ran into multiple issues.

The PatternTopicFilter code throws an NPE because the spout calls getTopicString before the Set is initialized (the method is used for logging). This prevents the topology from being submitted.

It looks like the Kafka consumer code is not being correctly shaded into the storm-kafka-monitor jar, so the bash script didn't work for me. I'm not sure if this was a quirk of my build, since the consumer code is present in the 1.1.0 release storm-kafka-monitor jar.

After fixing the two other issues, I tried submitting a topology consuming the "test*" pattern from a Kafka instance where topics "test", "test-1" and "test-2" were present. The Storm UI offset lag request returns the following, no matter how long I wait after the topology is running

{"kafka-spout":{"spoutId":"kafka-spout","errorInfo":"Unable to get offset lags for kafka. Reason: org.apache.kafka.shaded.common.errors.InvalidTopicException: Topic '' is invalid\n","spoutType":"KAFKA"}}

That code (

storm/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java

Line 91 in 64e29f3

commands.add((String)jsonConf.get(configKeyPrefix + "topics"));

) gets the topic list it should request offset lag for via the spout's getComponentConfiguration method, as far as I can tell through this call

storm/storm-client/src/jvm/org/apache/storm/topology/TopologyBuilder.java

Line 541 in 9e31509

Map<String, Object> conf = component.getComponentConfiguration();

. I don't think we can make the current storm-kafka-monitor implementation work for subscriptions that are not known at topology submission time, unless there is some way to update the component configuration when the subscription changes.

I'd also add that I have a few other issues with storm-kafka-monitor:

The jar is installed along with the cluster, and depends on the Kafka version specified in Storm's root POM. Kafka guarantees backwards compatible client-server communication for one release only, so there's a potential coupling between Storm cluster version and Kafka version. If users want to update the Kafka version in storm-kafka-monitor, they have to rebuild that module and replace the jar in their Storm install.

The UI integration uses the storm-kafka-monitor Bash script to start the monitoring code, in order to avoid a dependency between storm-core and storm-kafka-monitor. This prevents the UI integration from working on Windows. We could supply a Windows script as well, but then we'd need to keep the two in sync.

I think it makes sense to make the change you are suggesting simply because it makes the spout log a little nicer, but it still leaves the issue with storm-kafka-monitor not working for pattern subscriptions. It would be good if we could figure out a solution to the other issues as well. Maybe we could track offset lag through the metrics system instead, and just make the spout update it?

I'll create a separate issue for discussing storm-kafka-monitor.

@srdo I think it's good. I still think it's good practice for this class' internal API to return the pattern that it is using to compute the topics string. I am planning on leaving it on my trident change.

@hmcl Sure, I don't see a reason not to have a Pattern getter

srdo · 2017-06-25T15:27:17Z

Created the issue for storm-kafka-monitor here https://issues.apache.org/jira/browse/STORM-2600

priyank5485 · 2017-06-26T02:54:11Z

@srdo @harshach Some comments related to storm-kafka-monitor

[~Srdo] Regarding first point, because the lag request is a http pull request from UI, as long as kafkaSpoutConfig.getSubscription().getTopicsString(); returns the correct value it will work since the open method would have been called eventually. The only change we would need is that when the rebalance listener is invoked we would need to keep track of the topics currently subscribed. For e.g. PatternSubscription can have an instance variable called Set topics that would be correctly updated anytime onPartitionsRevoked or onPartitionsAssigned is called. We can use that instance variable to return the value when getTopicsString is called on that object. Does that work?

Regarding point 2, we could move the kafka client version to storm-kafka-monitor module. Can you elaborate a bit more on your concern? Is it that if kafka cluster is upgraded, storm-kafka-monitor wont work? In that case storm-kafka-client module will also have to be updated and topology jar rebuilt. Correct? I think in general, we have a certain compatibility restriction in that a given storm release works or has been tested with a certain version of kafka clients. Correct me if i am wrong.

Regarding 3, the main reason for creating a separate module and calling bash script from storm UI is so that storm-core does not have a direct kafka dependency since that made sense. For windows, cygwin can be a workaround. Plus, i dont know how many windows users we have. We can start a thread to see if there is really a user base for which we need to support a use case.

I dont know the details about how metrics would work. We could have opinions from other people. If that is indeed the right way to go about it then I am all for changing it.

srdo · 2017-06-26T06:15:13Z

@priyank5485 Thanks for responding, I replied on the linked issue

srdo · 2017-06-26T19:22:06Z

The issues raised in https://issues.apache.org/jira/browse/STORM-2600 aren't really related to this PR, since the pattern based subscriptions were already unsupported by storm-kafka-monitor prior to this change. I think we should look at those issues separately.

hmcl · 2017-06-27T16:17:52Z

+1

hmcl · 2017-06-29T23:04:46Z

@srdo I am +1 on this PR. Let's just add the getter method as we agreed on this comment, and have this PR consist of two commits only. The commit in STORM-2548 PR and its own commit.

hmcl · 2017-06-29T23:09:04Z

@srdo Thanks for your diligence and awesome work refactoring this code. It just made it much better.

srdo · 2017-07-04T20:32:24Z

@hmcl Sorry, I accidentally overwrote this PR with the STORM-2548 content. This shouldn't contain the STORM-2548 changes. I added the getter and fixed the content error.

hmcl · 2017-07-17T17:36:49Z

+1.
@srdo can you please fix the conflicts.
@harshach @priyank5485 can you please take one final look. If you don't have any objection, I suggest that we merge this patch in the next day or so.

hmcl · 2017-07-17T17:24:21Z

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

+    List<TopicPartition> getFilteredTopicPartitions(KafkaConsumer<?, ?> consumer);
+
+    /**
+     * @return A human-readable string representing the subscribed topics.


NIT: representing the topics that pass the filter

… to start consuming

srdo force-pushed the STORM-2541 branch from 0a7df0d to 4b4529a Compare June 5, 2017 19:42

srdo mentioned this pull request Jun 5, 2017

STORM-2542: Remove KafkaConsumer.subscribe API option, make KafkaConsumer.assign the default #2151

Merged

hmcl reviewed Jun 24, 2017

View reviewed changes

srdo force-pushed the STORM-2541 branch from 4b4529a to 41f0ce4 Compare June 24, 2017 14:46

srdo force-pushed the STORM-2541 branch 3 times, most recently from 92e963e to 21e7e30 Compare June 28, 2017 21:32

hmcl mentioned this pull request Jun 29, 2017

STORM-2548: Simplify KafkaSpoutConfig to avoid duplicating KafkaConsu… #2155

Merged

srdo force-pushed the STORM-2541 branch 3 times, most recently from 310683f to eefcf89 Compare July 4, 2017 20:28

srdo force-pushed the STORM-2541 branch from eefcf89 to ae76ff9 Compare July 4, 2017 20:55

hmcl reviewed Jul 17, 2017

View reviewed changes

srdo force-pushed the STORM-2541 branch 2 times, most recently from a360537 to a25e83a Compare July 18, 2017 09:54

srdo mentioned this pull request Jul 18, 2017

STORM-2541: Fix manual partition assignment (1.x) #2223

Merged

STORM-2541: Fix storm-kafka-client manual subscription not being able…

c7b7b89

… to start consuming

srdo force-pushed the STORM-2541 branch from a25e83a to c7b7b89 Compare July 18, 2017 20:37

asfgit merged commit c7b7b89 into apache:master Jul 18, 2017

		currentAssignment = newAssignment;
		consumer.assign(currentAssignment);

STORM-2541: Fix storm-kafka-client manual subscription not being able… #2150

STORM-2541: Fix storm-kafka-client manual subscription not being able… #2150

Uh oh!

Conversation

srdo commented Jun 5, 2017

Uh oh!

hmcl left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

srdo Jun 24, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hmcl Jun 24, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

srdo Jun 25, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

srdo commented Jun 25, 2017

Uh oh!

priyank5485 commented Jun 26, 2017

Uh oh!

srdo commented Jun 26, 2017

Uh oh!

srdo commented Jun 26, 2017

Uh oh!

hmcl commented Jun 27, 2017

Uh oh!

hmcl commented Jun 29, 2017

Uh oh!

hmcl commented Jun 29, 2017

Uh oh!

srdo commented Jul 4, 2017

Uh oh!

hmcl commented Jul 17, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

srdo Jun 24, 2017 •

edited

Loading

hmcl Jun 24, 2017 •

edited

Loading

srdo Jun 25, 2017 •

edited

Loading

hmcl commented Jul 17, 2017 •

edited

Loading