KAFKA-19765: Store the last used assignment configuration in the group metadata #20702

lucliu1108 · 2025-10-14T20:38:48Z

What

Ticket: https://issues.apache.org/jira/browse/KAFKA-19765
In KIP-1071, there are configurations that affect the assignment, and
that can be configured dynamically.

The previous assignment config of streams group is stored as part of
the StreamsGroupMetadata as a collection of key-value pairs.
If the assignment configuration is changed, group epoch is also
bumped.

Reviewers: Lucas Brutschy [email protected]

lucasbru

Thanks for the PR! I left a lot of comments, but nothing fundamental.

group-coordinator/src/main/resources/common/message/StreamsGroupMetadataValue.json

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java

lucasbru · 2025-10-15T07:17:20Z

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

        // We bump the group epoch.
        int groupEpoch = group.groupEpoch() + 1;
-        records.add(newStreamsGroupMetadataRecord(group.groupId(), groupEpoch, group.metadataHash(), group.validatedTopologyEpoch()));
+        records.add(newStreamsGroupMetadataRecord(group.groupId(), groupEpoch, group.metadataHash(), group.validatedTopologyEpoch(), Map.of()));


Wait, why are dropping the assignment configuration on the floor here? We should use group.assignmentConfiguration here.

...oordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupBuilder.java

...est/java/org/apache/kafka/coordinator/group/streams/StreamsCoordinatorRecordHelpersTest.java

...rc/main/java/org/apache/kafka/coordinator/group/streams/StreamsCoordinatorRecordHelpers.java

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

Copilot

Pull Request Overview

This PR implements support for storing assignment configurations in Kafka Streams group metadata to enable dynamic configuration changes that affect task assignment. The implementation tracks assignment configurations and bumps the group epoch when these configurations change, ensuring proper rebalancing behavior.

Added assignment configuration storage to streams group metadata
Implemented group epoch increment when assignment configurations change
Updated all test cases to accommodate the new assignment configuration parameter

Reviewed Changes

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

Show a summary per file

File	Description
StreamsGroupMetadataValue.json	Added AssignmentConfigs field to store assignment configuration parameters as key-value pairs
StreamsGroup.java	Added assignmentConfigs field and related methods to track assignment configurations
StreamsCoordinatorRecordHelpers.java	Updated newStreamsGroupMetadataRecord to include assignment configurations parameter
GroupMetadataManager.java	Added logic to detect assignment config changes and bump group epoch accordingly
StreamsGroupBuilder.java	Updated test builder to pass assignment configs parameter
StreamsCoordinatorRecordHelpersTest.java	Updated test cases to include assignment configs parameter
GroupMetadataManagerTest.java	Updated all test cases and added new test for assignment config changes

Comments suppressed due to low confidence (1)

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

Extra space before Map.of - should be single space between parameters.

