KAFKA-19779: Track assignment epochs in streams groups [3/N] #20730

lucasbru · 2025-10-20T11:05:16Z

This PR implements assignment epoch tracking for streams groups.

Fundamentally, this replaces the representation of current assignments
and pending revocations from Map<String, Set> to Map<String,
Map<Integer, Integer>> where the value of the inner map is the
assignment epoch. Target assignments, and the tasks reported as
currently owned, still use the former representation, since they do not
include assignment epochs. This change only applies to active tasks.
From TasksTuple, we create TasksTupleWithEpochs, which carries the
assignment epochs.

The core of the change is in CurrentAssignmentBuilder, which takes the
previous assignment epochs from assignedTasks and
tasksPendingRevocation and applies them to all tasks in the new
assignment. If a task in the new assignment was not previously assigned,
it gets the targetAssignmentEpoch, that the member transitions to.

There are a lot of mechanic follow-up changes to use
TasksTupleWithEpochs in-place of TasksTuple. In general, I tried to
follow the following strategy when adapting existing tests: - When
assignment epochs do no play a role in a test, we instantiate all tasks
with the same assignments epoch using TestAssignmentUtils. - The main
tests for correctly updating the assignment epoch are in
CurrentAssignmentBuilderTest.

Copilot

Pull Request Overview

This PR implements assignment epoch tracking for streams groups to support fencing of zombie commits. The core change replaces task assignments from Map<String, Set<Integer>> to Map<String, Map<Integer, Integer>> where the inner map value represents the assignment epoch for each partition. This change only applies to active tasks, as standby and warmup tasks don't commit offsets.

Key changes:

Introduces TasksTupleWithEpochs to carry assignment epochs alongside task assignments
Updates CurrentAssignmentBuilder to preserve assignment epochs when tasks remain assigned and applies the target assignment epoch to newly assigned tasks
Modifies the StreamsGroupCurrentMemberAssignmentValue record schema to include optional assignment epochs for active tasks

Reviewed Changes

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

Show a summary per file

File	Description
TasksTupleWithEpochs.java	New class to represent task assignments with epochs for active tasks
TasksTupleWithEpochsTest.java	Comprehensive test coverage for the new TasksTupleWithEpochs class
CurrentAssignmentBuilder.java	Updated to track and preserve assignment epochs through state transitions
CurrentAssignmentBuilderTest.java	Extended with tests for epoch assignment lookup paths
StreamsGroupMember.java	Changed to use TasksTupleWithEpochs instead of TasksTuple
StreamsCoordinatorRecordHelpers.java	Added conversion logic for tasks with epochs to records
StreamsGroup.java	Updated process ID tracking to work with epoched tasks
GroupMetadataManager.java	Modified validation and response building for epoched tasks
TaskAssignmentTestUtil.java	Added helper methods for creating tasks with epochs in tests
StreamsGroupCurrentMemberAssignmentValue.json	Schema updated with optional assignmentEpochs field
Various test files	Mechanical updates to use epoch-aware task creation utilities

Comments suppressed due to low confidence (2)

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

[nitpick] There's an empty line (16290) between the method chain continuation. While not incorrect, this inconsistent formatting reduces readability. Remove the blank line to maintain consistent method chaining style.

