KAFKA-20665: Add refiner building blocks by mjsax · Pull Request #23325 · apache/kafka

mjsax · 2026-09-01T09:15:59Z

Add building block code for the "streams" assignment refiner, that will
later be used in the overall refinement algorithm.

Reviewers: Lucas Brutschy lbrutschy@confluent.io

Add building block code for the "streams" assignment refiner, that will later be used in the overall refinement algorithm.

Copilot

🟡 Changes recommended

The new refiner building blocks silently overwrite duplicate task ownership in maps, which can make behavior non-deterministic and should be guarded with explicit validation.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Adds foundational implementation pieces for a future Kafka Streams assignment refiner in the group coordinator, plus supporting test refactors and new unit coverage for the building-block logic.

Changes:

Introduce a production TaskRole enum and migrate tests/utilities off the prior test-only enum.
Add AssignmentRefinerImpl with helper building blocks (current-assignment indexing, task-migration analysis, caught-up predicate), while keeping refine() as a stub for now.
Expand AssignmentRefinerTest with focused unit tests for the new helper logic.

File summaries

File	Description
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/StreamsGroupTestUtil.java	Switch test helper usage to production `TaskRole`.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/StreamsGroupMixedGroupMetadataManagerTest.java	Replace test-only `TaskRole` import with production `TaskRole`.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TaskAssignmentTestUtil.java	Remove test-local `TaskRole` enum; use production `TaskRole`.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java	Drop now-invalid nested-enum import after `TaskRole` promotion.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java	Drop now-invalid nested-enum import after `TaskRole` promotion.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsCoordinatorRecordHelpersTest.java	Drop now-invalid nested-enum import after `TaskRole` promotion.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilderTest.java	Drop now-invalid nested-enum import after `TaskRole` promotion.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerTest.java	Add extensive unit tests for `AssignmentRefinerImpl` building blocks.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java	Switch test usage to production `TaskRole`.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TaskRole.java	New production enum defining ACTIVE/STANDBY/WARMUP roles.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java	New refiner implementation scaffold + helper methods (not yet wired into `refine()`).
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefiner.java	Javadoc reflow/formatting for parameter docs and paragraphs.

Review details

Suppressed comments (1)

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:335

statefulActiveOwners uses owners.put(task, memberId), so if the target assignment (erroneously) contains the same active task under multiple members, the later entry silently wins. This makes the refiner’s decisions non-deterministic and harder to debug; consider rejecting duplicates with an explicit exception.

        final SortedMap<TaskId, String> owners = new TreeMap<>();
        targetAssignment.forEach((memberId, tasks) ->
            forEachStatefulTask(tasks.activeTasks(), subtopologies, task -> owners.put(task, memberId)));
        return owners;

Files reviewed: 12/12 changed files
Comments generated: 1
Review effort level: Lite

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

mjsax · 2026-09-05T06:15:12Z

+            forEachStatefulActiveTask(
+                member.assignedTasks().activeTasksWithEpochs(),
+                subtopologies,
+                task -> activeOwner.put(task, member.memberId())
+            );


Can't happen. We know that only one task can be ACTIVE, never two. Would be an assignor bug...

lucasbru · 2026-09-04T12:47:00Z

+            // deliberately not recorded as the task's active owner. Recording the member as the owner would make the
+            // case analysis try to keep the task there, undoing a hand-over that is already under way.
+            //
+            // The task does still occupy the member's process until the revocation completes, and that is what stops


Wouldn't this case be taken care of by the reconciler? I mean the reconciler makes sure that I do not assign a new task before the old task is revoked. Why do we need to handle this in two places?

You are right. This is over-engineered... Will remove

lucasbru · 2026-09-04T12:48:34Z

+            // a standby of the same task being placed there. The block applies to the whole process, not just this
+            // one member, so the process is what gets recorded.
+            forEachStatefulActiveTask(
+                member.tasksPendingRevocation().activeTasksWithEpochs(),


Why do we not care about standby tasks and warmup tasks pending revocation?

Not needed as per you comment above. We don't need to track pending revocations.

lucasbru · 2026-09-04T13:20:28Z

+     * the restore has not started, counts as not caught up. A slightly negative lag does count, because the offset is a
+     * position while the end offset is the last offset, so a fully restored task reports a lag of -1.
+     *
+     * @param memberTaskOffsets


So members without lag info are not handled here yet, right?

lucasbru · 2026-09-04T13:40:31Z

+        });
+    }
+
+    private static boolean isStateful(


this is a bit of misnomer because you can have stateful tasks that are not bakced by a log. So I sometimes wonder if it should be "LoggedStateful" or something

I guess you are technically correct, but is it a problem? From a GC POV, if a task is stateful w/o a changelog, we can only treat it the same as a stateless task because we cannot restore anything. So I don't see any advantage to introduce such a distinction.

lucasbru · 2026-09-04T13:41:57Z

+     * @param targetOwner
+     *        The member the task moves to.
+     */
+    record TaskGrant(


Wouldn't it be easier to also include the oldOwner in this struct, so that it's easier to apply the patches to the target assignment?

That's not necessary. For a TaskGrant there is no patch to be applied. We use the target assignment to apply patches, and if we grant a task, the target assignment already agrees -- it already assigns the granted task to the new owner, and revokes it from the old owner. We only need to patch the target assignment when we delay (ie stage) a migration.

mjsax added the KIP-1071 PRs related to KIP-1071 label Sep 1, 2026

github-actions Bot added the group-coordinator label Sep 1, 2026

mjsax force-pushed the kafka-20665-assignment-refiner branch from bf281be to ed2ce12 Compare September 1, 2026 19:02

KAFKA-20665: Add refiner building blocks

0f3ece1

Add building block code for the "streams" assignment refiner, that will later be used in the overall refinement algorithm.

mjsax force-pushed the kafka-20665-assignment-refiner branch from ed2ce12 to 0f3ece1 Compare September 3, 2026 07:00

lucasbru requested review from lucasbru and a lite review from Copilot and removed request for Copilot September 4, 2026 12:37

lucasbru self-assigned this Sep 4, 2026

Copilot started reviewing on behalf of lucasbru September 4, 2026 12:38 View session

Copilot AI reviewed Sep 4, 2026

View reviewed changes

lucasbru reviewed Sep 4, 2026

View reviewed changes

review comments -- also a few more improvements

220d67a

Navigation Menu

Uh oh!

KAFKA-20665: Add refiner building blocks - #23325

KAFKA-20665: Add refiner building blocks#23325
mjsax wants to merge 2 commits into
apache:trunkfrom
mjsax:kafka-20665-assignment-refiner

mjsax commented Sep 1, 2026 •

edited by github-actions Bot

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

mjsax Sep 5, 2026

Uh oh!

lucasbru Sep 4, 2026

Uh oh!

mjsax Sep 5, 2026

Uh oh!

lucasbru Sep 4, 2026

Uh oh!

mjsax Sep 5, 2026

Uh oh!

lucasbru Sep 4, 2026 •

edited

Loading

Uh oh!

mjsax Sep 5, 2026

Uh oh!

lucasbru Sep 4, 2026

Uh oh!

mjsax Sep 5, 2026

Uh oh!

lucasbru Sep 4, 2026

Uh oh!

mjsax Sep 5, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

mjsax commented Sep 1, 2026 • 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

🟡 Changes recommended

Suppressed comments (1)

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 Sep 4, 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!

3 participants

mjsax commented Sep 1, 2026 •

edited by github-actions Bot

Loading

lucasbru Sep 4, 2026 •

edited

Loading