KAFKA-20790: Promote streams assignor API to public module by gabriellefu · Pull Request #22788 · apache/kafka

gabriellefu · 2026-07-08T21:35:00Z

Part of KIP-1357.

This PR adds the newly defined public API as defined in KIP-1357 to
support customer broker side "streams" task assignors. It also
refactors the existing runtime code and StickyTaskAssignor to
use/implement the new public APIs.

org.apache.kafka.coordinator.group.api.assignor.streams (in
group-coordinator-api), all annotated with @InterfaceAudience.Public
and @InterfaceStability.Evolving.

Also fixes a latent bug: the old code incorrectly passed the old target
assignment into the assignor. This PR now ensures that we correctly pass
the current assignment instead.

Reviewers: Matthias J. Sax matthias@confluent.io, Sean Quah
squah@confluent.io

mjsax

Made a pass... Curious to hear from others.

mjsax · 2026-07-14T01:09:09Z

@@ -24,6 +27,8 @@
 *
 * @param members The member assignments keyed by member ID.
 */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
 public record GroupAssignment(Map<String, MemberAssignment> members) {


The KIP defines this as a class, not a record. -- Given that this is an output/return type, not an input to the assignor, it might actually be ok to keep as record, and we could also update the KIP?

\cc @lianetm for input

Thanks, I will update my kip and make it into a record.

\cc @squah-confluent

I didn't follow the KIP too closely but I think the reason we went with a class is that it's easier to evolve in the future. When adding a new field, we can add a constructor overload and initialize it with a sensible default, whereas a record will require constructor changes and break any code that tries to use the old constructor signature.

so maybe it's easier to maintain for future evolving as a record? I can change it back and adding the corresponding code today

When adding a new field, we can add a constructor overload and initialize it with a sensible default, whereas a record will require constructor changes and break any code that tries to use the old constructor signature.

record still allows to add an overloaded constructor... it's still true, that overall record is less flexible than class though, so I also don't have any objections to change all record types in public API to interface or class.

…get the actual warmuptask

gabriellefu · 2026-07-21T02:28:29Z

Have updated base on your comment, plz take a look. thanks! @mjsax

mjsax

Thanks Gabby. Made a pass. 95% LGTM.

mjsax · 2026-07-22T00:13:35Z

            member.instanceId(),
            member.rackId(),
            targetAssignment.activeTasks(),
            targetAssignment.standbyTasks(),


It seems odd to me, that we are passing the targetAssignment data here... This sounds like a bug to me? I believe we should pass in "current assignment" instead.

\cc @squah-confluent -- looking into code for KIP-848, it seems we are also passing the target assignment. Wondering why?

// Prepare the member spec for all members. members.forEach((memberId, member) -> memberSpecs.put(memberId, newMemberSubscriptionAndAssignment( member, targetAssignment.getOrDefault(memberId, Assignment.EMPTY), topicResolver )) );

Looking into StickyTaskAssignor we use it in initialize() to compute "prev active tasks" and "prev standby tasks" but if we did not reconcile to the target assignment yet, we would be "sticky" for no reason...

It's intentional and I don't recall the reasoning for it. It does lead to some weirdness, such as KAFKA-19405. I think we should discuss with @dajac.

…he/kafka/coordinator/group/api/streams/assignor/MemberAssignmentState.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

# Conflicts: # group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignmentState.java

…the assigned task

gabriellefu · 2026-07-23T23:25:54Z

            );
    }

-    private static List<StreamsGroupDescribeResponseData.TaskOffset> taskOffsetsFromMap(Map<TaskId, Long> offsets) {


change here cause the taskoffset has been changed from Map<TaskId, Long> to Map<String, Map<Integer, Long>>

mjsax · 2026-07-24T01:13:22Z

+                copyTasks(newMemberAssignment.activeTasks()),
+                copyTasks(newMemberAssignment.standbyTasks()),
+                // Warm-up tasks are not assigned by the assignor; they are decided during reconciliation.
+                Map.of()


Suggested change

Map.of()

new HashMap()

Map.of() return an unmodifiable map. If we make the extra effort to deep copy active and standby tasks to ensure it's modifiable, we should also ensure it for warmup tasks?

mjsax

Left on minor comment, but we can also address in a follow up PR. Plan to merge after CI passed.

) Part of KIP-1357. This PR adds the newly defined public API as defined in KIP-1357 to support customer broker side "streams" task assignors. It also refactors the existing runtime code and StickyTaskAssignor to use/implement the new public APIs. `org.apache.kafka.coordinator.group.api.assignor.streams` (in `group-coordinator-api`), all annotated with `@InterfaceAudience.Public` and `@InterfaceStability.Evolving`. Also fixes a latent bug: the old code incorrectly passed the old target assignment into the assignor. This PR now ensures that we correctly pass the current assignment instead. Reviewers: Matthias J. Sax <matthias@confluent.io>, Sean Quah <squah@confluent.io>

