KAFKA-20066: Assignment epochs for consumer groups by lucliu1108 · Pull Request #21508 · apache/kafka

lucliu1108 · 2026-02-18T16:04:39Z

Summary

This PR is an implementation of KIP-1251:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-1251%3A+Assignment+epochs+for+consumer+groups
This change introduces per-partition assignment epochs to relax the
strict member epoch validation for consumer group offset commits. When
receiving an offset commit request that includes the client-side member
epoch and a member ID, we previously require checking

clientEpoch == brokerEpoch

for a valid offset commit, which could lead to false fencing.

We now allow a relaxed offset commit check using an assignment epoch for
each assigned partition and each member,

assignmentEpoch <= clientEpoch <= brokerEpoch

This prevents false rejections of legitimate offset commits when a
member's epoch is bumped but the client hasn't received the update yet.

Changes

Add AssignmentEpochs field to TopicPartitions in
ConsumerGroupCurrentMemberAssignmentValue schema
Add assignedPartitionsWithEpochs and
partitionsPendingRevocationWithEpochs to ConsumerGroupMember
Update ConsumerGroup.validateOffsetCommit() with relaxed
validation logic
Add createAssignmentEpochValidator() for per-partition epoch
checks
Update GroupCoordinatorRecordHelpers to serialize assignment
epochs
Update GroupMetadataManager to reset assignment epoch when leaving
group for static members

Reviewers: @lucasbru

dajac · 2026-02-19T00:36:42Z

cc @squah-confluent

Copilot

Pull request overview

Implements KIP-1251 assignment epochs for modern consumer groups to relax offset-commit fencing by validating commits against per-partition assignment epochs rather than requiring a strict member-epoch match.

Changes:

Extends the ConsumerGroupCurrentMemberAssignmentValue schema to persist per-partition assignment epochs alongside assigned partitions.
Adds epoch-annotated assignment tracking to ConsumerGroupMember and serializes it via GroupCoordinatorRecordHelpers.
Updates ConsumerGroup.validateOffsetCommit() to allow assignmentEpoch <= clientEpoch <= brokerEpoch using a per-partition validator when epochs don’t strictly match.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
group-coordinator/src/main/resources/common/message/ConsumerGroupCurrentMemberAssignmentValue.json	Adds per-partition `AssignmentEpochs` field aligned with the partitions array.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java	Introduces epoch-annotated assignment maps + accessors and builder support (incl. reset to zero).
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java	Relaxes offset-commit epoch validation and adds per-partition assignment-epoch validator.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java	Resets assignment epochs for departing static members and clears pending-revocation epochs.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java	Serializes assignment epochs into coordinator records.

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

Copilot · 2026-02-19T12:16:30Z

...or/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java

+        public Builder setAssignedPartitionsWithEpochs(Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs) {
+            this.assignedPartitionsWithEpochs = assignedPartitionsWithEpochs;
+            this.assignedPartitions = assignedPartitionsWithEpochs.entrySet().stream()
+                .collect(Collectors.toMap(
+                    Map.Entry::getKey,
+                    e -> new HashSet<>(e.getValue().keySet())
+                ));
+            return this;
+        }
+
+        public Builder setPartitionsPendingRevocationWithEpochs(Map<Uuid, Map<Integer, Integer>> partitionsPendingRevocationWithEpochs) {
+            this.partitionsPendingRevocationWithEpochs = partitionsPendingRevocationWithEpochs;
+            this.partitionsPendingRevocation = partitionsPendingRevocationWithEpochs.entrySet().stream()
+                .collect(Collectors.toMap(
+                    Map.Entry::getKey,
+                    e -> new HashSet<>(e.getValue().keySet())
+                ));
+            return this;
+        }


setAssignedPartitionsWithEpochs / setPartitionsPendingRevocationWithEpochs keep the Set-based fields in sync, but the Builder still also has the older setAssignedPartitions(...) and setPartitionsPendingRevocation(...) setters. Given GroupCoordinatorRecordHelpers now serializes assignedPartitionsWithEpochs()/partitionsPendingRevocationWithEpochs(), any code path that uses the older setters (without also setting the epoch maps) can drop epoch information and even persist empty assignments. Consider enforcing invariants in build() (e.g., derive epoch maps when missing), or deprecating/removing the Set-based setters to prevent constructing an internally inconsistent ConsumerGroupMember.

Copilot · 2026-02-19T12:16:31Z

...rdinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java

                    .setMemberEpoch(member.memberEpoch())
                    .setPreviousMemberEpoch(member.previousMemberEpoch())
                    .setState(member.state().value())
