KAFKA-20066: Implement KIP-1251: Assignment epochs for consumer groups [3/N] by lucliu1108 · Pull Request #21692 · apache/kafka

lucliu1108 · 2026-03-09T18:05:50Z

Summary

This PR changes the offset commit fencing logic, as a follow-up to
#21558 change on assignment
structure to include epoch information.

It 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.

Reviewers: Sean Quah squah@confluent.io, Lucas Brutschy
lbrutschy@confluent.io

Copilot

Pull request overview

Updates consumer-group offset commit fencing to use per-partition assignment epochs (per KIP-1251) so that modern-protocol consumers aren’t falsely fenced when their member epoch lags behind the broker’s current epoch.

Changes:

Relax modern consumer-group OffsetCommit validation from strict clientEpoch == brokerEpoch to per-partition assignmentEpoch <= clientEpoch <= brokerEpoch.
Resolve Uuid.ZERO_UUID topic IDs from coordinator metadata during OffsetCommit validation to support assignment-epoch checks.
Add/adjust unit tests covering assignment-epoch validation across assigned, pending-revocation, and unassigned partitions.

Reviewed changes

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

Show a summary per file

File	Description
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java	Implements per-partition assignment-epoch validation for modern-protocol offset commits when the client epoch is older than the broker epoch.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java	Resolves `ZERO_UUID` topic IDs from metadata before per-partition validation during OffsetCommit handling.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupTest.java	Adds parameterized tests for relaxed offset commit validation using assignment epochs.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMemberTest.java	Adds tests verifying assigned vs pending-revocation epoch lookup behavior.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java	Adds a regression test ensuring `ZERO_UUID` topic IDs are resolved for assignment-epoch validation.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java	Updates test fixtures to use assignment-with-epochs structure.

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

Copilot · 2026-03-09T20:28:57Z

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

