Status

Current state: Draft

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: KAFKA-17747 - Getting issue details... STATUS

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

Motivation

KIP-848 introduces a next generation of consumer rebalance protocol which supports rack-aware partition assignment. In the first implement, the group coordinator computes subscribed topic metadata which contains topic UUID, name, number of partition, and rack set of each partition. If result is difference, the group coordinator bumps group epoch and calculates new target assignment which means it triggers a rebalance. However, the rack set of each partition takes too much space. In KAFKA-17578, a real case of memory usage shows that in a group with 500 members and 2K topic partitions,  partition racks account for 79% of whole ConsumerGroup object. This KIP will get rid of TopicMetadata and introduce a new way to detect two server side rebalance conditions:

• A topic with a new partition.
• A topic partition has rack change.

Public Interfaces

ConsumerGroupPartitionMetadataValue / ShareGroupPartitionMetadataValue

Since topic metadata will be deprecated, we don't need to store it in ConsumerGroupPartitionMetadataValue, either. Both ConsumerGroupPartitionMetadataValue from KIP-848 and ShareGroupPartitionMetadataValue from KIP-932 will be deprecated.

Proposed Changes

New Metadata Image

Without calculating subscribe topic metadata, the group coordinator needs to detect a topic change when it receives new MetadataImage and MetadataDelta. The MetadataDelta contains a topic change as TopicDelta.

New partitions can be detected by TopicDelta#newPartitions.

For rack change, it happens when broker.rack is updated. The broker.rack value is read-only config. It can only be changed when a broker restarts. If a broker stops, it will be removed from topic partition replicas. If it starts, it will be added to topic partition replicas again. In TopicDelta, each partition change is in PartitionRegistration. The rack change can be detected by PartitionRegistration#addingReplicas and PartitionRegistration#removingReplicas.

After catching all rebalance conditions, set a value in the group to indicate that the group need rebalance in the next group heartbeat.

ModernGroup

Add a new boolean attribute triggerRebalanceOnNextHeartbeat to the group with the initial value false. When the rebalance condition is matched, set the value to true. When a rebalance is handled, set the value to false. This will replace function ModernGroup#requestMetadataRefresh.

Remove subscribedTopicMetadata value and related functions like ModernGroup#setSubscriptionMetadata and ModernGroup#computeSubscriptionMetadata.

SubscribedTopicDescribeImpl

Since the group coordinator doesn't compute subscribed topic metadata, the SubscribedTopicDescribeImpl can't use Map<Uuid, TopicMetadata> to number of partition and rack set. We will use MetadataImage as input to replace it.

Compatibility, Deprecation, and Migration Plan

GroupMetadataManager

To replay deprecated records (ConsumerGroupPartitionMetadataValue and ShareGroupPartitionMetadataValue) in __consumer_offsets, the GroupMetadataManager has to keep replay functions and do nothing in it.

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.