[
https://issues.apache.org/jira/browse/KAFKA-21401?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18134813#comment-18134813
]
sanghyeok An commented on KAFKA-21401:
--------------------------------------
One simple approach worth considering might be the following.
If a static member has a different process ID, we could treat it as a different
static member. In that case, we would remove the existing static member and
generate the necessary records for its removal. We would then create a new
member, generate the corresponding records, and trigger a recalculation of the
target assignment.
I'll look into whether we can address this issue with this approach.
> Streams static member rejoin with a changed process ID loses task ownership
> tracking and permits premature assignment
> ---------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-21401
> URL: https://issues.apache.org/jira/browse/KAFKA-21401
> Project: Kafka
> Issue Type: Bug
> Components: group-coordinator, streams
> Reporter: sanghyeok An
> Assignee: sanghyeok An
> Priority: Major
>
> When a static member rejoins a Streams group with a new process ID,
> inheriting its previous task assignment can cause it to hold the same standby
> task as another member on that process.
> If one of those members subsequently acknowledges revocation, the coordinator
> removes the process from the task ownership index even though the other
> member still holds the task. A later assignment can then grant the active
> role to a new member on the same process without waiting for the remaining
> standby owner to release the task.
>
> *Scenario*
> The scenario assumes the following conditions:
> * The group uses the built-in Sticky assignor and the default
> NoOpAssignmentRefiner.
> * There are three stateful tasks, {{{}0_0{}}}, {{{}0_1{}}}, and {{{}0_2{}}},
> with two standby replicas configured.
> * Process Pold contains static members A1 and A2. Process Q contains member
> B.
> * A1 and A2 temporarily leave, then rejoin sequentially with their original
> instance IDs, new member IDs, and a new process ID, Pnew.
> * When A2 rejoins, a tie between standby candidates is resolved in favor of
> A2. The resulting target assignment retains Standby {{0_2}} on A2 and removes
> it from A1.
> * Target assignments are recomputed for the relevant rejoins. Before A3
> joins in the final step, A2 does not produce a metadata or current-assignment
> update that restores its ownership index entry.
> In the table below, *A means active* and {*}S means standby{*}. A1 and A2
> denote the logical members across rejoins; their actual member IDs change
> when they rejoin.
> The table shows the current assignments and pending revocations tracked by
> the coordinator at each step, rather than the final target assignment.
> ||Step||Pold||Pnew||Q||
> |*1. Initial state*|*A1:* A={{{}0_0{}}} / *A2:* A={{{}0_1{}}},
> S={{{}0_2{}}}|None|*B:* A={{{}0_2{}}}, S={{{}0_0, 0_1{}}}|
> |*2. A1 and A2 temporarily leave (memberEpoch=-2)*|*Retained records only:*
> A1: A={{{}0_0{}}} / A2: A={{{}0_1{}}}, S={{{}0_2{}}}|None|*B:* A={{{}0_2{}}},
> S={{{}0_0, 0_1{}}}|
> |*3. A1 rejoins first on Pnew*|*Only A2's record remains:* A={{{}0_1{}}},
> S={{{}0_2{}}}|*A1:* A={{{}0_0{}}}, S={{{}0_1, 0_2{}}}|*B:* A={{{}0_2{}}},
> S={{{}0_0, 0_1{}}}|
> |*4. A2 also rejoins on Pnew*|None|*A1:* A={{{}0_0{}}}, S={{{}0_1, 0_2{}}} /
> *A2:* A={{{}0_1{}}}, S={{{}0_2{}}}|*B:* A={{{}0_2{}}}, S={{{}0_0, 0_1{}}}|
> |*5. A1 is instructed to revoke its standby tasks*|None|*A1:* A={{{}0_0{}}},
> S={{{}0_1, 0_2{}}} pending revocation / *A2:* A={{{}0_1{}}},
> S={{{}0_2{}}}|*B:* A={{{}0_2{}}}, S={{{}0_0, 0_1{}}}|
> |*6. A1 acknowledges completed revocation*|None|*A1:* A={{{}0_0{}}} / *A2:*
> A={{{}0_1{}}}, S={{{}0_2{}}}|*B:* A={{{}0_2{}}}, S={{{}0_0, 0_1{}}}|
> |*7. B permanently leaves (memberEpoch=-1)*|None|*A1:* A={{{}0_0{}}} / *A2:*
> A={{{}0_1{}}}, S={{{}0_2{}}}|None|
> |*8. A new member A3 joins Pnew*|None|*A1:* A={{{}0_0{}}} / *A2:*
> A={{{}0_1{}}}, *S={{{}0_2{}}}* / *A3:* *A={{{}0_2{}}}*|None|
> At step 3, the coordinator sees Pold, Pnew, and Q as distinct processes. It
> can therefore assign Standby {{0_2}} to A1 on Pnew while retaining A2's
> existing standby assignment on Pold.
> At step 4, A2 moves to Pnew and inherits its previous assignment. Since
> Standby {{0_2}} appears in both its current and target assignments,
> reconciliation treats it as a retained task. It does not undergo the
> ownership check applied to newly assigned tasks. Consequently, A2 retains the
> standby without waiting for A1 to release it, creating duplicate ownership on
> the same process.
> *Loss of ownership tracking*
> The standby ownership index stores a {{Set<processId>}} for each task. It
> does not track individual member ownership or the number of owners within a
> process.
> ||Step||Standby owners of task 0_2 on Pnew||Processes recorded in the standby
> ownership index||
> |*4. Immediately after duplicate assignment*|Both A1 and A2 hold the
> task|Pnew|
> |*5. A1 is pending revocation*|A1 is pending revocation; A2 continues to hold
> the task|Pnew|
> |*6. A1 acknowledges completed revocation*|*A2 still holds the task*|*Empty*|
> At step 6, processing A1's revocation acknowledgement removes Pnew from the
> index without accounting for A2's continued ownership of Standby {{0_2}} on
> that process.
> Subsequent ordinary heartbeats from A2 do not necessarily repair the index.
> If A2 is already stable and no assignment update is generated, the missing
> entry remains absent.
>
> *Incorrect assignment*
> After B leaves at step 7 and A3 joins at step 8, the Sticky assignor retains
> the existing active tasks on A1 and A2 and assigns the remaining Active
> {{0_2}} to A3's target assignment.
> Because A2 still holds Standby {{0_2}} on the same process, the coordinator
> should withhold the active assignment from A3 until A2 releases the standby.
> However, Pnew is missing from the ownership index, so the coordinator does
> not detect the outstanding ownership and includes Active {{0_2}} in A3's
> heartbeat response.
> As a result, {*}different members on the same process simultaneously hold the
> standby and active roles for the same task{*}. This violates process-level
> task exclusivity and the cooperative handoff requirement to confirm release
> by the existing owner before granting the task to another member.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