chia7712 · 2026-08-13T08:27:33Z

+    public GroupAssignment(
+        Map<String, MemberAssignment> members
+    ) {
+        this.members = Objects.requireNonNull(members);


Excuse me, it seems most public interfaces introduced by KIP-1357 use immutable collections, but this one does not. Would you mind sharing the reasoning with me?

@gabriellefu -- Seems we missed this one. GroupAssignment is a return type from the assignor to the broker, so making a defensive copy using Map.copyOf(...) seems to be the right thing to do.

The output side is currently guarded in TargetAssignmentBuilder, as TargetAssignmentBuilder#newMemberAssignment will deep copy the assignor's output using copyTasks(), so the server never holds a reference the assignor could mutate.

I can do the change in a follow-up pr to make the api more aligned

updated in this pr, thanks #23165

adding the new api

da33159

github-actions Bot added triage PRs from the community performance group-coordinator labels Jul 8, 2026

gabriellefu added 2 commits July 8, 2026 16:11

reshape the api and fix the bugs

1a44a8b

clean up the extra code

d88e439

gabriellefu changed the title ~~KAFKA-20683: Add broker side custom assignors for "streams" groups~~ KAFKA-20790: Add broker side custom assignors for "streams" groups Jul 8, 2026

gabriellefu changed the title ~~KAFKA-20790: Add broker side custom assignors for "streams" groups~~ KAFKA-20790: Promote streams assignor API to public module Jul 9, 2026

rename the api

d7f6a4a

mjsax reviewed Jul 14, 2026

View reviewed changes

github-actions Bot removed the triage PRs from the community label Jul 14, 2026

gabriellefu added 12 commits July 16, 2026 13:45

Merge branch 'trunk' into kip1357

dc38322

change taskid to interface

316278d

adding comment about keep taskoffset

664ce5e

nit

034a7aa

mover compareto to taskidimpl

5fedce1

rename some param

9c6ac46

clean requirement logic

488e7cb

revert taskid

91f106e

rename taskidimpl back

8d3bd5d

rename memberSpec to membermetadata

aff0de5

clean the newline

29f7c13

wire memberassigmentstate to member.assignedTasks().warmupTasks() to …

6750fbb

…get the actual warmuptask

gabriellefu requested a review from mjsax July 20, 2026 16:41

update package name to o.a.k.coordinator.group.api.streams.assignor

fbdd529

mjsax reviewed Jul 22, 2026

View reviewed changes

gabriellefu and others added 3 commits July 22, 2026 00:03

Update group-coordinator/group-coordinator-api/src/main/java/org/apac…

e887d50

…he/kafka/coordinator/group/api/streams/assignor/MemberAssignmentState.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

change in comment

4104e01

# Conflicts: # group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignmentState.java

ensure active/standby/warmup task are the current assignmed task not …

cd6f5aa

…the assigned task

gabriellefu added 9 commits July 22, 2026 14:06

clear unnessary assignmentstateimpl

e936484

change work to current assignment instead of target assignment

0fe2658

reformatting function call

5f5da62

change group assignment as a record

8b1614d

delete the immutable output but copy it

67148c6

reordering MemberMetadataAndAssignmentImpl

fa5c6a4

update output to class type instead of interface

ccd8548

change name to MemberMetadataAndStateImpl

f27dd07

change name to be consistent

9650503

gabriellefu requested a review from mjsax July 23, 2026 02:33

gabriellefu added 3 commits July 22, 2026 22:58

Merge branch 'trunk' into kip1357

8864340

fix the change in member offset

2eac92a

more compatiable code for change of MemberTaskOffsets

80c5772

gabriellefu commented Jul 23, 2026

View reviewed changes

mjsax reviewed Jul 24, 2026

View reviewed changes

mjsax added ci-approved streams kip Requires or implements a KIP KIP-1071 PRs related to KIP-1071 labels Jul 24, 2026

mjsax approved these changes Jul 24, 2026

View reviewed changes

mjsax merged commit e51258d into apache:trunk Jul 24, 2026
52 of 55 checks passed

chia7712 reviewed Aug 13, 2026

View reviewed changes

Uh oh!

Conversation

gabriellefu commented Jul 8, 2026 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

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!

squah-confluent Jul 22, 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!

gabriellefu commented Jul 21, 2026

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!

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!

Uh oh!

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!

mjsax left a comment

Choose a reason for hiding this comment

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

gabriellefu commented Jul 8, 2026 •

edited by github-actions Bot

Loading

squah-confluent Jul 22, 2026 •

edited

Loading