Hi Lucas/Bruno, thanks for the great KIP! First comments: LM1. Related to where the KIP says: *“Group ID, member ID, member epoch are sent with each heartbeat request. Any other information that has not changed since the last heartbeat can be omitted.”. *I expect all the other info also needs to be sent whenever a full heartbeat is required (even if it didn’t change from the last heartbeat), ex. on fencing scenarios, correct?
LM2. For consumer groups we always send the groupInstanceId (if any) as part of every heartbeat, along with memberId, epoch and groupId. Should we consider that too here? LM3. We’re proposing returning a GROUP_ID_NOT_FOUND error in response to the stream-specific RPCs if the groupId is associated with a group type that is not streams (ie. consumer group or share group). I wonder if at this point, where we're getting several new group types added, each with RPCs that are supposed to include groupId of a certain type, we should be more explicit about this situation. Maybe a kind of INVALID_GROUP_TYPE (group exists but not with a valid type for this RPC) vs a GROUP_ID_NOT_FOUND (group does not exist). Those errors would be consistently used across consumer, share, and streams RPCs whenever the group id is not of the expected type. This is truly not specific to this KIP, and should be addressed with all group types and their RPCs in mind. I just wanted to bring out my concern and get thoughts around it. LM4. On a related note, StreamsGroupDescribe returns INVALID_REQUEST if groupId is empty. There is already an INVALID_GROUP_ID error, that seems more specific to this situation. Error handling of specific errors would definitely be easier than having to deal with a generic INVALID_REQUEST (and probably its custom message). I know that for KIP-848 we have INVALID_REQUEST for similar situations, so if ever we take down this path we should review it there too for consistency. Thoughts? LM5. The dependency between the StreamsGroupHeartbeat RPC and the StreamsGroupInitialize RPC is one-way only right? HB requires a previous StreamsGroupInitialize request, but StreamsGroupInitialize processing is totally independent of heartbeats (and could perfectly be processed without a previous HB, even though the client implementation we’re proposing won’t go down that path). Is my understanding correct? Just to double check, seems sensible like that at the protocol level. LM6. With KIP-848, there is an important improvement that brings a difference in behaviour around the static membership: with the classic protocol, if a static member joins with a group instance already in use, it makes the initial member fail with a FENCED_INSTANCED_ID exception, vs. with the new consumer group protocol, the second member trying to join fails with an UNRELEASED_INSTANCE_ID. Does this change need to be considered in any way for the streams app? (I'm not familiar with KS yet, but thought it was worth asking. If it doesn't affect in any way, still maybe helpful to call it out on a section for static membership) LM7. Regarding the admin tool to manage streams groups. We can discuss whether to have it here or separately, but I think we should aim for some basic admin capabilities from the start, mainly because I believe it will be very helpful/needed in practice during the impl of the KIP. From experience with KIP-848, we felt a bit blindfolded in the initial phase where we still didn't have kafka-consumer-groups dealing with the new groups (and then it was very helpful and used when we were able to easily inspect them from the console) LM8. nit: the links the KIP-848 are not quite right (pointing to an unrelated “Future work section” at the end of KIP-848) Thanks! Lianet On Fri, Jul 19, 2024 at 11:13 AM Lucas Brutschy <lbruts...@confluent.io.invalid> wrote: > Hi Andrew, > > AS2: I added a note for now. If others feel strongly about it, we can > still add more administrative tools to the KIP - it should not change > the overall story significantly. > > AS8: "streams.group.assignor.name" sounds good to me to distinguish > the config from class names. Not sure if I like the "default". To be > consistent, we'd then have to call it > `group.streams.default.session.timeout.ms` as well. I only added the > `.name` on both broker and group level for now. > > AS10: Ah, I misread your comment, now I know what you meant. Good > point, fixed (by Bruno). > > Cheers, > Lucas > > On Fri, Jul 19, 2024 at 4:44 PM Andrew Schofield > <andrew_schofi...@live.com> wrote: > > > > Hi Lucas, > > I see that I hit send too quickly. One more comment: > > > > AS2: I think stating that there will be a `kafka-streams-group.sh` in a > > future KIP is fine to keep this KIP focused. Personally, I would probably > > put all of the gory details in this KIP, but then it’s not my KIP. A > future > > pointer is fine too. > > > > Thanks, > > Andrew > > > > > > > On 19 Jul 2024, at 13:46, Lucas Brutschy <lbruts...@confluent.io.INVALID> > wrote: > > > > > > Hi Andrew, > > > > > > thanks for getting the discussion going! Here are my responses. > > > > > > AS1: Good point, done. > > > > > > AS2: We were planning to add more administrative tools to the > > > interface in a follow-up KIP, to not make this KIP too large. If > > > people think that it would help to understand the overall picture if > > > we already add something like `kafka-streams-groups.sh`, we will do > > > that. I also agree that we should address how this relates to > > > KIP-1043, we'll add it. > > > > > > AS3: Good idea, that's more consistent with `assigning` and > `reconciling` etc. > > > > > > AS4: Thanks, Fixed. > > > > > > AS5: Good catch. This was supposed to mean that we require CREATE on > > > cluster or CREATE on all topics, not both. Fixed. > > > > > > AS6: Thanks, Fixed. > > > > > > AS7. Thanks, Fixed. > > > > > > AS8: I think this works a bit different in this KIP than in consumer > > > groups. KIP-848 lets the members vote for a preferred assignor, and > > > the broker-side assignor is picked by majority vote. The > > > `group.consumer.assignors` specifies the list of assignors that are > > > supported on the broker, and is configurable because the interface is > > > pluggable. In this KIP, the task assignor is not voted on by members > > > but configured on the broker-side. `group.streams.assignor` is used > > > for this, and uses a specific name. If we'll make the task assignor > > > pluggable on the broker-side, we'd introduce a separate config > > > `group.streams.assignors`, which would indeed be a list of class > > > names. I think there is no conflict here, the two configurations serve > > > different purposes. The only gripe I'd have here is naming as > > > `group.streams.assignor` and `group.streams.assignors` would be a bit > > > similar, but I cannot really think of a better name for > > > `group.streams.assignor`, so I'd probably rather introduce > > > `group.streams.assignors` as `group.streams.possible_assignors` or > > > something like that. > > > > > > AS9: I added explanations for the various record types. Apart from the > > > new topology record, and the partition metadata (which is based on the > > > topology and can only be created once we have a topology initialized) > > > the lifecycle for the records is basically identical as in KIP-848. > > > > > > AS10: In the consumer offset topic, the version in the key is used to > > > differentiate different schema types with the same content. So the > > > keys are not versioned, but the version field is "abused" as a type > > > tag. This is the same in KIP-848, we followed it for consistency. > > > > > > Cheers, > > > Lucas > > > > > > > > > On Thu, Jul 18, 2024 at 1:27 PM Andrew Schofield > > > <andrew_schofi...@live.com> wrote: > > >> > > >> Hi Lucas and Bruno, > > >> > > >> Thanks for the great KIP. > > >> > > >> I've read through the document and have some initial comments. > > >> > > >> AS1: I suppose that there is a new o.a.k.common.GroupType.STREAMS > enumeration > > >> constant. This is a change to the public interface and should be > called out. > > >> > > >> AS2: Since streams groups are no longer consumer groups, how does the > user > > >> manipulate them, observe lag and so on? Will you add > `kafka-streams-groups.sh` > > >> or extend `kafka-streams-application-reset.sh`? Of course, KIP-1043 > can easily > > >> be extended to support streams groups, but that only lets the user > see the > > >> groups, not manipulate them. > > >> > > >> AS3: I wonder whether the streams group state of UNINITIALIZED would > be > > >> better expressed as INITIALIZING. > > >> > > >> AS4: In StreamsGroupInitializeRequest, Topology[].SourceTopicRegex > should > > >> be nullable. > > >> > > >> AS5: Why does StreamsGroupInitialize require CREATE permission on the > > >> cluster resource? I imagine that this is one of the ways that the > request might > > >> be granted permission to create the StateChangelogTopics and > > >> RepartitionSourceTopics, but if it is granted permission to create > those topics > > >> with specific ACLs, would CREATE on the cluster resource still be > required? > > >> > > >> AS6: StreamsGroupInitialize can also fail with > TOPIC_AUTHORIZATION_FAILED > > >> and (subject to AS5) CLUSTER_AUTHORIZATION_FAILED. > > >> > > >> AS7: A tiny nit. You've used TopologyID (capitals) in > StreamsGroupHeartbeatRequest > > >> and a few others, but in all other cases the fields which are ids are > spelled Id. > > >> I suggest TopologyId. > > >> > > >> Also, "interal" is probably meant to be "interval”. > > >> > > >> AS8: For consumer groups, the `group.consumer.assignors` > configuration is a > > >> list of class names. The assignors do have names too, but the > configuration which > > >> enables them is in terms of class names. I wonder whether the broker’s > > >> group.streams.assignor could actually be `group.streams.assignors` > and specified > > >> as a list of the class names of the supplied assignors. I know you're > not supporting > > >> other assignors yet, but when you do, I expect you would prefer to > have used class > > >> names from the start. > > >> > > >> The use of assignor names in the other places looks good to me. > > >> > > >> AS9: I'd find it really helpful to have a bit of a description about > the purpose and > > >> lifecycle of the 9 record types you've introduced on the > __consumer_offsets topic. > > >> I did a cursory review but without really understanding what's > written when, > > >> I can't do a thorough job of reviewing. > > >> > > >> AS10: In the definitions of the record keys, such as > > >> StreamsGroupCurrentMemberAssignmentKey, the versions of the fields > must > > >> match the versions of the types. > > >> > > >> Thanks, > > >> Andrew > > >> > > >>> On 12 Jul 2024, at 09:04, Lucas Brutschy > > >>> <lbruts...@confluent.io.INVALID> > wrote: > > >>> > > >>> Hi all, > > >>> > > >>> I would like to start a discussion thread on KIP-1071: Streams > > >>> Rebalance Protocol. With this KIP, we aim to bring the principles > laid > > >>> down by KIP-848 to Kafka Streams, to make rebalances more reliable > and > > >>> scalable, and make Kafka Streams overall easier to deploy and > operate. > > >>> The KIP proposed moving the assignment logic to the broker, and > > >>> introducing a dedicated group type and dedicated RPCs for Kafka > > >>> Streams. > > >>> > > >>> The KIP is here: > > >>> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-1071%3A+Streams+Rebalance+Protocol > > >>> > > >>> This is joint work with Bruno Cadonna. > > >>> > > >>> Please take a look and let us know what you think. > > >>> > > >>> Best, > > >>> Lucas > > >> > > >