+        return (topicName, topicId, partitionId) -> {
+            // Search for the partition in the assigned partitions, then in partitions pending revocation.
+            Integer assignmentEpoch = member.assignmentEpoch(topicId, partitionId);
+            if (assignmentEpoch == null) {
+                assignmentEpoch = member.pendingRevocationEpoch(topicId, partitionId);
+            }
+
+            if (assignmentEpoch == null) {
+                throw new StaleMemberEpochException(String.format(
+                    "Partition %s-%d is not assigned or pending revocation for member.",
+                    topicName, partitionId));
+            }
+
+            if (receivedMemberEpoch < assignmentEpoch) {
+                throw new StaleMemberEpochException(
+                    String.format("The received member epoch %d is older than the assignment epoch %d for partition %s-%d.",


createAssignmentEpochValidator relies on a non-ZERO_UUID topicId to find assignment epochs (it indexes member assignments by topic ID). However, OffsetMetadataManager.commitTransactionalOffset(...) currently invokes validator.validate(...) with Uuid.ZERO_UUID for TxnOffsetCommit requests, which will cause assignment-epoch validation to fail as "not assigned" whenever the member epoch is older than the broker epoch. To keep behavior consistent with the non-transactional OffsetCommit path, resolve the topic ID from metadata (topic name -> id) for TxnOffsetCommit as well, and pass the resolved ID into the validator.

Copilot · 2026-03-09T20:28:58Z

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

                    // Validate commit per-partition
                    validator.validate(
                        topic.name(),
-                        topic.topicId(),
+                        resolvedTopicId,
                        partition.partitionIndex()
                    );


CommitPartitionValidator.validate(...) can now throw StaleMemberEpochException on a per-partition basis (assignment-epoch checks). In commitOffset, that exception currently bubbles out of the per-partition loop, causing the whole OffsetCommit request to fail (and typically returning the same error for all partitions) instead of returning per-partition error codes and still committing valid partitions. Catch StaleMemberEpochException around validator.validate(...), set the partition error (e.g., Errors.STALE_MEMBER_EPOCH), and continue without appending a record for that partition.

Copilot · 2026-03-09T20:28:58Z

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

+                    // Resolve topic ID if it's ZERO_UUID
+                    Uuid resolvedTopicId = topic.topicId();
+                    if (resolvedTopicId.equals(Uuid.ZERO_UUID)) {
+                        resolvedTopicId = groupMetadataManager.image()
+                            .topicMetadata(topic.name())
+                            .map(CoordinatorMetadataImage.TopicMetadata::id)
+                            .orElse(Uuid.ZERO_UUID);
+                    }


Topic ID resolution for Uuid.ZERO_UUID is done inside the partition loop, but the resolved value is identical for every partition in the topic. Resolve once per topic (before iterating partitions) to avoid redundant metadata lookups and make the control flow clearer.

Copilot · 2026-03-09T20:28:58Z

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

+                    // Resolve topic ID if it's ZERO_UUID
+                    Uuid resolvedTopicId = topic.topicId();
+                    if (resolvedTopicId.equals(Uuid.ZERO_UUID)) {
+                        resolvedTopicId = groupMetadataManager.image()
+                            .topicMetadata(topic.name())
+                            .map(CoordinatorMetadataImage.TopicMetadata::id)
+                            .orElse(Uuid.ZERO_UUID);
+                    }
                    // Validate commit per-partition
                    validator.validate(
                        topic.name(),
-                        topic.topicId(),
+                        resolvedTopicId,
                        partition.partitionIndex()
                    );



resolvedTopicId is used for validation, but below this block the code still logs, responds, and persists offsets using topic.topicId() from the request. When the request uses ZERO_UUID, this will keep writing legacy records with ZERO_UUID even though the coordinator can resolve the real topic ID, weakening topic-id mismatch detection in offset fetch/delete logic. Consider using resolvedTopicId consistently for the stored OffsetAndMetadata (and potentially the response/log fields) once it has been resolved.

lucasbru

Hi @lucliu1108, thanks for the PR! I left three comments

cc @squah-confluent

lucasbru · 2026-03-10T09:47:52Z

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

+        return (topicName, topicId, partitionId) -> {
+            // Resolve topic ID if it's ZERO_UUID (for older API versions without topic ID).
+            Uuid resolvedTopicId = topicId;
+            if (topicId.equals(Uuid.ZERO_UUID) && metadataImageSupplier != null) {


Why do we need this null check?

lucasbru · 2026-03-10T09:50:09Z

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

+    /**
+     * Creates a validator that checks if the received member epoch is valid for each partition's assignment epoch.
+     * A commit is rejected if the partition is not assigned to the member
+     * or if the received client-side epoch is older than the partition's assignment epoch(KIP-1251).


Suggested change

* or if the received client-side epoch is older than the partition's assignment epoch(KIP-1251).

* or if the received client-side epoch is older than the partition's assignment epoch (KIP-1251).

lucasbru · 2026-03-10T10:06:09Z

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

+    /**
+     * The supplier for the metadata image, used to resolve topic names to IDs.
+     */
+    private final Supplier<CoordinatorMetadataImage> metadataImageSupplier;


Not sure I like this. There are two problems with this

we need to re-resolve each topic id for each partition, causing performance overhead.

we complicate the ConsumerGroup object quit a bit.

How about we already resolve topic IDs in OffsetMetadataManager.commitOffset() before calling validator.validate()?

Uuid resolvedTopicId = topic.topicId(); if (resolvedTopicId.equals(Uuid.ZERO_UUID)) { resolvedTopicId = groupMetadataManager.image() .topicMetadata(topic.name()) .map(CoordinatorMetadataImage.TopicMetadata::id) .orElse(Uuid.ZERO_UUID); }

squah-confluent

Thanks for the PR!

squah-confluent · 2026-03-11T05:09:43Z

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

+        // For members using the consumer protocol, the epoch must either match the last epoch sent
+        // in a heartbeat or be greater than or equal to the partition's assignment epoch.


This comment looks misplaced.

Suggested change

// For members using the consumer protocol, the epoch must either match the last epoch sent

// in a heartbeat or be greater than or equal to the partition's assignment epoch.

// For members using the classic protocol, the epoch must match the last epoch sent

// in a heartbeat.

squah-confluent · 2026-03-11T05:10:28Z

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

+            return CommitPartitionValidator.NO_OP;
+        }
+
+        // For members using the consumer protocol


Suggested change

// For members using the consumer protocol

// For members using the consumer protocol, the epoch must either match the last epoch sent

// in a heartbeat or be greater than or equal to the partition's assignment epoch.

squah-confluent · 2026-03-11T05:10:41Z

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

+        }
+        if (memberEpoch > member.memberEpoch()) {


nit: newline

Suggested change

}

if (memberEpoch > member.memberEpoch()) {

}

if (memberEpoch > member.memberEpoch()) {

squah-confluent · 2026-03-11T05:12:58Z

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

+     * @param member              The consumer group member.
+     * @param receivedMemberEpoch The member epoch from the offset commit request.
+     * @return A validator that checks each partition's assignment epoch.


Could we re-use the streams javadoc wording?

Suggested change

* @param member The consumer group member.

* @param receivedMemberEpoch The member epoch from the offset commit request.

* @return A validator that checks each partition's assignment epoch.

* @param member The member whose assignments are being validated.

* @param receivedMemberEpoch The received member epoch.

* @return A validator for per-partition validation.

squah-confluent · 2026-03-11T05:16:56Z

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


                    final OffsetAndMetadata offsetAndMetadata = OffsetAndMetadata.fromRequest(
-                        topic.topicId(),
+                        resolvedTopicId,


We shouldn't bake the topic id when the offset commit specifies a topic name only. This impacts behavior when topics are recreated.

Suggested change

resolvedTopicId,

topic.topicId(),

squah-confluent · 2026-03-11T06:27:37Z

...ator/src/test/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupTest.java

+        int partitionId = 0;
+        int memberEpoch = 10;
+        int assignmentEpoch = 7;
+        boolean isTransactional = false;


Since the main driver for the KIP was transactional offset commits, it's best that we test both transactional and non-transactional offset commits.

squah-confluent · 2026-03-11T06:30:19Z

...ator/src/test/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupTest.java

+                "member-id", "", assignmentEpoch, isTransactional, version
+            );
+            StaleMemberEpochException ex = assertThrows(StaleMemberEpochException.class, () ->
+                validator.validate(unassignedTopicName, unassignedTopicId, partitionId));


Can we test both an unassigned topic and unassigned partition?

squah-confluent · 2026-03-11T06:33:49Z

...ator/src/test/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupTest.java

+            assertThrows(UnsupportedVersionException.class, () ->
+                group.validateOffsetCommit("member-id", "", memberEpoch, isTransactional, version));
+        }
+        // client epoch (11) > broker epoch (10) - exception thrown directly from validateOffsetCommit


missing newline

Suggested change

// client epoch (11) > broker epoch (10) - exception thrown directly from validateOffsetCommit

// client epoch (11) > broker epoch (10) - exception thrown directly from validateOffsetCommit

squah-confluent · 2026-03-11T06:34:18Z

...ator/src/test/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupTest.java

+                mkTopicAssignmentWithEpochs(topicId, assignmentEpoch, partitionId)))
+            .build());
+
+        // client epoch = broker epoch


style nit: Could we write these as short sentences?

squah-confluent · 2026-03-11T06:36:18Z

...ator/src/test/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupTest.java

+            .setAssignedPartitions(mkAssignmentWithEpochs(
+                mkTopicAssignmentWithEpochs(assignedTopicId, assignmentEpoch, partitionId)))
+            .setPartitionsPendingRevocation(mkAssignmentWithEpochs(
+                    mkTopicAssignmentWithEpochs(assignedTopicId, assignmentEpoch, partitionId + 1)))


Suggested change

mkTopicAssignmentWithEpochs(assignedTopicId, assignmentEpoch, partitionId + 1)))

