It would be nice to keep this asynchronous. Instead of calling listTopics, which blocks, could we initiate a new metadata fetch directly and add a listener to handle the completion?

Hey @hachikuji , even I was initially thinking to have it asynchronous, but I could not convince myself that we will gain much with it? schedulePatternSubscriptionTask is already a delayed task that user do not have to wait on. Moreover initiating a new metadata fetch directly needs to handle retries, so more params for max retries, interval, etc. I was under impression that we made ListTopics handle all that so that we will not have to worry about it again during regex implementation. I might be missing something.

@SinghAsDev Since KafkaConsumer has only one thread, even scheduled tasks have to be executed in that thread, which means the user has to wait for them. Since you can't really control when the tasks will be executed, in the worst case, it could turn a non-blocking call into a blocking one. And I don't see why error handling can't be handled asynchronously. For updating regex subscriptions, I wouldn't think it too much of a big deal even if we just ignored failures and waited for the next metadata update, though it would be easy to implement retries with backoff (I think we do this for heartbeats already).

True. For error handling, I was saying that it might require more configs not the complexity. However, we should be able to reuse the existing configs. Will update the patch. Thanks!

@SinghAsDev Are there any error cases to check here? Disconnects for example?

Do you mean in case of errors, I should log it and schedule the task as in case of onFailure?

@SinghAsDev Haha, I can understand why this is a little confusing, but the network layer considers the request a "success" if it gets a response. However, that doesn't mean that there wasn't an error code in that response. It might be nice to handle this at the network layer, but it seems to me that there wasn't a generic way to check for errors. Each response object had the error code at a different location in its schema, so the only thing we could do is pass the response back and let the application determine if there was an error. The one case where we might be able to infer errors generically is by checking ClientResponse.wasDisconnected(), but I don't think we do this either (I've forgotten if there's a good reason for that).

@hachikuji updated. Let me know if I got that right :)

@SinghAsDev I think that works! I wasn't sure if we needed to handle errors in the MetadataResponse itself, but it looks like the topics are only added to the Cluster if they had no error in the response.

Do we need a new config for subscription interval? Could we piggy-back the logic on regular metadata refresh?

Makes sense. Removed.

@SinghAsDev you are not increasing pos! It should be topicsToSubscribe[pos++] = topic, right?

In fact, I would use a list as below:

List<String> topicsToSubscribe = new ArrayList<>(partitions.size());

blahblahblah
....

subscribeTopics(topicsToSubscribe.toArray(new String[0]), true);

But I am fine with the array too.

Aha, nice catch. Will fix. Thanks!

I wonder if we should clear the topic collection when this is set. If we did that, then the change in NetworkClient would be unnecessary. Is there any reason not to?

This will be called only when a pattern is subscribed. At that time yes, what you are saying makes sense. However, in subsequent metadata updates the changes in NetworkClient is still needed as metadata will have topics that matched pattern and were added in last update. Makes sense?

Yeah, I guess that makes sense. It kind of feels weird to be tracking a topic list in Metadata that we're not going to use, however. My feeling is that maybe we shouldn't update Metadata for topics subscribed from a regex. Instead we just update metadata once in subscribe(pattern). Does that make sense?

I am a bit lost here. Even if we clear topics here, topics will be added by KafkaConsumer.onMetadataUpdate and any metadata update after that will fetch metadata for only those topics if NetworkClient's changes are removed. I think I am missing something here :)

Probably my fault. :) I was trying to suggest to not have KafkaConsumer.onMetadataUpdate modify the Metadata topic list, just the subscription topic list. So basically in KafkaConsumer.subscribe(pattern), we set the Metadata allTopics flag, and we don't modify Metadata again (unless the user changes their subscription). Does that make any sense?

This might not work as well as I thought since the metadata is also updated in partitionsFor(). I actually think it would be better for partitionsFor() to use the same approach as listTopics(), but that is out of scope for this issue.

Haha.. I actually made the changes and realized the same.. It does make it look cleaner though. Also, one more issue would be is when subscribed via pattern, metadata will always store metadata for all topics and a fetch will return metadata for all topics, not just for the topics that are actually subscribed via pattern subscription.

For now, I think we can just leave it as it is. If you think we should refactor partitionsFor() then let me know, we probably should create a separate JIRA for it though.

Refactoring partitionsFor does seem worthwhile to me since then Metadata is only modified by subscription (and metadata for subscribed topics is all we care about in a steady state). Maybe we can leave this patch as it is and plan to fix it in that JIRA.

Agreed. Created KAFKA-2506.

To be clear, we'll continue fetching metadata for all the topics that were added to Metadata directly. Is that right? If instead we cleared those topics, would the NetworkClient fetch metadata for all topics until we've set a new subscription?

I think this is a bug in the protocol that if the metadata request contains empty list, then returns the full topic list. This should be fixed in another ticket.

Another two general comments: 1) we could remove a duplicate ConsumerRebalanceListener / RebalanceListener, and then remove the Consumer parameter in the callback functions; 2) we probably can create a member Metadata.Listener inside KafkaConsumer instead of letting KafkaConsumer to implement this interface along with Consumer, since it is only used in subscribe(Pattern).

Just to not drag this ticket further I suggest we do these in later tickets. @hachikuji Does that sound good to you?

@guozhangwang Yep, sounds good to me.

@guozhangwang @hachikuji thanks for the reviews and helping in getting this committed. If there is no JIRA already I can create the JIRA and PR for fixing the issues mentioned by @guozhangwang . Let me know. Thanks!