Thanks for the plentiful comments! My responses below: AJS1) Fair point, I had "plugin" in there initially but removed it thinking I would better follow conventions. But I see that it is confusing. Changed back to group.streams.topology.description.plugin.class.
AJS2) Not sure which Java object you mean, but sink-topics are modeled in the "Sink" node, so I don't see this missing. Predecessors are omitted on the broker-side, since they are not present on the wire and reconstruction seems unnecessary. AJS3) Good point. Added it. AJS4) I went back-and-forth on this one as well, but if we don't have the error code, all we can return is UNKNOWN_SERVER_ERROR when the topology description update fails. I don't think we want to do that, since it may trigger alerts set up to detect implementation bugs in the Kafka broker. So the point of TOPOLOGY_DESCRIPTION_UPDATE_FAILED is just letting the client know that the update failed, but there is nothing wrong with the Kafka broker. The client would likely log that at INFO level, opposed to WARN or ERRROR for UNKNOWN_SERVER_ERROR. AJS5) Sure, we can do that. AJS6) It was not supposed to be a user-facing concept. And I think it introduced more problems than it solved, so I removed the concept altogether. AJS7) Good point. This, in part, prompted me to bring back a previous version of the design, without the topology decription ID. Now, the topology description is created using only the topologyEpoch and a group creation timestamp. Then, the plugin can trivially implement concurrency control: The plugin can decide to store all versions, which means updates for separate topology epochs are independent. Or it can just store the latest one, which means that a topology description that comes later in the (groupCreationTimestamp, topologyEpoch) order takes preference over any previous update. AJS8) No, there is no direct relationship between a group epoch and a topology description. BB1) The asymmetry is intentional. Predecessors are redundant with sucessor information. We can reconstruct it any time. In the admin client, which are user-facing, we do reconstruct the predecessor for usability. On the broker-side, the main purpose is passing the data around, so we do not reconstruct the relation. I'll clarify this in the javadoc. BB2) Direct neighbours only — exactly as you described. I'll clarify this in the javadoc. BB3) The design intentionally pushes transient-retry policy entirely onto the plugin: when setTopology fails for a recoverable reason, the plugin is expected to keep returning true from requiresTopologyPush on subsequent heartbeats. I will clarify this in the Plugin Implementation section. MJS1) I dropped the UUID, as it introduced more complexities than it solved. MJS2a) I think we benefit from not modeling this too tightly. A sink is a sink, and the sink topic is a just a metadata annotation on top of it. There are corner cases, which I think would be more difficult to add later if we do not include NodeType. One is regular expressions. We do not model regular expressions here (because KIP-1071 does not support them yet -- and the existing client-side describe doesn't support them). We can still add regular expressions later, but then source topics will still be empty. Even clients that do not know the regular expressions field in the future should understand that it's a source node, even if maybe it does not understand how the source topics are determined. Similarly for sink nodes with topicNameExtractors, or any other nifty features people will cook up in the future. There is no defined sink topic for that node, but it's still a sink. The NodeType should define the identity, not the metadata attached to the node. I actually weakened the guarantees in the documentation around this a bit in the KIP, to make this clear. MJS2b) Fair — but it would make the schema a lot less uniform, because the corner case of global stores would require it's own record types with some fields fixed. Let me know if you feel strongly about it, otherwise, I'd vote for re-using the generic topologyRecords. MJS3) There is no Kafka RPC without a response, so we'd break with all conventions and I don't see why we'd even consider it. Apart from that, there are useful things to do in reaction to the response - e.g. log that the topology is too large, and throttleTimeMs. MJS4) The intent here is "this broker doesn't advertise this API in ApiVersions, but you sent it anyway, so I'm rejecting". This seems like a perfectly fine way to use UNSUPPORTED_VERSION to me. MJS5) This question comes up in every KIP I have the feeling. There is no strong reason to do either, but the preference is to use version bumps, because they are more compact on the wire and old versions without the field can be retired. I cleaned up some prose around this. MJS6) I think having the module `group-coordinator-api` is a decent place to put it. It's the place for broker-side plugin interfaces related to group management. Adding a new module broker-plugins for all broker-side plugins seems like scope creep for this KIP, and I am not even sure it would make things better. MJS7) Correct, this is a bit misleading. Exceptions are signaled by completing the future exceptionally, never by throwing synchronously from setTopology. I will clarify this in the KIP. MJS8) Yes, similar to ASH01 - the plugin needs to make sure that the futures are completed. MJS9) I agree that 3 POJOs are somewhat awkward, but creating dependencies, e.g. between the admin client and a broker-side plugin seems even worse to me. Where would we put the common data structure? MJS10) In principle we are modeling a sum type here (topology description | reason of absence), but Java doesn't natively model these, so we store two separate fields. Including a status that doesn't model the AVAILABLE case because it can be inferred seems more of a pitfall than necessary. Avoiding redundancy in APIs is a good pattern but not a strict rule, and I think having topologyDescription().isPresent() == (status == AVAILABLE) is perfectly fine. Also Optional would just add a layer of wrapping what can be modeled directly via the enum. But I agree that my distinction (omit the AVAILABLE in the RPC and include it in the user-facing API) may be unnecessarily confusing, see AJS5. MJS11) Personally, I think only the plugin knows if something went wrong and why, so the best place to define metrics is inside the plugin. Similar reasoning to why I want to keep configs, deduplication inside the plugin. If people strongly feel that we want to have AK metrics for this plugin, we can add them. MJS12) If the feature is disabled on the client but the broker plugin is enabled, the broker would indeed try to resolicitate on the client-side with back-off. But the solicitation is very light-weight (a single boolean, and requireTopologyId is supposed to be fast), and clients will just ignore it, so I don't see a problem with this. MJS13) The plugin doesn't (and shouldn't need to) know the heartbeat interval — it should throttle based on its own clock plus its own in-flight tracking. Concrete strategy: for each (groupId, groupCreationTimeMs, topologyEpoch) tuple the plugin tracks (i) whether a push has been initiated (set when requiresTopologyPush first returns true) and (ii) the last requiresTopologyPush=true time. While (i) is set and the push hasn't completed, requiresTopologyPush returns false. After the push completes successfully it returns false permanently for that tuple. On a transient failure, after a back-off window (independent of heartbeat cadence — say, exponential starting at 1s) it returns true again. On permanent failure (TOPOLOGY_DESCRIPTION_TOO_LARGE or plugin-semantic INVALID_REQUEST) it returns false permanently and logs. Hope that makes sense! On Sat, May 9, 2026 at 4:43 AM Matthias J. Sax <[email protected]> wrote: > > Thanks for the KIP Lucas. I made a first pass. Couple of comments/questions. > > > MJS1(a): This is a follow up to ASH03. I am not sure if I understand the > problem? The KIP says, we call `requiresTopologyPush` on every HB. So > after an upgrade, the plugin will be called for every existing group, > allowing the group to send its topology? So we can just create a UUID > before this call? In the end, we need to somehow cache all currently > in-use UUID anyway (ie, one for each active group)? > > > MJS1(b): Related to the above. The KIP does not say that UUIDs would be > stored by the GC -- so after a broker bounce, or GC fail-over to a > different broker, it seems we would forget all currently in-use UUIDs > and generate new ones? This would align to what I did ask about above, > and it should be fine from plugin POV to just get new UUIDs if none is > cached? > > > MJS1(c): However, I am not even sure about TopologyDescriptionId? Should > we use <groupId,topologyEpoch> instead, avoid this UUID all together? On > the other hand, with regard to ASH03, we might introduce the problem > Sanghyeok describe when doing this? On the other hand, on startup, the > GC could also check all existing groups, and just call > `requiresTopologyPush` pro-actively for each group? > > > MJS1(d): KIP-1313 proposes that clients create their UUID -- should we > do the same to integrate with KIP-1313 (in case we keep UUID, and not > move to <groupId,topologyEpoch>), to align the behavior across the > board? In the end, the topology can only change during a roll, which > aligns to a topology-epoch bump anyway? > > > > MJS2(a): The KIP introduces a `NodeType` field for the topology > description. I am wondering if we need it? We have sources, processors, > sink. Only sources can have input topics, and only sinks can have output > topic, and processor never have any input/output topic, so it seems just > inspecting if input/output topic are present, tells us what node type we > have, and have an explicit types seems to be redundant? > > > MJS2(b): I am wondering why we model GlobalStores with two > TopologyNodes? We know that for this case, there is exactly one source > node, and one processor. Should we simplify this? > > > > MJS3: Do we actually need to add > `UpdateStreamsGroupTopologyDescriptionResponse`, or could we use a "fire > and forget" approache? If an topology update failed, and the plugin > re-request the push, the next HB-response would take care of it > naturally it seems? Atm the only value we get is to send back some error > code. It this worth it? > > > > MJS4: The KIP add `UNSUPPORTED_VERSION` error. > > > UNSUPPORTED_VERSION — the coordinator cannot serve this RPC because no > > topology description plugin is configured > Is this the right name for this error? > > > > MJS5: the KIP proposed to bump the version of both > StreamsGroupDescribeRequest and StreamsGroupDescribeResponse. No > objection, but wondering why we prefer it over tagged fields? > > > > MJS6: Is `org.apache.kafka.coordinator.group.api.streams` the right > place for the plugin interface? It seems we add a very heavy dependency > for people implementing the plugin. Would it make sense to add a new > module `broker-plugins` instead to make it more light weight? Yes, > KIP-714 does the same, but it might be a nice improvement for 714, too, > to move their plugin into such a new module? > > > MJS7: The KIP says that `setTopology()` may throw > TopologyDescriptionTooLargeException? Later in "plugin guidelines" it says: > > > Reject pushes that exceed it by completing the setTopology future with > > TopologyDescriptionTooLargeException > > Which one is it? We might not want to do it both ways? Also applies to > other exception. > > > > MJS8: It seems the broker will need to cache all non-yet-completed > ComparableFutures. What if the plugin has a bug, and never completes > it's future? Would we get some leak? -- Or is your answer the same as > for ASH01? > > > > MJS9: Should we move `StreamsGroupTopologyDescription` to the same > propose `broker-plugin` module? I am also wondering about Alieh's > question: Do we keep it's own class in purpose, of should be unify with > the existing interface? And why do we get a second > `StreamsGroupTopologyDescription` for the admin client? -- I understand > your argument about "this may evolve independently", but do we need 3 > copies of the same? > > > MJS10(a): Do we need to expose `StreamsGroupTopologyDescriptionStatus` > via `topologyDescriptionStatus()`? I am wondering if > `StreamsGroupTopologyDescription` could model this directly? Also not > sure if `topologyDescriptionStatus` should return an `Optional` or not? > > > > MJS10(b): I can see argument to ignore what I said in MJS10(b), but if > we have an Optional, why do we need status `AVAILABLE`? > > > > MJS11: The KIP says we don't add any broker side metrics. KIP-714 did. > Wondering why we won't need any for this KIP? > > > > MJS12: If this feature is disabled client side, can the broker learn > about it? Or might it keep requesting the topology over and over again, > and the client would just keep ignoring the request? Would we want some > error-code the clients sends to the broker for this case instead? > > > MJS13: In the "plugin guidelines", the KIP says: > > > Avoid concurrent or repetitive pushes > > Does the plugin know the HB interval to implement this in reasonable > way? Any better guidance we can give how to implement this? > > > > -Matthias > > > > On 5/5/26 6:26 AM, Lucas Brutschy via dev wrote: > > Thanks Sanghyeok and Alieh! > > > > ASH01: Both risks are real. But I would argue that anyways, the > > plugins that Kafka currently defines need to be implemented correctly > > for Kafka to reliably work. This is the case here as well - just like > > a consumer group assignor needs to be correct and fast, > > requiresTopologyPush needs to be implemented correct and fast. I did > > give most of the responsibility to the plugin here, because it depends > > on the plugin in which situations it requires a new topology push from > > the client. I feel like adding extra logic to time out slow > > requireTopologyPush calls, or enforcing minimum intervals would > > require more configuration options and an extra level of deduplication > > logic - confusing users that do not use the plugin (configurations are > > irrelevant for them) and confusing users that do use the plugin (two > > potentially conflicting levels need to be configured). > > > > ASH02: Good point and this is actually something that I considered. > > But it's actually not that easy - in principle, there is only one > > current topology description, but there may be an arbitrary number of > > stale topologies active in the group. Would we require the plugin to > > store all stale topologies? I think this would be a feasible > > extension, but would definitely add some complexity. I would propose > > adding this as a follow-up item. I can add this to the future work > > section. > > > > ASH03: Good catch. I don't really want to have an "upgrade logic" that > > we need to preserve forever in the group coordinator. I think it would > > be okay to allow "ZERO_UUID" for any topology that exists when the > > broker upgraded. > > > > ASH04: I noted this in the future work section. In principle we can > > detect mismatches between topology descriptions on the client, but we > > do not include it in this KIP, since it would complicate things. The > > first successfully stored topology is authoritative. > > > > AS01) Both topologies are derived from the same Topology instance on > > the client at a given epoch, so at the source they're consistent by > > construction. During a topology update, the topology ID changes, and > > initially we will not have a new topology description passed to the > > plugin. In this case, we can get an intermediate NOT_STORED response > > when we try to describe the streams group, until a client pushes the > > new topology description. But the result will be consistent with the > > topology information used for assignments. This assumes that all > > clients with the same topology epoch use the same topology > > description. Mismatch detection is noted as future work. > > > > AS02) I think your point about retrying with too large descriptions is > > valid. It would actually make sense to leave topology size checking to > > the plugin as well - it can decide the maximum topology size and stop > > returning requiresTopologyPush for topologies that are confirmed to be > > too large. I will make this change in the next revision of the KIP. > > > > AS03) Yes, this is mostly to avoid dependencies between Kafka > > packages. Note that we do not necessarily need to keep the two > > implementations in sync. The streams-side TopologyDescription may > > evolve differently than the admin-side TopologyDescription. The two > > are only weakly linked through the RPC definition. > > > > AS04) Agreed. However, topology has the slowest-changing lifecycle of > > the three, so it should be less confusing than assignments and > > members. > > > > AS05) Correct. The plugin is free to forward, persist, fan out, mirror > > to multiple sinks, or anything else. The KIP intentionally doesn't > > constrain the storage backend. > > > > On Tue, May 5, 2026 at 1:19 PM Alieh Saeedi <[email protected]> wrote: > >> > >> Thanks Lucas for the KIP. The KIP is already in very good shape and covers > >> the edge cases. I still have a few questions and considerations I’d like > >> to share. > >> > >> AS01: Are Assignment topology (defined in KIP-1071) and the Description > >> topology (defined in KIP-1331) guaranteed to be consistent views of the > >> same logical topology, or can they drift? Are we guaranteeing that every > >> assignment we surface references only nodes/topics present in the current > >> description topology, or can operators see combinations that don’t line up? > >> > >> AS02: I'm cusrious about the rationale or empirical data behind the 350 KB > >> default (e.g., based on observed real-world topologies)? Also the KIP says > >> the broker measures topology size and rejects oversized payloads with > >> TOPOLOGY_DESCRIPTION_TOO_LARGE. Should the Streams client attempt a > >> best-effort pre-check of the serialized size to avoid repeated failing > >> pushes and log a clearer local error? Or is the intent to keep the client > >> simple and rely entirely on the broker response + plugin behavior for this > >> case? > >> > >> AS03: Why do we introduce a separate Admin-side POJO instead of reusing > >> TopologyDescription from the Streams API—for dependency/semantic reasons? > >> And how do we plan to keep the two representations in sync? > >> > >> AS04: Somewhat related to AS01.... In practice we’ve seen that because > >> members and assignments change so dynamically, a user may see different > >> assignments or members over just a few seconds, or a member with a > >> specific memberId may disappear entirely. Having the topology visible > >> might help users understand what’s going on—but it could also make things > >> more confusing, depending on the situation. > >> > >> AS05: I assume that even with a single plugin, multiple downstream systems > >> can still benefit from it (the plugin can of course fan out to multiple > >> downstream systems). Am I right? > >> > >> Thanks, > >> Alieh > >> > >> > >> > >> On Mon, May 4, 2026 at 11:39 AM Lucas Brutschy via dev > >> <[email protected]> wrote: > >>> > >>> Hi all, > >>> > >>> I would like to start the discussion on KIP-1331. The idea is to > >>> optionally make a topology description available to the broker, in the > >>> spirit of KIP-714. Looking forward to your feedback! > >>> > >>> https://urldefense.com/v3/__https://cwiki.apache.org/confluence/display/KAFKA/KIP-1331*3A*Streams*Group*Topology*Description*Plugin__;JSsrKysr!!Ayb5sqE7!sxqGDUcjOzRpt9Gk0jE1XnVSit-FZMIihk2UsXWUI0jmdYK2nTcO1hP-9WiW5sLBMw8amIUxG2PGvhdRhok$ > >>> > >>> Best, > >>> Lucas >
