...

Current state: Under Discussion

Discussion thread: TBD Here

JIRA:

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Consumer groups are an essential mechanism of Kafka. They allow consumers to share load and elastically scale by dynamically assigning the partitions of topics to consumers. In our current model of consumer groups, whenever a rebalance happens every consumer from that group experiences downtime - their poll() calls block until every other consumer in the group calls poll(). That is due to the fact that every consumer needs to call JoinGroup in a rebalance scenario in order to confirm it is still in the group. 

Today, if the client has configured `max.poll.interval.ms` to a large value, the group coordinator broker will take in an unlimited number of join group requests into the membership metadataand the rebalance could therefore continue for an unbounded amount of time. There is a also the
potential risk described in

Public Interfaces

We propose to add a new configuration into KafkaConfig.scala, and its behavior will affect the following coordinator APIs:

def handleJoinGroup(...)

where we shall enforce the group size capping rules upon requests.

Proposed Changes

We shall add a config called group.max.size on the coordinator side.

val GroupMaxSizeProp = "group.max.size"
...
val GroupMaxSize = 1000000
...
.define(GroupMaxSizeProp, INT, Defaults.GroupMaxSize, MEDIUM, GroupMaxSizeDoc)

The default value 1_000_000 proposed here is based on a rough size estimation of member metadata (120B or even larger), so the max allowed memory usage per group is 120B * 1_000_000 = 120 MB which should be sufficient large number of 5X~10X for most use cases I know. Further discussion is welcomed on defining the default value.

occurs and puts additional CPU strain).  
The root of the problem isn't necessarily the client's behavior (clients can behave any way they want), it is the fact that the broker has no way to shield itself from such a scenario.

Further, large consumer groups are not very practical with our current model due to two reasons:
1. The more consumers there are, the likelier it is that one will fail/timeout its session, causing a rebalance
2. Rebalances are upper-bounded in time by the slowest-reacting consumer. The more consumers, the higher the chance one is slow (e.g called poll() right before the rebalance and is busy processing the records offline). This means that rebalances are more likely to be long-lived and disruptive to consumer applications


To ensure stability of the broker, this KIP proposes the addition of a configurable upper-bound for the number of consumers in a consumer group. Adding such a config with a sensible default value and documentation would ensure broker protection and help guide users on using consumer groups effectively.

Public Interfaces

Add a new cluster-level group.max.size config with a default value of 250.

Add a new response errorImplementation wise we shall block registration of new member once a group reaches its capacity, and define a new error type:

GROUP_MAX_SIZE_REACHED(77, "Consumer group is already at its full capacity.",
 GroupMaxSizeReachedException::new);

Proposed Changes

We shall block registration of new member once a group reaches its configured capacity. Any subsequent `JoinGroup` requests will receive a response with the `GROUP_MAX_SIZE_REACHED` error.

Since the cap should never should never be reached in practice, the consumer would fail itself  the consumer will fatally exit upon receiving this error message to reduce load on broker side because reaching capacity limit is a red flag indicating some client side logic bug and should be prohibited to ensure server stability.

Compatibility, Deprecation, and Migration Plan

This is a backward compatible change. Old clients will still fail by converting the new error to the non-retriable UnknownServerException

Rejected Alternatives

...

• Topic-level config
  • It is harder to enforce since a consumer group may touch multiple topics. One approach would be to take the min/max of every topic's group size configuration.
  • This fine-grained configurability does not seem needed for the time being and may best be left for the future if the need arises
• There are other ways of limiting how long a rebalance can take, discussed here
  • In the form of time - have a max rebalance timeout (decoupled from `max.poll.interval.ms`)
    • Lack strictness, a sufficiently buggy/malicious client could still overload the broker in a small time period
  • In the form of memory - have a maximum memory bound that can be taken up by a single group
    • Lacks intuitiveness, users shouldn't think about how much memory a consumer group is taking