/*

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

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java

Copilot

Pull Request Overview

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

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

Copilot · 2025-10-16T06:23:23Z

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

+            streamsGroup.setLastAssignmentConfigs(
+                value.assignmentConfigs().stream()
+                    .collect(Collectors.toMap(
+                        StreamsGroupMetadataValue.AssignmentConfig::key,
+                        StreamsGroupMetadataValue.AssignmentConfig::value
+                    ))
+            );


The schema marks AssignmentConfigs as nullable with a default of null; for older (pre-field) or null-valued records value.assignmentConfigs() will be null, causing a NullPointerException when calling stream(). Guard with a null/empty check (e.g., List<...> configs = value.assignmentConfigs(); if (configs != null) { ... } else set empty) to ensure backward-compatible replay.

Suggested change

streamsGroup.setLastAssignmentConfigs(

value.assignmentConfigs().stream()

.collect(Collectors.toMap(

StreamsGroupMetadataValue.AssignmentConfig::key,

StreamsGroupMetadataValue.AssignmentConfig::value

))

);

if (value.assignmentConfigs() != null) {

streamsGroup.setLastAssignmentConfigs(

value.assignmentConfigs().stream()

.collect(Collectors.toMap(

StreamsGroupMetadataValue.AssignmentConfig::key,

StreamsGroupMetadataValue.AssignmentConfig::value

))

);

} else {

streamsGroup.setLastAssignmentConfigs(Collections.emptyMap());

}

group-coordinator/src/main/resources/common/message/StreamsGroupMetadataValue.json

Copilot · 2025-10-16T06:23:24Z

group-coordinator/src/main/resources/common/message/StreamsGroupMetadataValue.json

      "about": "The hash of all topics in the group." },
    { "name": "ValidatedTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 0, "default": -1, "type": "int32",
-      "about": "The topology epoch whose topics where validated to be present in a valid configuration in the metadata." }
+      "about": "The topology epoch whose topics where validated to be present in a valid configuration in the metadata." },


Corrected spelling of 'where' to 'were'.

Suggested change

"about": "The topology epoch whose topics where validated to be present in a valid configuration in the metadata." },

"about": "The topology epoch whose topics were validated to be present in a valid configuration in the metadata." },

...oordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupBuilder.java

lucasbru

Thanks for the updates, looks really good. Left 4 minor nits, but this is mostly ready to merge.

lucasbru · 2025-10-16T06:26:07Z

group-coordinator/src/main/resources/common/message/StreamsGroupMetadataValue.json

    { "name": "ValidatedTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 0, "default": -1, "type": "int32",
-      "about": "The topology epoch whose topics where validated to be present in a valid configuration in the metadata." }
+      "about": "The topology epoch whose topics where validated to be present in a valid configuration in the metadata." },
+    {


nit, we typically use 2 lines per field in these files

lucasbru · 2025-10-16T06:27:32Z

...est/java/org/apache/kafka/coordinator/group/streams/StreamsCoordinatorRecordHelpersTest.java

    }

    @Test
    public void testNewStreamsGroupEpochRecordNullGroupId() {


nit: should be called testNewStreamsGroupMetadataRecord (sorry, not your change, but while you are at it...)

lucasbru · 2025-10-16T06:30:42Z

...rc/main/java/org/apache/kafka/coordinator/group/streams/StreamsCoordinatorRecordHelpers.java

+        Objects.requireNonNull(assignmentConfigs, "assignmentConfigs should not be null here");
+
+        List<StreamsGroupMetadataValue.AssignmentConfig> assignmentConfigList = new ArrayList<>();
+        if (!assignmentConfigs.isEmpty()) {


nit: I think it's easier to read without this if

lucasbru · 2025-10-16T06:31:25Z

group-coordinator/src/main/resources/common/message/StreamsGroupMetadataValue.json

+      "nullableVersions": "0+",
+      "tag": 1,
+      "default": null,
+      "about": "Assignment configuration parameters as key-value pairs."


nit: "Last used assignment configuration..."?

lucasbru

Actually, yes. I missed that we are not handling null in replay.

lucasbru

LGTM, thanks!

lucliu1108 added 6 commits October 13, 2025 14:21

add group metadata config

4203e78

revise tests

1416ba2

add default config value to avoid unexpected epoch bump

7e85d2f

remove unnecessary spacing

d26b635

revise

e8a83b6

add test

9004010

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

lucliu1108 marked this pull request as ready for review October 14, 2025 20:39

resolve merge conflict

1291a93

lucasbru reviewed Oct 15, 2025

View reviewed changes

lucasbru self-assigned this Oct 15, 2025

lucasbru requested a review from Copilot October 15, 2025 09:32

lucasbru added streams KIP-1071 PRs related to KIP-1071 ci-approved and removed streams labels Oct 15, 2025

Copilot AI reviewed Oct 15, 2025

View reviewed changes

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java Outdated Show resolved Hide resolved

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java Outdated Show resolved Hide resolved

lucliu1108 added 2 commits October 15, 2025 16:45

revise

d4bbf61

adjust import order

b0560b6

github-actions bot removed the triage PRs from the community label Oct 16, 2025

lucliu1108 requested a review from lucasbru October 16, 2025 04:09

lucasbru requested a review from Copilot October 16, 2025 06:21

Copilot AI reviewed Oct 16, 2025

View reviewed changes

lucasbru approved these changes Oct 16, 2025

View reviewed changes

lucasbru requested changes Oct 16, 2025

View reviewed changes

lucliu1108 added 2 commits October 16, 2025 11:10

revise

2f50d5d

revise typo

a323bf5

lucliu1108 requested a review from lucasbru October 16, 2025 16:18

address copilot comments

b69b89c

lucasbru approved these changes Oct 17, 2025

View reviewed changes

lucasbru merged commit 36f17cf into apache:trunk Oct 17, 2025
21 checks passed

	"about": "The topology epoch whose topics where validated to be present in a valid configuration in the metadata." },
	"about": "The topology epoch whose topics were validated to be present in a valid configuration in the metadata." },

KAFKA-19765: Store the last used assignment configuration in the group metadata #20702

KAFKA-19765: Store the last used assignment configuration in the group metadata #20702

Conversation

lucliu1108 commented Oct 14, 2025 • edited by lucasbru Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

lucasbru Oct 15, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Uh oh!

Copilot AI Oct 16, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Copilot AI Oct 16, 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!

lucasbru Oct 16, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru Oct 16, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru Oct 16, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru Oct 16, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

lucliu1108 commented Oct 14, 2025 •

edited by lucasbru

Loading