[
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)
