Thanks Matthias, I will update my kip accordingly. Best, Gabriella On Wed, Jul 8, 2026 at 9:10 AM Matthias J. Sax <[email protected]> wrote:
> Thanks Gabriella, > > one last thing. While the assignor does not return any warmup tasks (ie, > it does not assign any) via `MemberAssignment`, we would still want to > pass in currently assigned warmup tasks `MemberAssignmentState`, so > `MemberAssignmentState` interface should also have a method > > Map<String, Set<Integer>> warmupTasks(); > > > > -Matthias > > > On 7/7/26 11:40 AM, Gabriella Fu via dev wrote: > > Thanks David, > > DJ_04: I will mention we will have some follow up work on this, but we > > won't implement it in this kip > > > > DJ_05: I will fix the typo, thanks > > > > Best, > > Gabriella > > > > On Tue, Jun 30, 2026 at 2:06 AM David Jacot <[email protected]> > wrote: > > > >> Hi Gabriella, > >> > >> Thanks for your answers. Overall, LGTM. I have a few minor comments. > >> > >> DJ_04: I still wonder whether we should move all the validation of the > >> dynamic configurations to the broker. It is an implementation detail > >> in the end but let's please consider it. Having the configurations > >> validated in two places will be confusing. It is also helpful for the > >> other dynamic configurations as we could state they are only validated > >> against the broker configs (e.g. min/max). We would only need to think > >> about the upgrade path or we may need to keep it in both places for a > >> while. > >> > >> DJ_05: There are two interfaces called `MemberAssignmentState`. This > >> seems to be a mistake. > >> > >> Best, > >> David > >> > >> On Wed, Jun 24, 2026 at 6:41 PM Gabriella Fu via dev > >> <[email protected]> wrote: > >>> > >>> Hi Alieh,Sean, and Lucas, > >>> > >>> Thanks for the feedback. > >>> > >>> AS1: No, as Lucas mentioned, there's no guard in the customer assignor, > >> and > >>> assignors are responsible for themselves. > >>> > >>> AS2: If it's the downgrade case, the broker won't recognize this config > >> and > >>> will just ignore it. If it an assignor being removed, if it's not in > the > >>> streams.assignors.name, then it has no impace, since the new list of > >>> assignor will be reloaded when the broker restart. If it's > map.get(name) > >>> returns nothing, the assignor will fall back to the default assignor > and > >> a > >>> warming will be log. > >>> > >>> AS3: The broker config is group.streams.assignors to match with > consumer > >>> group and the group config use streams.assignor name to deperate from > it > >>> > >>> Nit: Thanks! I will check my typos > >>> > >>> SQ01: Thanks! If that's more accurate I can rename it. > >>> > >>> Best, > >>> Gabriella > >>> > >>> > >>> > >>> On Wed, Jun 24, 2026 at 10:15 AM Sean Quah via dev < > [email protected] > >>> > >>> wrote: > >>> > >>>> Hi Gabriella, > >>>> Thanks for the KIP. It looks good overall and consistent with the > >> KIP-848 > >>>> consumer assignors design. > >>>> > >>>> DJ_03: Could we go a bit further with documenting `configs()`? Does it > >>>> always return the full set of config keys and where can a developer > >> writing > >>>> a custom assignor find the full list of config keys they need to > >> consider? > >>>> > >>>> AS3: Right now the convention for group configs is to prefix with the > >> group > >>>> type. eg. we have "consumer.session.timeout.ms" and " > >>>> streams.session.timeout.ms" instead of a single session timeout > >> config. > >>>> > >>>> sq01: nit: `MemberSubscription` doesn't contain any subscription info. > >> I > >>>> understand the name came from the consumer assignor interface. In > other > >>>> places, we have chosen to rename interfaces to be more accurate > >>>> (MemberAssignmentState). Do we want to do the same or stick to the > >> consumer > >>>> assignor names? > >>>> > >>>> Thanks, > >>>> Sean > >>>> > >>>> On Wed, Jun 24, 2026 at 2:45 PM Lucas Brutschy via dev < > >>>> [email protected]> > >>>> wrote: > >>>> > >>>>> Hi, > >>>>> > >>>>> Thanks for the updates Gabby, lgtm! > >>>>> > >>>>> On AS1, consumer assignors also have no timeout, so I'd stay > >>>>> consistent with that. KIP could note assignors are responsible for > >>>>> bounded execution and that a hung assignor blocks the coordinator > >>>>> thread. > >>>>> > >>>>> Cheers, > >>>>> Lucas > >>>>> > >>>>> On Wed, Jun 24, 2026 at 1:46 PM Alieh Saeedi via dev > >>>>> <[email protected]> wrote: > >>>>>> > >>>>>> Hey Gabriella, > >>>>>> > >>>>>> Thanks for putting this KIP together. I think the main concerns > >> have > >>>>>> already been raised and addressed well. I just wanted to note a > >> couple > >>>> of > >>>>>> personal questions: > >>>>>> > >>>>>> AS1: Is there any guardrail for a custom assignor that is slow or > >> gets > >>>>>> stuck in a loop, such as a timeout, given that it runs on the group > >>>>>> coordinator thread? If not, it may be worth stating in the KIP that > >>>>>> assignor authors are responsible for keeping execution bounded, > >> along > >>>>> with > >>>>>> the operational impact of a hung assignor. I remember the initial > >>>>>> implementation of the sticky assignor being slow, which I think > >> will > >>>>> affect > >>>>>> rebalance behavior, so this made me think this might be important > >> to > >>>> call > >>>>>> out. > >>>>>> > >>>>>> AS2: What happens during downgrade or config removal? For example, > >> if a > >>>>>> group has a persisted streams.assignor.name and the broker is > >> later > >>>>>> downgraded to a version that does not support this feature, or the > >>>>> assignor > >>>>>> is removed from group.streams.assignors, is the persisted per-group > >>>>> config > >>>>>> ignored or cleared? Does the group cleanly fall back to the > >> default? > >>>> I’m > >>>>>> not sure whether covering the downgrade path makes sense in this > >> KIP, > >>>> but > >>>>>> it came to mind and I was curious whether you had thought about it. > >>>>>> > >>>>>> AS3: For the per-group config streams.assignor.name, I think > >>>>>> group.assignor.name reads more clearly, but it’s up to you. > >>>>>> > >>>>>> Nit: There’s also a typo to fix — group.streams.assignors is > >> listed in > >>>>>> singular form twice in the KIP. > >>>>>> > >>>>>> -Alieh > >>>>>> > >>>>>> On Tue, Jun 23, 2026 at 11:18 PM Gabriella Fu via dev < > >>>>> [email protected]> > >>>>>> wrote: > >>>>>> > >>>>>>> Hi all, > >>>>>>> > >>>>>>> I updated the KIP based on the discussion. Here are the changes I > >>>> made: > >>>>>>> > >>>>>>> - > >>>>>>> > >>>>>>> DJ_01 (configs) / LB / MJS: I collapsed this to a single > >> broker > >>>>> config, > >>>>>>> group.streams.assignors (LIST, default [sticky]), mirroring > >>>>>>> group.consumer.assignors in KIP-848. The first entry is the > >>>> cluster > >>>>>>> default, and selectable short names are derived from > >>>>> TaskAssignor.name() > >>>>>>> rather than declared separately, so the previous > >>>>>>> group.streams.assignors.names config is removed. I also > >> removed > >>>> the > >>>>>>> misleading “avoids leaking class names” motivation. > >>>>>>> - > >>>>>>> > >>>>>>> DJ_02 / MJS : All input/output types are now interfaces rather > >>>> than > >>>>>>> records. GroupSpec is now an interface with per-member > >> accessors > >>>>>>> instead of > >>>>>>> returning a raw Map. AssignmentMemberSpec is split into > >>>>>>> MemberSubscription > >>>>>>> (static metadata: processId, instanceId, rackId, clientTags) > >> and > >>>> the > >>>>>>> member’s runtime assignment, matching the KIP-848 split. > >>>>>>> - > >>>>>>> > >>>>>>> DJ_03: I renamed assignmentConfigs() to configs() and > >> clarified > >>>>> that it > >>>>>>> returns only the assignment-relevant subset of group configs > >> (for > >>>>>>> example, > >>>>>>> num.standby.replicas), not all group configs. > >>>>>>> - > >>>>>>> > >>>>>>> DJ_04 : no change. > >>>>>>> - > >>>>>>> > >>>>>>> LB1: I clarified that loaded assignors implementing > >>>>>>> org.apache.kafka.common.Configurable have configure() invoked > >> with > >>>>> the > >>>>>>> broker configuration at startup, the same as consumer-side > >>>>> assignors. > >>>>>>> Per-group dynamic settings remain a separate channel via > >>>>>>> GroupSpec.configs(). > >>>>>>> - > >>>>>>> > >>>>>>> MJS1: All newly public types are now annotated with > >>>>>>> @InterfaceStability.Evolving so we can refine the API in minor > >>>>> releases > >>>>>>> without making a compatibility commitment. > >>>>>>> - > >>>>>>> > >>>>>>> MJS2s: I dropped warm-up tasks from the assignor input and > >> output. > >>>>>>> MemberAssignment now exposes only active and standby tasks. > >>>>>>> > >>>>>>> Let me know if you have more concern. > >>>>>>> > >>>>>>> Thanks, > >>>>>>> > >>>>>>> Gabriella > >>>>>>> > >>>>>>> On Tue, Jun 23, 2026 at 12:23 PM Gabriella Fu <[email protected]> > >>>>> wrote: > >>>>>>> > >>>>>>>> Hi all, thank you so much for the discussion > >>>>>>>> > >>>>>>>> LB1: Yes. Custom TaskAssignor loading does the same as the > >> consumer > >>>>> side. > >>>>>>>> Each entry in group.streams.assignors is resolved to an > >> instance at > >>>>>>> broker > >>>>>>>> startup, and if it implements Configurable, the broker calls > >>>>>>>> configure(config.originals()) on it. So a custom assignor can > >> read > >>>>>>>> broker‑level configuration the same way a consumer assignor > >> can. > >>>>>>> Per‑group > >>>>>>>> dynamic settings (e.g. num.standby.replicas) remain a separate > >>>>> channel > >>>>>>> via > >>>>>>>> GroupSpec.configs(). I'll make this explicit in the KIP. > >>>>>>>> > >>>>>>>> > >>>>>>>> MJS1:Agreed. I'll mark all newly public-facing interfaces, > >> classes, > >>>>> and > >>>>>>>> methods as @InterfaceStability.Evolving so we can refine them > >> in > >>>>> minor > >>>>>>>> releases without a compatibility commitment. I'll update the > >> KIP to > >>>>> state > >>>>>>>> this for the whole new API surface. > >>>>>>>> > >>>>>>>> MJS2: Thank you for pointing it out. I will drop the warmup > >> tasks > >>>>> from > >>>>>>> the > >>>>>>>> output MemberAssignment > >>>>>>>> > >>>>>>>> Thanks. > >>>>>>>> Gabriella > >>>>>>>> > >>>>>>>> > >>>>>>>> > >>>>>>>> On Sun, Jun 21, 2026 at 7:03 PM Matthias J. Sax < > >> [email protected]> > >>>>>>> wrote: > >>>>>>>> > >>>>>>>>> Thanks for the KIP Gabby. > >>>>>>>>> > >>>>>>>>> Two comments about the newly added interfaces/records. > >>>>>>>>> > >>>>>>>>> MJS1: I would like to mark all new public facing API as > >> @Evolving > >>>>> for > >>>>>>>>> now. We might not get it right with the first release, and > >> marking > >>>>> as > >>>>>>>>> evolving would indicate that the API is not stable yet, and > >> we can > >>>>>>>>> introduce breaking changes in minor releases, allowing us to > >> fix > >>>>> errors > >>>>>>>>> quickly in future minor releases. > >>>>>>>>> > >>>>>>>>> MJS2: The existing (non public) assignor API returns > >>>>> `GroupAssignment` > >>>>>>>>> which includes warmup task. I believe it's an artifact from > >> the > >>>>>>>>> "classic" case, in which the HA assignor was required to > >> compute > >>>>>>>>> warmups. However, with "streams" we can simplify the assignor > >> and > >>>> it > >>>>>>>>> would only compute active and standby tasks, while only the GC > >>>> (ie, > >>>>>>>>> reconciler) will use warmup tasks. Thus, we should change the > >>>>> interface > >>>>>>>>> accordingly, dropping warmup tasks from `MemberAssignment`. > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> About DJ_02: I have no experience with evolving Java > >> `records` but > >>>>> it > >>>>>>>>> seems they are more similar to `classes`. In general, > >> `interfaces` > >>>>>>>>> provide much higher flexibility than `classes`, so I agree > >> with > >>>>> David > >>>>>>>>> that it might be good to _only_ use `interfaces` and no > >> `records`. > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> About the configs (DJ_01): I guess I am fine either way, to > >> either > >>>>>>>>> follow the current proposal, or mimic KIP-848 more closely > >> with a > >>>>> single > >>>>>>>>> broker config containing a mix of built-in assignor-names and > >>>>>>>>> fully-qualified class names. > >>>>>>>>> > >>>>>>>>> In doubt I would agree with Lucas: the KIP-848 approach is a > >>>> little > >>>>> bit > >>>>>>>>> cleaner, as having one config seems easier than having two. > >> Let's > >>>>> hear > >>>>>>>>> from David about it. > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> -Matthias > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> On 6/16/26 1:55 AM, Lucas Brutschy via dev wrote: > >>>>>>>>>> Hi all, > >>>>>>>>>> > >>>>>>>>>> DJ_01: Agree, and I'd push it a little further. On the > >> consumer > >>>>> side > >>>>>>>>>> the short name isn't configured at all — > >>>> group.consumer.assignors > >>>>> is a > >>>>>>>>>> single list of built-in names and class names, each entry is > >>>>> resolved > >>>>>>>>>> to an instance, and they're mapped by name(); the first > >> entry is > >>>>> the > >>>>>>>>>> cluster default. So once the streams assignors are loaded, > >> their > >>>>> short > >>>>>>>>>> names already come from TaskAssignor.name(), which makes a > >>>>> separate > >>>>>>>>>> group.streams.assignors.names redundant — a single > >>>>>>>>>> group.streams.assignors list (default [sticky]) reproduces > >> the > >>>>>>>>>> consumer behavior. That also runs against the "mixed > >> name/class > >>>>>>>>>> configuration" rejected alternative, since the consumer > >> config > >>>> is > >>>>>>>>>> exactly that mixed list. > >>>>>>>>>> > >>>>>>>>>> DJ_04: I agree with Gabby; "validate all dynamic group > >> configs > >>>> on > >>>>> the > >>>>>>>>>> broker" seems out of scope for this KIP. > >>>>>>>>>> > >>>>>>>>>> LB1: When consumer assignors are loaded from > >>>>> group.consumer.assignors, > >>>>>>>>>> instances implementing Configurable get the broker configs > >>>> passed > >>>>> to > >>>>>>>>>> them. Will the custom TaskAssignor loading do the same, so a > >>>>> custom > >>>>>>>>>> assignor can read broker-level configuration the way a > >> consumer > >>>>>>>>>> assignor can? > >>>>>>>>>> > >>>>>>>>>> Thanks, > >>>>>>>>>> Lucas > >>>>>>>>>> > >>>>>>>>>> On Fri, Jun 12, 2026 at 9:53 PM Gabriella Fu via dev > >>>>>>>>>> <[email protected]> wrote: > >>>>>>>>>>> > >>>>>>>>>>> Hi David, > >>>>>>>>>>> > >>>>>>>>>>> Thanks a lot for the detailed review! > >>>>>>>>>>> > >>>>>>>>>>> *DJ_01:* That's a fair point. I'll update the KIP to use > >>>>>>>>>>> group.streams.assignors to align with KIP-848 and remove > >> the > >>>>>>> misleading > >>>>>>>>>>> motivation section. > >>>>>>>>>>> > >>>>>>>>>>> *DJ_02:* Thanks for the suggestion. I went through the > >> KIP-848 > >>>>>>>>> interfaces > >>>>>>>>>>> in org.apache.kafka.coordinator.group.api.assignor as well > >> as > >>>> the > >>>>>>>>> existing > >>>>>>>>>>> implementations (UniformAssignor, RangeAssignor), and the > >>>> pattern > >>>>>>> makes > >>>>>>>>>>> total sense now. I'll update the input types in the KIP > >>>>> accordingly: > >>>>>>>>>>> > >>>>>>>>>>> - > >>>>>>>>>>> > >>>>>>>>>>> GroupSpec will be changed to an interface with > >> per-member > >>>>>>>>> accessors, > >>>>>>>>>>> instead of returning a raw Map<String, > >>>> AssignmentMemberSpec>. > >>>>>>>>>>> - > >>>>>>>>>>> > >>>>>>>>>>> AssignmentMemberSpec will be replaced by two distinct > >>>>> interfaces: > >>>>>>>>>>> MemberSubscription and MemberAssignment. This will > >> separate > >>>>> the > >>>>>>>>> member's > >>>>>>>>>>> static metadata (processId, instanceId, rackId, > >> clientTags) > >>>>> from > >>>>>>>>> its > >>>>>>>>>>> runtime task assignment (active/standby/warmup tasks), > >>>>> mirroring > >>>>>>>>> the > >>>>>>>>>>> clean split in KIP-848. > >>>>>>>>>>> > >>>>>>>>>>> *DJ_03:* Thank you for the suggestion. I'll rename it to > >>>>> configs() > >>>>>>> and > >>>>>>>>>>> clarify in the text that it will only return configurations > >>>>> relevant > >>>>>>>>> to the > >>>>>>>>>>> assignment. > >>>>>>>>>>> > >>>>>>>>>>> *DJ_04:*Moving all dynamic group config validation to the > >>>> broker > >>>>>>> makes > >>>>>>>>>>> sense, but that would be a bit beyond the scope of this > >> KIP. In > >>>>> this > >>>>>>>>> case, > >>>>>>>>>>> maybe we should just keep the config like this? > >>>>>>>>>>> > >>>>>>>>>>> Best, > >>>>>>>>>>> Gabriella > >>>>>>>>>>> > >>>>>>>>>>> On Fri, Jun 12, 2026 at 10:54 AM David Jacot < > >> [email protected] > >>>>> > >>>>>>> wrote: > >>>>>>>>>>> > >>>>>>>>>>>> Hi Gabriella, > >>>>>>>>>>>> > >>>>>>>>>>>> Thanks for the KIP. I have a few high level comments: > >>>>>>>>>>>> > >>>>>>>>>>>> DJ_01: The motivation to not use > >> `group.streams.assignors` and > >>>>>>> follow > >>>>>>>>>>>> the pattern introduced by KIP-848 is pretty weak in my > >>>> opinion, > >>>>>>>>>>>> especially the "avoids leaking implementation class names > >> into > >>>>>>>>>>>> per-group dynamic configuration" part. The class names > >> wont > >>>> leak > >>>>>>> into > >>>>>>>>>>>> the group config as the group config requires the name of > >> the > >>>>>>>>>>>> assignor. From a user perspective, it is exactly the same > >>>>> concept so > >>>>>>>>>>>> using a different way to express it is wrong in my > >> opinion. > >>>>> Should > >>>>>>> we > >>>>>>>>>>>> just use `group.streams.assignors` to be consistent in our > >>>>> configs? > >>>>>>>>>>>> > >>>>>>>>>>>> DJ_02: I would suggest reconsidering the interface of the > >>>>> assignor. > >>>>>>> We > >>>>>>>>>>>> started with a similar interface in KIP-848 and we > >> realized > >>>>> during > >>>>>>> the > >>>>>>>>>>>> implementation that using POJOS and returning Maps (e.g. > >> for > >>>>>>>>>>>> members()) was really inflexible. Moreover, I am not sure > >> if > >>>>> using > >>>>>>>>>>>> records is good from an evolutionary point of view. Using > >>>>> interfaces > >>>>>>>>>>>> may be better. It also allows us to wrap internal objects > >> to > >>>>> expose > >>>>>>>>>>>> them to the assignor. Have you looked into the KIP-848's > >>>>> interface > >>>>>>> and > >>>>>>>>>>>> the various assignors? > >>>>>>>>>>>> > >>>>>>>>>>>> DJ_03: Regarding the `assignmentConfigs()`, should we > >> call it > >>>>>>>>>>>> `configs()` as it is already clear that it is for the > >>>>> assignment. > >>>>>>>>>>>> Moreover, I wonder if it is going to return all group > >> configs > >>>> or > >>>>>>> only > >>>>>>>>>>>> a subset of them. Could you please clarify in the KIP? > >>>>>>>>>>>> > >>>>>>>>>>>> DJ_04: The dynamic group configs validation makes sense > >> to me. > >>>>> Note > >>>>>>>>>>>> that this pattern is not used today. I also wonder > >> whether we > >>>>> should > >>>>>>>>>>>> move the validation of all dynamic group configs there. > >> To be > >>>>>>>>>>>> discussed. > >>>>>>>>>>>> > >>>>>>>>>>>> Best, > >>>>>>>>>>>> David > >>>>>>>>>>>> > >>>>>>>>>>>> On Thu, Jun 11, 2026 at 8:37 PM Gabriella Fu via dev > >>>>>>>>>>>> <[email protected]> wrote: > >>>>>>>>>>>>> > >>>>>>>>>>>>> Hi all, > >>>>>>>>>>>>> > >>>>>>>>>>>>> I’d like to start the discussion for KIP-1357: Add broker > >>>> side > >>>>>>> custom > >>>>>>>>>>>>> assignors for "streams" groups > >>>>>>>>>>>>> > >>>>>>>>>>>>> KIP: > >>>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>> > >>>>>>> > >>>>> > >>>> > >> > https://urldefense.com/v3/__https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=430408758__;!!Ayb5sqE7!t7TDJ8XzjCLooFy9FhJLiHKHu5c7RJf8-41tQuqRykmMLMB2yl4ApQsRMSEVKCIIl0FuvbznfZcDww$ > >>>>>>>>>>>>> JIRA: > >>>>>>>>>>>> > >>>>>>>>> > >>>>>>> > >>>>> > >>>> > >> > https://urldefense.com/v3/__https://issues.apache.org/jira/browse/KAFKA-20683__;!!Ayb5sqE7!t7TDJ8XzjCLooFy9FhJLiHKHu5c7RJf8-41tQuqRykmMLMB2yl4ApQsRMSEVKCIIl0Fuvbzx-7tAbA$ > >>>>>>>>>>>>> > >>>>>>>>>>>>> Summary: > >>>>>>>>>>>>> The Streams Rebalance Protocol (KIP-1071) moves task > >>>> assignment > >>>>>>> from > >>>>>>>>> the > >>>>>>>>>>>>> client to the broker, but unlike the classic protocol > >> and > >>>>> KIP-848 > >>>>>>>>>>>> consumer > >>>>>>>>>>>>> groups, it offers no way to plug in a custom assignor. > >> This > >>>> KIP > >>>>>>>>> closes > >>>>>>>>>>>> that > >>>>>>>>>>>>> gap by making the existing streams task assignor > >> interfaces > >>>>> public > >>>>>>>>> API > >>>>>>>>>>>> and > >>>>>>>>>>>>> adding broker and group configurations so operators can > >> load > >>>>> custom > >>>>>>>>>>>>> assignor implementations and select them per group by > >> short > >>>>> name, > >>>>>>>>> with no > >>>>>>>>>>>>> client-side involvement. > >>>>>>>>>>>>> > >>>>>>>>>>>>> Please let me know your feedback > >>>>>>>>>>>>> > >>>>>>>>>>>>> Thanks, > >>>>>>>>>>>>> Gabriella Fu > >>>>>>>>>>>> > >>>>>>>>> > >>>>>>>>> > >>>>>>> > >>>>> > >>>> > >> > > > >
