[jira] [Comment Edited] (KAFKA-20170) Support topology updates for streams groups without requiring a new group

["sanghyeok An (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-20170?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18103075#comment-18103075 ] 

sanghyeok An edited comment on KAFKA-20170 at 8/9/26 3:20 AM:
--------------------------------------------------------------

Hi, [~lucasbru] !

 
While looking into this issue, I noticed a potential correctness problem during a topology update. I may be missing an existing safeguard, so I would appreciate your thoughts.
 
Assume that the *orders* and *clicks* topics have compatible partition counts.
With topology epoch 9, the group has two independent subtopologies: * *Subtopology 0* reads from {*}orders{*}.
 * *Subtopology 1* reads from {*}clicks{*}.

The current assignment could be: * Member A (topology {*}epoch 9{*}): *task 0_0* processes {*}orders-0{*}.
 * Member B (topology {*}epoch 9{*}): *task 1_0* processes {*}clicks-0{*}.

Member A then shuts down and releases its old task. It rejoins with topology epoch 10. In the new topology, the two sources are joined and copartitioned:
 * *Subtopology 0* reads from both *orders* and {*}clicks{*}.
 * The new *task 0_0* processes both *orders-0* and {*}clicks-0{*}.

During reconciliation, the following state may be possible:
 * Member A (topology {*}epoch 10{*}): new *task 0_0* processes *orders-0* and {color:#de350b}*clicks-0*{color}.
 * Member B (topology {*}epoch 9{*}): old *task 1_0* is still processing {color:#de350b}*clicks-0*{color} until it acknowledges its revocation.

 

The old and new tasks have different task IDs ({*}1_0{*} and {*}0_0{*}). Therefore, the normal cooperative handoff for the same active task may not make the new task wait for the old one. If task 0_0 is assigned before Member B has revoked task 1_0, both members could process clicks-0 concurrently using different topology versions.

This seems to suggest that preventing stale members from receiving new tasks may not be sufficient when a topology update changes the mapping between tasks and the topics used by their subtopologies, for example when subtopologies are merged or split.

I think the concern may not be limited to user source topics. A changed subtopology may also use repartition source topics, repartition sink topics, state changelog topics, and updated copartition groups. Old and new tasks may use the same internal topic partitions under different topology definitions. For example, repartition records produced by the old topology may be mixed with records from the new topology, or stateful tasks may write to or restore from the same changelog partition using incompatible state definitions.

One possible direction may be for the coordinator to retain enough information about the previous and current topologies during the transition. It could then delay an affected new task until any old tasks using overlapping source, repartition, or changelog topic partitions have been revoked.

Some of this transition information may also need to be persisted in __consumer_offsets so that the coordinator can safely recover after a failover instead of losing the relationship between the old and new tasks.

There may also be a sequencing issue if topology epoch 11 is introduced while the transition from epoch 9 to 10 is still in progress. It may be simpler to serialize topology updates and defer a new update until the previous rollout has completed, rather than retaining an unbounded number of topology versions and transition states.

I would appreciate your thoughts on this potential issue. If it seems valid, I can investigate the transition behavior further and prepare a small proof of concept to better understand the required changes. If the discussion shows that additional protocol or design changes are needed beyond KIP-1071, I would also be willing to work on a follow-up KIP.

 

What do you think? Thanks for reading this! 


was (Author: JIRAUSER303328):
[~lucasbru] 

 
While looking into this issue, I noticed a potential correctness problem during a topology update. I may be missing an existing safeguard, so I would appreciate your thoughts.
 
Assume that the *orders* and *clicks* topics have compatible partition counts.
With topology epoch 9, the group has two independent subtopologies: * *Subtopology 0* reads from {*}orders{*}.
 * *Subtopology 1* reads from {*}clicks{*}.


The current assignment could be: * Member A (topology {*}epoch 9{*}): *task 0_0* processes {*}orders-0{*}.
 * Member B (topology {*}epoch 9{*}): *task 1_0* processes {*}clicks-0{*}.

Member A then shuts down and releases its old task. It rejoins with topology epoch 10. In the new topology, the two sources are joined and copartitioned:
 * *Subtopology 0* reads from both *orders* and {*}clicks{*}.
 * The new *task 0_0* processes both *orders-0* and {*}clicks-0{*}.

During reconciliation, the following state may be possible:
 * Member A (topology {*}epoch 10{*}): new *task 0_0* processes *orders-0* and {color:#de350b}*clicks-0*{color}.
 * Member B (topology {*}epoch 9{*}): old *task 1_0* is still processing {color:#de350b}*clicks-0*{color} until it acknowledges its revocation.

 

The old and new tasks have different task IDs ({*}1_0{*} and {*}0_0{*}). Therefore, the normal cooperative handoff for the same active task may not make the new task wait for the old one. If task 0_0 is assigned before Member B has revoked task 1_0, both members could process clicks-0 concurrently using different topology versions.


This seems to suggest that preventing stale members from receiving new tasks may not be sufficient when a topology update changes the mapping between tasks and the topics used by their subtopologies, for example when subtopologies are merged or split.


I think the concern may not be limited to user source topics. A changed subtopology may also use repartition source topics, repartition sink topics, state changelog topics, and updated copartition groups. Old and new tasks may use the same internal topic partitions under different topology definitions. For example, repartition records produced by the old topology may be mixed with records from the new topology, or stateful tasks may write to or restore from the same changelog partition using incompatible state definitions.


One possible direction may be for the coordinator to retain enough information about the previous and current topologies during the transition. It could then delay an affected new task until any old tasks using overlapping source, repartition, or changelog topic partitions have been revoked.


Some of this transition information may also need to be persisted in __consumer_offsets so that the coordinator can safely recover after a failover instead of losing the relationship between the old and new tasks.


There may also be a sequencing issue if topology epoch 11 is introduced while the transition from epoch 9 to 10 is still in progress. It may be simpler to serialize topology updates and defer a new update until the previous rollout has completed, rather than retaining an unbounded number of topology versions and transition states.


I would appreciate your thoughts on this potential issue. If it seems valid, I can investigate the transition behavior further and prepare a small proof of concept to better understand the required changes. If the discussion shows that additional protocol or design changes are needed beyond KIP-1071, I would also be willing to work on a follow-up KIP.

 

What do you think? Thanks for reading this! 

> Support topology updates for streams groups without requiring a new group
> -------------------------------------------------------------------------
>
>                 Key: KAFKA-20170
>                 URL: https://issues.apache.org/jira/browse/KAFKA-20170
>             Project: Kafka
>          Issue Type: Task
>          Components: group-coordinator, streams
>            Reporter: Lucas Brutschy
>            Assignee: sanghyeok An
>            Priority: Major
>              Labels: kip1071
>
> The streams rebalance protocol \(KIP\-1071\) does not currently support topology updates. If a topology is changed significantly \(e.g., by adding new source topics or changing the number of subtopologies\), users must create a new streams group. This is a significant limitation for applications that need to evolve over time.
> As specified in KIP\-1071, topology updates should be supported through explicit versioning via a topology.epoch configuration. When a member joins with a different topology, the broker should compare epochs: if the epoch is the same but metadata differs, return STREAMS\_INVALID\_TOPOLOGY\_EPOCH; if the epoch is lower, return STREAMS\_TOPOLOGY\_FENCED; if the epoch is bumped by one, accept and update the group topology. Members with stale topologies should receive a STALE\_TOPOLOGY status in heartbeat responses but continue processing with their current tasks until upgraded. This enables rolling upgrades of the topology across application instances.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)