mkTopicAssignmentWithEpochs(assignedTopicId, assignmentEpoch, partitionId + 1)))

lucliu1108 added 3 commits March 9, 2026 13:30

fix merge mess

d54e7c5

resolve zero_uuid in offsetMetadataManager

817daf2

align method formats

b99ce8e

github-actions bot added triage PRs from the community group-coordinator labels Mar 9, 2026

lucasbru requested review from Copilot and lucasbru March 9, 2026 20:24

lucasbru self-assigned this Mar 9, 2026

Copilot started reviewing on behalf of lucasbru March 9, 2026 20:25 View session

Copilot AI reviewed Mar 9, 2026

View reviewed changes

github-actions bot removed the triage PRs from the community label Mar 10, 2026

lucliu1108 added 2 commits March 10, 2026 00:30

add supplier for metadata image to identy topicId in case of ZERO_UUID

8ad649e

adjust comments and add test for zero_uuid

df01cd4

lucasbru added the ci-approved label Mar 10, 2026

lucasbru reviewed Mar 10, 2026

View reviewed changes

lucliu1108 added 3 commits March 10, 2026 08:47

move resolve topic id step back to OffsetMetadataManager

b4831ea

minor fixes

15965fa

Merge branch 'trunk' into KAFKA-20066-3-new

c32ba9e

lucliu1108 requested a review from lucasbru March 10, 2026 14:52

squah-confluent reviewed Mar 11, 2026

View reviewed changes

	* or if the received client-side epoch is older than the partition's assignment epoch(KIP-1251).
	* or if the received client-side epoch is older than the partition's assignment epoch (KIP-1251).

		// For members using the consumer protocol, the epoch must either match the last epoch sent
		// in a heartbeat or be greater than or equal to the partition's assignment epoch.

	// client epoch (11) > broker epoch (10) - exception thrown directly from validateOffsetCommit

	// client epoch (11) > broker epoch (10) - exception thrown directly from validateOffsetCommit

	mkTopicAssignmentWithEpochs(assignedTopicId, assignmentEpoch, partitionId + 1)))
	mkTopicAssignmentWithEpochs(assignedTopicId, assignmentEpoch, partitionId + 1)))

Conversation

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

Uh oh!

Summary

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Mar 9, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 9, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 9, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 9, 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!

squah-confluent 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!

Choose a reason for hiding this comment

Uh oh!

squah-confluent Mar 11, 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!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

lucliu1108 commented Mar 9, 2026 •

edited by github-actions bot

Loading

squah-confluent Mar 11, 2026 •

edited

Loading