suzhiking opened a new pull request, #23899: URL: https://github.com/apache/kafka/pull/23899
In a streams group heartbeat, the coordinator refines the target assignment into an intermediate assignment with warm-up tasks. The target assignment it refines already reflects the membership change of the heartbeat: maybeUpdateStreamsTargetAssignment applies that change through an UpdatedMembersAndTargetAssignmentView, which adds a joining member and relabels a replaced static member to its new member ID. The refiner, however, was handed group.members(), which does not contain that member yet, because the records of the heartbeat are only replayed after the handler returns. The refiner therefore treated the member as one that had left the group. The worst effect is on a static member that rejoins under a new member ID, as it does on every restart. The refiner kept the member's stateful active tasks on the old member ID, which the replacement removes, so the rejoined member was not given them and no member ran them until the next refinement step. In addition, a new member did not get a warm-up task for a stateful task that the target assignment moves to it, and a new member that is the target standby holder of a migrating task was not held back, so the task ran one standby more than configured while the migration lasted. maybeUpdateStreamsTargetAssignment now returns the view's members together with the target assignment, and the refiner is handed those members whenever it refines that target assignment. The other call site, which refines the current target assignment of a Stable group when the heartbeat changes nothing, keeps using group.members(): a heartbeat that adds or replaces a member never takes that path, because a new member bumps the group epoch and a replaced static member has left with epoch -2, so the group is not Stable. See https://issues.apache.org/jira/browse/KAFKA-21321. The tests cover two static member rejoins with the real refiner and warm-up tasks enabled: one from a new process, where the target assignment is recomputed, and one from the same process with no intermediate assignment cached for the epoch, as after a coordinator failover. Both fail without this change, because the rejoined member is given no active task. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