-                    .setAssignedPartitions(toTopicPartitions(member.assignedPartitions()))
-                    .setPartitionsPendingRevocation(toTopicPartitions(member.partitionsPendingRevocation())),
+                    .setAssignedPartitions(toTopicPartitions(member.assignedPartitionsWithEpochs()))
+                    .setPartitionsPendingRevocation(toTopicPartitions(member.partitionsPendingRevocationWithEpochs())),


newConsumerGroupCurrentAssignmentRecord now exclusively serializes member.assignedPartitionsWithEpochs()/partitionsPendingRevocationWithEpochs(). If those maps are empty while the Set-based assignment fields are populated (possible via existing Builder setters), this will write records with empty assignments and effectively lose the current assignment on disk. Consider adding a defensive fallback (e.g., derive epoch maps from assignedPartitions()/partitionsPendingRevocation() with a sensible default epoch) or validate/throw if the member is internally inconsistent before serializing.

Copilot · 2026-02-19T12:16:31Z

...rdinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java

+        // For member using the classic protocol, use strict epoch validation.
+        if (member.useClassicProtocol()) {


Grammar in this comment is off: consider changing "For member using the classic protocol" to "For members using the classic protocol" (or "For a member using...") for clarity.

Copilot · 2026-02-19T12:16:31Z

...rdinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java

+        if (memberEpoch == member.memberEpoch()) {
+            return CommitPartitionValidator.NO_OP;
+        }
+        // Case 2:Client epoch > broker epoch, which is an invalid request


Minor comment formatting: add a space after the colon in "Case 2:Client".

Suggested change

// Case 2:Client epoch > broker epoch, which is an invalid request

// Case 2: Client epoch > broker epoch, which is an invalid request

Copilot · 2026-02-19T12:16:31Z

...rdinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java

+        // For member using the consumer protocol
+        // Case 1: Strict epoch match
+        if (memberEpoch == member.memberEpoch()) {
+            return CommitPartitionValidator.NO_OP;
+        }
+        // Case 2:Client epoch > broker epoch, which is an invalid request
+        if (memberEpoch > member.memberEpoch()) {
+            throw new StaleMemberEpochException(String.format("The received member epoch %d is larger than "
+                + "the expected member epoch %d.", memberEpoch, member.memberEpoch()));
+        }
+        return createAssignmentEpochValidator(member, memberEpoch);


The new relaxed offset-commit validation path (returning a per-partition validator when receivedMemberEpoch < broker member epoch) doesn’t appear to be covered by unit tests. It would be useful to add ConsumerGroupTest cases that (1) accept commits when assignmentEpoch <= receivedEpoch < brokerEpoch for assigned/pending-revocation partitions, and (2) reject commits when the partition isn’t assigned or when receivedEpoch < assignmentEpoch, to prevent regressions in the new behavior.

Copilot · 2026-02-19T12:16:32Z

...or/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java

 import java.util.ArrayList;
+import java.util.stream.Collectors;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;


Import order looks inconsistent with the rest of the module: java.util.stream.Collectors is currently placed between java.util.ArrayList and java.util.Collections. Other files keep java.util.* imports together and place java.util.stream.* afterwards (e.g., Utils.java:36-49). Please reorder these imports to match the established style (this may also be enforced by Checkstyle).

lucasbru

Thanks for the PR @lucliu1108 !

I left some comments. I think the biggest one to me is right now that we should not have two collections representing the assigned partitions in ConsumerGroupMember.

I guess this is meant as a draft, we will not be able to merge this without tests. Ideally, we merge it as a series of smaller PRs. But it can make sense to create one draft PR with everything first for the conceptual review, and then split it up into a couple smaller PRs in the final merge.

lucasbru · 2026-02-19T12:37:16Z

...or/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java

        private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
        private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
        private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+        private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();


Why are we adding assignedPartitionsWithEpochs in addition to assignedPartitions? This will get out of sync. I think we should just have the second, and replace all uses of assignedPartitions with assignedPartitionsWithEpochs.

I have removed all setters for assignedPartitions and partitionsPendingRevocation, but still keep the getters and as private fields to avoid redundant conversions in other methods that previously call these getters.

Since these private fields will be re-constructed every time the builder is called to set assignedPartitionsWithEpochs and partitionsRevocationsWithEpochs, the getters won't be async.

I think this makes sense - are there so many places that require assignedPartitions and partitionsPendingRevocation that we cannot convert the calling context to use the map instead of the set? Or do we require it because of the sharing with the share group?

I wonder if it could be worth implementing a view on the keySet to avoid the second collection here, as it's probably one of the larger collections in the GC.

For the assignedPartitions, it is part of the fields inherited from ModernGroupMember that ShareGroupMember also uses, so i left it there for the getter method.
Or another way is not to initialize it in the constructor (keep it null?) but maybe this is not good practice.

For the. partitionsPendingRevocation, there are 6 or 7 places it is directly called. Guess I could remove this method and do extra conversions for where its' called instead, or use a view.

lucasbru · 2026-02-19T12:44:31Z

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

        // We will write a member epoch of -2 for this departing static member.
+        // Assignment epochs are reset to 0 so when the static member rejoins, partitions
+        // are considered assigned from epoch 0 to the new member ID.
        ConsumerGroupMember leavingStaticMember = new ConsumerGroupMember.Builder(member)


It seems we also need to update the dynamic member leave code. We seem to set setAssignedPartitions there.

lucasbru · 2026-02-19T12:48:22Z

...rdinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java

+            // If the partition is not assigned to this member, reject.
+            if (assignmentEpoch == null) {
+                throw new StaleMemberEpochException(
+                    String.format("Partition %s-%d is not assigned to member %s.",


Above, we are allowing committing unassigned partitions if the "strict epoch match" passes, that is client-side member epoch = server-side member epoch.

Then this error message is probably not sufficient as a reason. We should also include that the client-side member epoch != server-side member epoch in this error.

I'm in favor of aligning with the streams implementation as much as possible. Streams has the same pattern and uses the message Task %s-%d is not assigned or pending revocation for member.. Shall we update the streams message too (in a separate PR maybe)?

Sound good!

…mutable

lucasbru

On a high level, this makes sense to me. I left a comment, but you can definitely work on the tests

lucasbru · 2026-02-20T20:14:14Z

...rdinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java

+        topicPartitionsWithEpochs.forEach((topicId, partitionEpochMap) -> {
+            List<Integer> partitionList = new ArrayList<>(partitionEpochMap.keySet());
+            List<Integer> epochList = partitionList.stream()
+                .map(partitionId -> partitionEpochMap.getOrDefault(partitionId, 0))


why getOrDefault and not just get

lucasbru · 2026-02-20T20:23:00Z

...or/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java

        private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
        private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
        private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+        private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();


I think this makes sense - are there so many places that require assignedPartitions and partitionsPendingRevocation that we cannot convert the calling context to use the map instead of the set? Or do we require it because of the sharing with the share group?

I wonder if it could be worth implementing a view on the keySet to avoid the second collection here, as it's probably one of the larger collections in the GC.

lucasbru · 2026-02-20T20:26:54Z

...c/main/java/org/apache/kafka/coordinator/group/modern/consumer/CurrentAssignmentBuilder.java

        Map<Uuid, Set<Integer>> newPartitionsPendingAssignment = new HashMap<>();

+        // Get existing epochs from member
+        Map<Uuid, Map<Integer, Integer>> existingAssignedEpochs = member.assignedPartitionsWithEpochs();


If an partition was pending revocation, and is added back to the target assignment of the member, it seems we want to keep the original assignment epoch. So we have to look at member.partitionPendingRevociations here as well, no?

lucliu1108 added 2 commits February 18, 2026 08:55

add assignment epoch basic functionality

6541f94

clean up

6d9578f

github-actions bot added triage PRs from the community group-coordinator labels Feb 18, 2026

add handling of static member when leaving the group

3d5cb68

lucasbru self-assigned this Feb 19, 2026

lucasbru requested review from Copilot and lucasbru and removed request for lucasbru February 19, 2026 12:09

Copilot started reviewing on behalf of lucasbru February 19, 2026 12:09 View session

Copilot AI reviewed Feb 19, 2026

View reviewed changes

lucasbru reviewed Feb 19, 2026

View reviewed changes

github-actions bot removed the triage PRs from the community label Feb 20, 2026

lucliu1108 added 5 commits February 19, 2026 23:56

revise all methods with the new assignmentWithEpoch

9f8857a

revert back partitionsPendingRevocation getter

d75a678

adjust some orders

1cb58a3

make getter for partitionsRendingRevocation and assignedpartitions im…

b608c70

…mutable

revise comments

e78dd69

lucliu1108 requested review from lucasbru and squah-confluent February 20, 2026 18:16

lucasbru reviewed Feb 20, 2026

View reviewed changes

lucasbru added the ci-approved label Feb 20, 2026

lucasbru reviewed Feb 20, 2026

View reviewed changes

		// For member using the classic protocol, use strict epoch validation.
		if (member.useClassicProtocol()) {

	// Case 2:Client epoch > broker epoch, which is an invalid request
	// Case 2: Client epoch > broker epoch, which is an invalid request

Comments

Conversation

lucliu1108 commented Feb 18, 2026 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Changes

Uh oh!

dajac commented Feb 19, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Feb 19, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 19, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 19, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 19, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 19, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 19, 2026

Choose a reason for hiding this comment

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lucliu1108 Feb 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

lucliu1108 commented Feb 18, 2026 •

edited by github-actions bot

Loading

lucliu1108 Feb 20, 2026 •

edited

Loading