/*

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

[nitpick] The epoch parameter (11) is placed on a separate line, which is inconsistent with other similar calls in the file where the epoch appears on the same line as the method name. Consider moving it to line 18132 for consistency.

/*

_{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

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

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

mjsax

Did not go over test code yet.

...ordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TasksTupleWithEpochs.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 you comments @mjsax ! Answered / Addressed.

I also updated the code in CurrentAssingmentBuilder to be more direct / not create intermediate collections and work a lot more like the previous code.

Instead of using computeAssignmentDifference in standby tasks and warmup tasks, active tasks now use computeAssignmentDifferenceWithEpoch, which updates epochs correctly but works otherwise very similar. This is also the code that consumer groups can use.

lucasbru · 2025-10-22T12:33:46Z

While playing through an example, I realized I can simplify the logic - I was trying to preserve assignment epochs from tasks that were previously pending revocation, in case the task is reassigned. But we will get no tasks assigned while pending revocations, even not the task that we are currently pending revocation. So we will only get the task back after having confirmed revocations (and completing commits) for the task, so then it is fine to get a new assignment epoch in this case.

dajac

Thanks for the patch. I made a first pass on it and it looks pretty good to me overall. However, I am not too familiar with Streams' concepts. I left a few comments for consideration.

dajac · 2025-10-22T14:03:47Z

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

+                .setPartitions(partitions)
+                .setAssignmentEpochs(epochs));
+        });
+        taskIds.sort(Comparator.comparing(StreamsGroupCurrentMemberAssignmentValue.TaskIds::subtopologyId));


I guess that this is not strictly necessary but convenient for testing?

Yes, convenient for testing and I like the persistent records to be deterministic, but there is no strict reason why we must do this.

dajac · 2025-10-22T14:05:20Z

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

+     * This method creates a new map on each call. Consider using {@link #activeTasksWithEpochs()}
+     * directly when possible to avoid the conversion.


Is it possible to just get rid of the method if we don't recommend using it?

Good point. I removed it

dajac · 2025-10-22T14:06:25Z

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

            .collect(Collectors.toList());
    }

+    private List<StreamsGroupHeartbeatResponseData.TaskIds> createStreamsGroupHeartbeatResponseTaskIdsFromEpochs(


nit: static?

dajac · 2025-10-22T14:06:47Z

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

            // If the member comes with the previous epoch and has a subset of the current assignment partitions,
            // we accept it because the response with the bumped epoch may have been lost.
            if (receivedMemberEpoch != member.previousMemberEpoch()
-                || !areOwnedTasksContainedInAssignedTasks(ownedActiveTasks, member.assignedTasks().activeTasks())


Is areOwnedTasksContainedInAssignedTasks still used or could we remove it?

Yes, it's still used for standby tasks.

dajac · 2025-10-22T14:20:27Z

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

+     * - For tasks in resultAssignedTasks and resultTasksPendingRevocation, the epoch from currentAssignment is preserved.
+     * - For tasks in resultTasksPendingAssignment, the targetAssignmentEpoch is used.
+     */
+    private boolean computeAssignmentDifferenceWithEpoch(Map<String, Map<Integer, Integer>> currentAssignment,


Those new methods are basically copies of the previous ones. The only difference is that they operate on Map<String, Map<Integer, Integer>> vs Map<String, Set<Integer>>. This is a bit unfortunate but I guess that we don't really have the choice unless we use Map<String, Map<Integer, Integer>> everywhere.

Yeah, as I mentioned, the change for consumer groups will be cleaner. For streams we have the added complexity of having to track standby tasks and warmup tasks as well.

We could consider simplifying this code and tracking assignment epochs also for standby tasks and warmup tasks. I don't think there is much runtime overhead, so I'd be open to that. I just did not do it since it may become confusing that we are tracking assingment epochs for standby tasks but then never use them.

lucasbru

Thanks @mjsax and @dajac for the review comments. All addressed

lucasbru · 2025-10-23T11:03:30Z

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

+                .setPartitions(partitions)
+                .setAssignmentEpochs(epochs));
+        });
+        taskIds.sort(Comparator.comparing(StreamsGroupCurrentMemberAssignmentValue.TaskIds::subtopologyId));


Yes, convenient for testing and I like the persistent records to be deterministic, but there is no strict reason why we must do this.

lucasbru · 2025-10-23T11:04:45Z

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

            .collect(Collectors.toList());
    }

+    private List<StreamsGroupHeartbeatResponseData.TaskIds> createStreamsGroupHeartbeatResponseTaskIdsFromEpochs(


lucasbru · 2025-10-23T11:05:32Z

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

+     * This method creates a new map on each call. Consider using {@link #activeTasksWithEpochs()}
+     * directly when possible to avoid the conversion.


Good point. I removed it

lucasbru requested a review from Copilot October 20, 2025 11:05

github-actions bot added the group-coordinator label Oct 20, 2025

lucasbru requested review from dajac and mjsax October 20, 2025 11:05

Copilot AI reviewed Oct 20, 2025

View reviewed changes

lucasbru force-pushed the KAFKA-19779-3 branch 4 times, most recently from b1a708d to 6ca6534 Compare October 20, 2025 15:10

KAFKA-19779: Track assignment epochs to streams groups

af9ad43

lucasbru force-pushed the KAFKA-19779-3 branch from 6ca6534 to af9ad43 Compare October 20, 2025 19:31

mjsax added the streams label Oct 21, 2025

mjsax reviewed Oct 21, 2025

View reviewed changes

lucasbru changed the title ~~KAFKA-19779: Track assignment epochs to streams groups~~ KAFKA-19779: Track assignment epochs in streams groups Oct 21, 2025

updates

4e1b79f

lucasbru commented Oct 21, 2025

View reviewed changes

simplification

cd0bbde

dajac reviewed Oct 22, 2025

View reviewed changes

comments

40d2445

lucasbru commented Oct 23, 2025

View reviewed changes

lucasbru changed the title ~~KAFKA-19779: Track assignment epochs in streams groups~~ KAFKA-19779: Track assignment epochs in streams groups [3/N] Oct 24, 2025

		* This method creates a new map on each call. Consider using {@link #activeTasksWithEpochs()}
		* directly when possible to avoid the conversion.

KAFKA-19779: Track assignment epochs in streams groups [3/N] #20730

Are you sure you want to change the base?

KAFKA-19779: Track assignment epochs in streams groups [3/N] #20730

Conversation

lucasbru commented Oct 20, 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!

Uh oh!

Uh oh!

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

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 commented Oct 22, 2025

Uh oh!

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

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!

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!

3 participants

lucasbru commented Oct 20, 2025 •

edited by github-actions bot

Loading