KAFKA-19779: Add per-partition epoch validation to streams groups [4/N] #20760

lucasbru · 2025-10-23T15:02:19Z

This change enhances the offset commit validation logic in streams
groups to validate against per-partition assignment epochs. When a
member attempts to commit offsets with an older member epoch, the logic
now validates that the epoch is not older than the assignment epoch for
each individual partition being committed.

The implementation adds a new createAssignmentEpochValidator method
that creates partition-level validators, checking each partition against
its assignment epoch from either assigned tasks or tasks pending
revocation.

We extend the SmokeTestDriverIntegrationTest to detect if we have
processed more records than needed, which, in this restricted scenario,
should only happen when offset commits are failing.

We re-enable the previously flaky test in EosIntegrationTest, which
failed due to previously failing offset commits.

Both tests have been run 100x in their streams protocol variation to
validate that they are not flaky anymore.

Reviewers: David Jacot [email protected], Matthias J. Sax
[email protected]

lucasbru · 2025-10-23T15:05:51Z

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

+        return (topicName, topicId, partitionId) -> {
+            final StreamsGroupTopologyValue.Subtopology subtopology = streamsTopology.sourceTopicMap().get(topicName);
+            if (subtopology == null) {
+                throw new StaleMemberEpochException("Topic " + topicName + " is not in the topology.");


This case is actually impossible right now because we do not allow updating the topology yet. But I think this would be the correct behavior once we allow changing the topology: We are trying to commit for a subtopology that does not exist anymore, so we should fence the member.

lucasbru · 2025-10-23T15:07:59Z

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

+    ) {
+        // Retrieve topology once for all partitions - not per partition!
+        final StreamsTopology streamsTopology = topology.get().orElseThrow(() ->
+            new StaleMemberEpochException("Topology is not available for offset commit validation."));


We do not allow removing the topology, so I think this may almost impossible. We'd have to recreate the group of the same name, and get the same member ID back to reach this point. If that would ever happen, I think fencing the member would be okay.

Copilot

Pull Request Overview

This PR enhances offset commit validation in streams groups by implementing per-partition epoch validation. Instead of just checking the member epoch, the system now validates that each partition being committed has an assignment epoch that is not newer than the commit request's epoch, enabling proper fencing of zombie commit requests.

Key changes:

Introduction of TasksTupleWithEpochs to track assignment epochs per partition for active tasks
Addition of createAssignmentEpochValidator method for partition-level validation
Extension of SmokeTestDriverIntegrationTest to detect excessive record processing
Re-enabling of previously flaky EosIntegrationTest

Reviewed Changes

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

Show a summary per file

File	Description
`TasksTupleWithEpochs.java`	New class storing active tasks with per-partition assignment epochs
`TasksTuple.java`	Updated to work with `TasksTupleWithEpochs` for containment checks
`StreamsGroup.java`	Implements per-partition epoch validation via `createAssignmentEpochValidator`
`StreamsGroupMember.java`	Changed to use `TasksTupleWithEpochs` for assigned tasks
`CurrentAssignmentBuilder.java`	Updated to preserve and assign epochs when building assignments
`StreamsCoordinatorRecordHelpers.java`	Serialization support for assignment epochs
`StreamsGroupCurrentMemberAssignmentValue.json`	Added `AssignmentEpochs` field to schema
`SmokeTestClient.java`	Tracks total data records processed
`SmokeTestDriverIntegrationTest.java`	Validates no excessive record reprocessing occurs
`EosIntegrationTest.java`	Removed `@Flaky` annotation from test
Various test files	Updated to work with epoch-aware task assignments

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

...ordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TasksTupleWithEpochs.java

...tests/src/test/java/org/apache/kafka/streams/integration/SmokeTestDriverIntegrationTest.java

mjsax · 2025-11-05T03:41:31Z

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

+        group.updateMember(new StreamsGroupMember.Builder("member-1")
+            .setMemberEpoch(5)
+            .setAssignedTasks(new TasksTupleWithEpochs(
+                Map.of("0", Map.of(0, 2, 1, 2)),


Seem we only test epoch quality? Should we use two different epochs for each partition to extend test coverage?

Good point. I made them different

mjsax · 2025-11-05T03:42:39Z

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

+            .build());
+
+        CommitPartitionValidator validator = group.validateOffsetCommit(
+            "member-1", "", 2, false, ApiKeys.OFFSET_COMMIT.latestVersion());


If we set member epoch to 5 above, should we also pass in 5 in here for the commit?

Well, with the relaxed check, it should just be smaller I guess, but larger or equals than the assignment epoch.

Yes, exactly.

mjsax · 2025-11-05T03:46:24Z

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

+            "member-1", "", 1, false, ApiKeys.OFFSET_COMMIT.latestVersion());
+
+        // Partition 0 assigned with epoch 2, received epoch 1 should throw
+        assertThrows(StaleMemberEpochException.class, () ->


Seem this check is somewhat redundant to testValidateOffsetCommitWithOlderEpoch from above?

You are right. This is not really what we are testing in this test. Removed this assertion

mjsax · 2025-11-05T03:49:50Z

...-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java

+                Map.of(), Map.of()))
+            .build());
+
+        // Commit with member epoch 3 should fail (3 < assignment epochs 5 and 8)


We should even fail for epoch 7 (< 8) right?

Good point. It's a better test if only one partition fails the check.

mjsax · 2025-11-05T03:52:03Z

...-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java

+    }
+
+    @Test
+    public void testStreamsGroupOffsetCommitWithOlderMemberEpochValidAssignments() {


Wondering to what extend this test, and the above testStreamsGroupOffsetCommitWithAssignmentEpochValid contain each other? For both cases, we expect success, what means member epoch > every assignment epoch.

What increase test coverage to we get by having both tests?

Yes. Removed this test.

mjsax

This lock overall GTM.

lucasbru · 2025-11-05T13:10:04Z

@mjsax Comments addressed and PR rebased

This commit enhances the offset commit validation logic in streams groups to validate against per-partition assignment epochs. When a member attempts to commit offsets with an older member epoch, the logic now validates that the epoch is not older than the assignment epoch for each individual partition being committed. The implementation adds a new `createAssignmentEpochValidator` method that creates partition-level validators, checking each partition against its assignment epoch from either assigned tasks or tasks pending revocation. We extend the SmokeTestDriverIntegrationTest to detect if we have processed more records than needed, which, in this restricted scenario, should only happen when offset commits are failing. We re-enable the previously flaky test in EosIntegrationTest, which failed due to previously failing offset commits. Both tests have been run 100x in there streams protocol variation to validate that they are not flaky anymore.

…onTest.shouldNotViolateEosIfOneTaskFails` as flaky (apache#20733)" This reverts commit 8dfaedd.

dajac

lgtm

github-actions bot added streams build Gradle build or GitHub Actions group-coordinator labels Oct 23, 2025

lucasbru commented Oct 23, 2025

View reviewed changes

lucasbru requested a review from Copilot October 23, 2025 15:08

Copilot AI reviewed Oct 23, 2025

View reviewed changes

...ordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TasksTupleWithEpochs.java Show resolved Hide resolved

...tests/src/test/java/org/apache/kafka/streams/integration/SmokeTestDriverIntegrationTest.java Outdated Show resolved Hide resolved

lucasbru changed the title ~~KAFKA-19779: Add per-partition epoch validation to streams groups [5/N]~~ KAFKA-19779: Add per-partition epoch validation to streams groups [4/N] Oct 23, 2025

mjsax reviewed Nov 5, 2025

View reviewed changes

lucasbru force-pushed the KAFKA-19779-4 branch from fe84e73 to f270733 Compare November 5, 2025 13:11

github-actions bot added the performance label Nov 5, 2025

lucasbru and others added 3 commits November 6, 2025 11:04

address comments

5876279

checkstyle

3444f13

lucasbru force-pushed the KAFKA-19779-4 branch from 169e5e4 to 3444f13 Compare November 6, 2025 10:06

lucasbru marked this pull request as ready for review November 6, 2025 10:06

Revert "MINOR: Mark`org.apache.kafka.streams.integration.EosIntegrati…

bdecaf8

…onTest.shouldNotViolateEosIfOneTaskFails` as flaky (apache#20733)" This reverts commit 8dfaedd.

dajac approved these changes Nov 6, 2025

View reviewed changes

lucasbru merged commit d5d9892 into apache:trunk Nov 6, 2025
23 checks passed

KAFKA-19779: Add per-partition epoch validation to streams groups [4/N] #20760

KAFKA-19779: Add per-partition epoch validation to streams groups [4/N] #20760

Conversation

lucasbru 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

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!

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!

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!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

lucasbru commented Nov 5, 2025

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

lucasbru commented Oct 23, 2025 •

edited by github-actions bot

Loading