KAFKA-19829: Implement group-level initial rebalance delay #20755

RaidenE1 · 2025-10-23T04:00:14Z

During testing, an artifact of the new rebalance protocol showed up. In
some cases, the first joining member gets all active tasks assigned, and
is slow to revoke the tasks after more member has joined the group. This
affects in particular cases where the first member is slow (possibly
overloaded in the case of cloudlimits benchmarks) and there are a lot of
tasks to be assigned.

To help with this situation, we want to introduce a new group-specific
configuration to delay the initial rebalance.

Main changes:

add 3s as default delay to GroupConfig & GroupCoordinatorConfig
add is_scheduled method to timer
add rebalance delay logic
- check if it's initial rebalance by using bumpGroup & groupEpoch == 0
- if it's initial rebalance then trigger the timer and bump the group
  by two
- when the timer is end, compute a new target assignment
add related tests
ignore epoch check in DescribeStreamsGroupTest since it's flaky now
add setGroupStreamsInitialRebalanceDelay to EmbeddedKafkaCluster` so
we can manully set the config.
Set delay to 0 in most of the tests

Copilot

Pull Request Overview

This PR introduces a new group-level configuration for streams groups to delay the initial rebalance. This helps prevent the first joining member from being assigned all active tasks and then slowly revoking them when additional members join, which was causing performance issues particularly with slow or overloaded members.

Key changes:

Added STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG configuration with a default value of 3000ms at both the group coordinator and group config levels
Implemented accessor methods to retrieve the initial rebalance delay setting

Reviewed Changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

File	Description
GroupCoordinatorConfig.java	Added broker-level configuration constant, field, initialization, and accessor for streams group initial rebalance delay
GroupConfig.java	Added group-level configuration constant, field, initialization, and accessor for streams initial rebalance delay

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

lucasbru

looks good to me as a first step, but I am a bit uneasy introducing a config that is not used at all, not even in tests. Maybe we should just go ahead and extend this PR to a working subset of the feature.

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java

lucasbru · 2025-10-23T12:37:38Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java

    public static final String STREAMS_GROUP_MAX_STANDBY_REPLICAS_DOC = "The maximum allowed value for the group-level configuration of " + GroupConfig.STREAMS_NUM_STANDBY_REPLICAS_CONFIG;

+    public static final String STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG = "group.streams.initial.rebalance.delay.ms";
+    public static final int STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT = 3000;


@mjsax Are we good with a default of 3 seconds for delaying the initial rebalance?

RaidenE1 · 2025-10-23T14:02:02Z

@lucasbru Hi Lucas, I'm not going to merge it now, I'll add more things to it. Just want you to take a quick look to make sure I'm on the right direction!

lucasbru · 2025-10-24T14:22:03Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

+     *
+     * @return An empty result.
+     */
+    private CoordinatorResult<Void, CoordinatorRecord> fireStreamsInitialRebalance(String groupId) {


If this is not doing anything, I think we can inline this.

lucasbru · 2025-10-24T14:23:01Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java


        // Actually bump the group epoch
        int groupEpoch = group.groupEpoch();
+        boolean isInitialRebalance = group.isEmpty();


Group being empty does not mean initial rebalance right? It could be that the group became empty again?

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

lucasbru

I think this makes sense to me (except for the weirdness around initial epoch that I commented on), but I am wondering now if we could indeed compute a new target assignment when the timer triggers. This could be especially useful when we have offloaded assignments (in the future) and we'd otherwise even not get an assignment on the next heartbeat.

Maybe @squah-confluent could have a quick look how this would interact with the offloaded assignment code that he is going to implement in AK.

I think inside fireStreamsInitialRebalance, we could, if group.epoch() > group.assignmentEpoch(), just create the targetAssignmentbuilder for the current group ID. You can get all the information about the group from the groups map instance variable.

I can think of one corner case: If configuredTopology is not defined, we should just skip computing the target assignment. It will be computed on the next heartbeat.

mjsax · 2025-10-25T23:59:35Z

Let's make sure we do not forget to update the docs, for the new config.

Copilot

Pull Request Overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated 3 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-10-29T16:25:16Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

+        boolean isInitialRebalance = (group.isEmpty() && groupEpoch == 0);
        if (bumpGroupEpoch) {
-            groupEpoch += 1;
+            if (isInitialRebalance) {
+                groupEpoch += 2;
+            } else {
+                groupEpoch += 1;
+            }


The isInitialRebalance check on line 1988 is evaluated before bumpGroupEpoch is processed (line 1989-1999), but the condition relies on group.isEmpty() which becomes false after the member is added to the group earlier in the method. This means isInitialRebalance will always be false for the first member join since the member is already added to the group before this check. The timer scheduling on line 2003 may never execute as intended.

Copilot · 2025-10-29T16:25:17Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

+        // Schedule initial rebalance delay for new streams groups to coalesce joins.
+        int initialDelayMs = streamsGroupInitialRebalanceDelayMs(groupId);
+        if (isInitialRebalance && initialDelayMs > 0) {


The isInitialRebalance check on line 1988 is evaluated before bumpGroupEpoch is processed (line 1989-1999), but the condition relies on group.isEmpty() which becomes false after the member is added to the group earlier in the method. This means isInitialRebalance will always be false for the first member join since the member is already added to the group before this check. The timer scheduling on line 2003 may never execute as intended.

Copilot · 2025-10-29T16:25:17Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

+        if (isInitialRebalance && initialDelayMs > 0) {
+            timer.scheduleIfAbsent(
+                    streamsInitialRebalanceKey(groupId),
+                    initialDelayMs,
+                    TimeUnit.MILLISECONDS,
+                    false,
+                    () -> fireStreamsInitialRebalance(groupId)
+            );
+        }


The timer is only scheduled when bumpGroupEpoch is true (due to placement after line 1999), but isInitialRebalance is checked independently. If the first member join doesn't trigger a group epoch bump (which could happen if no metadata changes occur), the initial rebalance timer may not be scheduled. Consider scheduling the timer before the bumpGroupEpoch condition or ensuring it's always scheduled for the first member.

Copilot

Pull Request Overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated 4 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

...nator/src/main/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilder.java

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

lucasbru

Thanks for the PR! I made a pass on the production code.

...nator/src/main/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilder.java

...coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

lucasbru

Mostly looks good to me, I have a another few comments.

@squah-confluent could you maybe have a look at the code around the timer?

lucasbru · 2025-11-05T19:49:10Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

        }

+        // Schedule initial rebalance delay for new streams groups to coalesce joins.
+        boolean isInitialRebalance = (initialGroupEpoch == 0);


could we just use group.groupEpoch() here?

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

lucasbru · 2025-11-05T20:21:28Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

            // replayed because its coordinator result has a non-null appendFuture.
            createGroupTombstoneRecords(group, records);
+            // Cancel initial rebalance timer.
+            timer.cancel(streamsInitialRebalanceKey(groupId));


I think we have to move this into createGroupTombstoneRecords, so that it's also done when delete group is called

It's an odd place to move the cancellation, purely because of the naming, but I can't see where else it'd go.

squah-confluent · 2025-11-06T17:34:45Z

Thanks for the PR!

The interaction with delayed assignments would be fine. The initial rebalance delay would be just another condition to take into account when deciding whether to start an assignment run.

The timer usage looks fine. We're missing some cleanup as @lucasbru pointed out.

config setup

fc586ac

github-actions bot added triage PRs from the community group-coordinator small Small PRs labels Oct 23, 2025

lucasbru requested review from Copilot and lucasbru October 23, 2025 07:48

lucasbru self-assigned this Oct 23, 2025

Copilot AI reviewed Oct 23, 2025

View reviewed changes

lucasbru reviewed Oct 23, 2025

View reviewed changes

add rebalance delay logic

33f9667

github-actions bot removed small Small PRs triage PRs from the community labels Oct 24, 2025

lucasbru reviewed Oct 24, 2025

View reviewed changes

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java Show resolved Hide resolved

lucasbru reviewed Oct 24, 2025

View reviewed changes

mjsax added the streams label Oct 24, 2025

fix

e57d4a1

RaidenE1 added 4 commits October 27, 2025 14:21

fire initial rebalance

164143f

add test for groupcoordinator config

e2410bc

test for groupconfig and groupcoordinatorconfig

3c163b7

add test for coordinator timer

b0814e2

lucasbru added the ci-approved label Oct 29, 2025

lucasbru requested a review from Copilot October 29, 2025 16:12

Copilot AI reviewed Oct 29, 2025

View reviewed changes

RaidenE1 added 2 commits October 29, 2025 19:43

update existing tests

f402c05

test for rebalance delay logic

6ea4ee5

RaidenE1 requested a review from Copilot October 30, 2025 18:08

Copilot AI reviewed Oct 30, 2025

View reviewed changes

RaidenE1 added 3 commits October 30, 2025 15:06

remove empty line

6867fca

fix targetassignment epoch bug

c94fed0

fix tests

90a29c5

github-actions bot added core Kafka Broker tools labels Nov 3, 2025

fix RestoreIntegrationTest

82bf5f8

lucasbru reviewed Nov 4, 2025

View reviewed changes

fix

b581df2

lucasbru reviewed Nov 5, 2025

View reviewed changes

RaidenE1 added 2 commits November 6, 2025 21:22

fix

7569d3c

fix

c3824bf

KAFKA-19829: Implement group-level initial rebalance delay #20755

Are you sure you want to change the base?

KAFKA-19829: Implement group-level initial rebalance delay #20755

Uh oh!

Conversation

RaidenE1 commented Oct 23, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

lucasbru Oct 23, 2025

Choose a reason for hiding this comment

Uh oh!

RaidenE1 commented Oct 23, 2025

Uh oh!

lucasbru Oct 24, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru Oct 24, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

mjsax commented Oct 25, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Uh oh!

Copilot AI Oct 29, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 29, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 29, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

lucasbru Nov 5, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

lucasbru Nov 5, 2025

Choose a reason for hiding this comment

Uh oh!

RaidenE1 commented Oct 23, 2025 •

edited by github-actions bot

Loading