Newly added partitions are not assigned to any consumer in non-uniform subscription scenario

[searoz]

Description

In some scenarios, adding new partitions to existing topics does not assign these partitions to consumers that are currently subscribed to these topics. This happens only when:

• New partitions are added to topics that are not consumed by a group leader
• partition.assignment.strategy is either range or roundrobin. cooperative-sticky seems to fix this bug

I haven't provided any logs because the scenario seems easy enough to replicate, and because I need to clear sending you the logs with our security department. But if you need any - please feel free to ask, I'll see what I can do.

How to reproduce

• Create 3 topics with arbitrary amount of partitions
• Create 3 consumers in the same consumer group and subscribe each consumer to one topic each. Make sure that one consumer joins a bit earlier than the other 2 (so it becomes a group leader)
• Add new partitions to topics that are not consumed by a group leader

Expected behavior

New partitions are assigned to non-leader consumers

Actual behavior

New partitions remain unassigned. Group rebalancing just re-assigns old partitions.
Note that if you add new partitions to a topic consumed by a group leader then everything works as expected.

Checklist

• librdkafka version (release number or git tag): 2.3.0
• Apache Kafka version: 3.6.0
• librdkafka client configuration:

"auto.offset.reset" = "earliest"
"enable.auto.commit" = "true"
"enable.auto.offset.store" = "false"
"topic.metadata.refresh.interval.ms" = "1000"
"metadata.max.age.ms" = "1000"

• Operating system: Windows 10 x64
• Provide logs (with debug=.. as necessary) from librdkafka
• Provide broker log excerpts
• Critical issue

[kalduzov]

Greetings.

Because Our scenarios involve using multiple topics for one group and dynamically adding partitions during periods of increased load - this case is very critical.

We really expect that the consumer leader will correctly rebalance new partitions for the topic. However, this only happens when the leader is first connected, and does not happen dynamically while it is running.

[kalduzov]

@milindl, hi.

I saw you took over this issue. Please tell me, did you manage to reproduce this behavior?

[milindl]

Hi @kalduzov , @searoz , I was about to start investigating this, but can you help me with understanding what's the case for using multiple consumers in one consumer group, but which are consuming from a disjoint set of topics? That's what the situation is, if I'm understanding correctly.

[mensfeld]

what's the case for using multiple consumers in one consumer group, but which are consuming from a disjoint set of topics?

I can answer that in a different context (not related to the OP): you can do this to achieve connection multiplexing where polling is independent and then you can achieve different resources allocation for those topics. Sometimes it balances better under load.

[searoz]

Hi @kalduzov , @searoz , I was about to start investigating this, but can you help me with understanding what's the case for using multiple consumers in one consumer group, but which are consuming from a disjoint set of topics? That's what the situation is, if I'm understanding correctly.

Sometimes we subscribe one consumer to several topics to lessen cpu and memory consumption.

[emasab]

What happens sometimes if that the consumer group leader doesn't see the new partitions because of metadata eventual consistency.
To explain it better:

1. in librdkafka a member only checks its topics and if it sees changes it re-joins the group
2. let's say there are two consumers A and B and A is subscribed to topicA while B is subscribed to topicB
3. B sees 2 partitions are added to topicB that now has 4 partitions. It re-joins the group. The leader remains A as it's sticky on the coordinator.
4. A receives B assigned partititons and subscribed topics and refreshes metadata for their topic (topicA) and the other members topics (topicB). This time a different broker replies with stale metadata and A calculates the assignment with 2 partitions instead of 4.
5. Now the assignment is distributed with SyncGroup. Member B won't re-join the group as it already saw the metadata change. Member A won't see the changes to topicB later as it's not monitoring that topic. The partitions remain unassigned until a subsequent re-balance.

Have to see what happens when there are more members subscribed to topicB. For the moment it's suggested to have uniform subscriptions within the same group if possible.

A possible solution would be that the leader is monitoring the topics for the whole group, so it's the only one to re-join and has all the metadata changes when it tries to calculate the assignment.