THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
For KStream, we are going to take a trade-off between “revoking all” and “revoking none” solution: we shall only revoke tasks that are being learned since last round. So when we assign learner tasks to new member, we shall also mark active tasks as "being learned task" on current owners. Every time when a rebalance begins, the task owners will revoke the being learned tasks and join group without affecting other ongoing tasks. This way learned tasks could immediately transfer ownership without attempting for a second round of rebalance. Compared with KIP-415, we are optimizing for fewer rebalances, but increasing the metadata size and sacrificing partial availability of the learner tasks.
Algorithm Walkthrough
Let's use some annotation to define the new learner algorithm for a holistic view.
| LaTeX Formatting |
|---|
$ax^2 = 5;$ |
...
| Code Block | ||||
|---|---|---|---|---|
| ||||
Group stable state: S1[T1, T2], S2[T3, T4], S3[T5]
Scaling down the application, S2 will be leaving.
#First Rebalance
Member S2 joins the group and claim that it is leaving
S1 performs task assignments:
S1(assigned: [T1, T2], revoked: [], learning: [T3])
S2(assigned: [T3, T4], revoked: [], learning: [])
S3(assigned: [T5], revoked: [], learning: [T4])
#Second Rebalance
S3 finishes replay first and trigger another rebalance
Member S1~S3 join with following status:(S1 is not ready yet)
S1(assigned: [T1, T2], revoked: [], learning: [T3])
S2(assigned: [T3], revoked: [T4], learning: [])
S3(assigned: [T5], revoked: [], learning: [T4])
S1 performs task assignments:
S1(assigned: [T1, T2], revoked: [], learning: [T3])
S2(assigned: [T3], revoked: [T4], learning: [])
S3(assigned: [T4, T5], revoked: [], learning: [])
#Third Rebalance
S1 finishes replay and trigger rebalance.
Member S1~S3 join with following status:
S1(assigned: [T1, T2], revoked: [], learning: [T3])
S2(assigned: [], revoked: [T3], learning: [])
S3(assigned: [T4, T5], revoked: [], learning: [])
S1 performs task assignments:
S1(assigned: [T1, T2, T3], revoked: [], learning: [])
S2(assigned: [], revoked: [T3], learning: [])
S3(assigned: [T4, T5], revoked: [], learning: [])
S2 will shutdown itself upon new assignment since there is no assigned task left. |
Leader Transfer During Scaling
Leader crash could cause a missing of historical assignment information. For the learner assignment, however, each worker maintains its own assignment status, so when the learner task's id has no active task running, the transfer will happen immediately. Leader switch in this case is not a big concern.
| Code Block | ||||
|---|---|---|---|---|
| ||||
Cluster has 3 stream workers S1(leader), S2, S3, and they own tasks 1 ~ 5
Group stable state: S1[T1, T2], S2[T3, T4], S3[T5]
#First Rebalance
New member S4 joins the group
S1 performs task assignments:
S1(assigned: [T1, T2], revoked: [], learning: [])
S2(assigned: [T3, T4], revoked: [], learning: [])
S3(assigned: [T5], revoked: [], learning: [])
S4(assigned: [], revoked: [], learning: [T1])
#Second Rebalance
S1 crashes/gets killed before S4 is ready, S2 takes over the leader.
Member S2~S4 join with following status:
S2(assigned: [T3, T4], revoked: [], learning: [])
S3(assigned: [T5], revoked: [], learning: [])
S4(assigned: [], revoked: [], learning: [T1])
Note that T2 is unassigned, and S4 is learning T1 which has no current active task. We
could rebalance T1, T2 immediately.
S2 performs task assignments:
S2(assigned: [T3, T4], revoked: [], learning: [])
S3(assigned: [T5, T2], revoked: [], learning: [])
S4(assigned: [T1], revoked: [], learning: [])
Now the group reaches balance. |
Summary
Note that to make sure the above resource shuffling could happen as expected, we need to have at least three new task status indicator to be provided:
...