sanghyeok An created KAFKA-21401:
------------------------------------
Summary: 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
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)
