Librdkafka: How can the C++ consumer detect topic created after the consumer has started

[zwaheed-rms]

I have a C++ consumer created using RdKafka::KafkaConsumer::create and then we call KafkaConsumer::subscribe to subscribe to a topic. And later we have a loop to consume messages using KafkaConsumer::consume.

The question I have is what happens if that topic does not exist and will the consumer detect if the topic is created later. Currently it seems like that the consumer does not detect if the topic is created after consumer has started. Is there a way to configure the consumer to detect the created topic?

[MaximGurschi]

Yes i second that. Just posted the same question on the .NET wrapper:
[https://github.com/ah-/rdkafka-dotnet/issues/27]

The suggestion was to forward this to librdkafka.

Quick summary:
Under the same scenario described above, the Consumer's log eventually indicates that the returned metadata includes the name of the new topic... yet no message is received.

So far i tried reducing "topic.metadata.refresh.interval.ms" this did not help. Setting "auto.offset.reset" in the Consumer's config throws. Is this expected? On a related note, is there any reason the consumer has to retrieve the whole list of topics? It is a consideration in case the list is large and the consumer needs to retrieve all that data for no reason.

[MaximGurschi]

Please disregard the part of the question about "auto.offset.reset" - i needed to set it in the topic config. Still though this does not help the consumer.

[edenhill]

The KafkaConsumer will currently not pick up on new topics matching the existing subscription.
This will be implemented for the 0.9.2 release.

[MaximGurschi]

That is great, thanks for that. Do you have an idea when 0.9.2 will be out?

[edenhill]

0.9.1 will probably be released during the coming week, after that 0.9.2 development will start. So I would within 1-2 months.

[MaximGurschi]

Thank you!

[zwaheed-rms]

closing the issue, as it's answered.

[zwaheed-rms]

Reopening. Just saw the milestone field, maybe this should be closed once the feature is implemented.

[andrewkcarter]

Any status on this? This is pretty important.

Do you have any recommendations on how to identify this condition and work around it? I'm using confluent-kafka-python.

[edenhill]

I think the plan of attack is something along these lines:

• hook in the metadata refresher with the consumer group handling, scan through subscriptions on each metadata update to check if the subscription set would change (added partitions) or the partition assignment (partition count change).
• if changed: re-Join group with updated subscription set
• to avoid group members with different metadata refresh times to cause multiple group rebalances (e.g., C1 detects metadata change first and triggers a rejoin, some time later C2's metadata refresh picks up the same thing and triggers a rejoin, .. then C3..) we force a metadata refresh prior to re-Join:ing. If the metadata is not older than 10s reuse the existing metadata.
• optionally provide a setting where only the group leader monitors metadata updates and triggers the rejoin process. This saves resources in the cluster as it prevents all group members to pull metadata info.

This will fix:

• subscription updated on newly created topics in the current subscription set.
• subscription updated on newly created topics for regex matches in the subscription set
• assignment (possibly) updated on partition count change

[schaab]

Any update?

[edenhill]

This issue is now fixed on master, please try to verify the fix in your environment.
Thank you.

[MaximGurschi]

Thanks Eden! When do you think the official 0.9.2 release will happen?

[edenhill]

@MaximGurschi Late october