You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by Gabriella Fu via dev <de...@kafka.apache.org> on 2026/07/07 18:40:23 UTC
Re: [DISCUSS] KIP-1357: Add broker side custom assignors for "streams" groups
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 <da...@gmail.com> 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
> <de...@kafka.apache.org> 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 <dev@kafka.apache.org
> >
> > 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 <
> > > dev@kafka.apache.org>
> > > 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
> > > > <de...@kafka.apache.org> 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 <
> > > > dev@kafka.apache.org>
> > > > > 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 <gf...@confluent.io>
> > > > 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 <
> mjsax@apache.org>
> > > > > > 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
> > > > > > >> > <de...@kafka.apache.org> 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 <
> dajac@apache.org
> > > >
> > > > > > 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
> > > > > > >> >>> <de...@kafka.apache.org> 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
> > > > > > >> >>>
> > > > > > >>
> > > > > > >>
> > > > > >
> > > >
> > >
>
Re: [DISCUSS] KIP-1357: Add broker side custom assignors for "streams" groups
Posted by Gabriella Fu via dev <de...@kafka.apache.org>.
Thanks Matthias,
I will update my kip accordingly.
Best,
Gabriella
On Wed, Jul 8, 2026 at 9:10 AM Matthias J. Sax <mj...@apache.org> 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 <da...@gmail.com>
> 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
> >> <de...@kafka.apache.org> 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 <
> dev@kafka.apache.org
> >>>
> >>> 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 <
> >>>> dev@kafka.apache.org>
> >>>> 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
> >>>>> <de...@kafka.apache.org> 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 <
> >>>>> dev@kafka.apache.org>
> >>>>>> 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 <gf...@confluent.io>
> >>>>> 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 <
> >> mjsax@apache.org>
> >>>>>>> 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
> >>>>>>>>>> <de...@kafka.apache.org> 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 <
> >> dajac@apache.org
> >>>>>
> >>>>>>> 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
> >>>>>>>>>>>> <de...@kafka.apache.org> 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
> >>>>>>>>>>>>
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>
> >>>>>
> >>>>
> >>
> >
>
>
Re: [DISCUSS] KIP-1357: Add broker side custom assignors for "streams" groups
Posted by "Matthias J. Sax" <mj...@apache.org>.
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 <da...@gmail.com> 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
>> <de...@kafka.apache.org> 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 <dev@kafka.apache.org
>>>
>>> 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 <
>>>> dev@kafka.apache.org>
>>>> 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
>>>>> <de...@kafka.apache.org> 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 <
>>>>> dev@kafka.apache.org>
>>>>>> 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 <gf...@confluent.io>
>>>>> 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 <
>> mjsax@apache.org>
>>>>>>> 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
>>>>>>>>>> <de...@kafka.apache.org> 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 <
>> dajac@apache.org
>>>>>
>>>>>>> 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
>>>>>>>>>>>> <de...@kafka.apache.org> 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
>>>>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>
>>>>>
>>>>
>>
>