[PR] KAFKA-21321: Let the refiner see a member that joins in the same heartbeat [kafka]

["suzhiking (via GitHub)" <gi...@apache.org>]

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: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-21321: Let the refiner see a member that joins in the same heartbeat [kafka]

["suzhiking (via GitHub)" <gi...@apache.org>]

suzhiking commented on code in PR #23899:
URL: https://github.com/apache/kafka/pull/23899#discussion_r4418495792


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -316,6 +316,20 @@ private static UpdateTargetAssignmentResult<Assignment> fromLastTargetAssignment
         }
     }
 
+    /**
+     * The result of updating a streams group's target assignment.
+     *
+     * @param targetAssignmentEpoch The target assignment epoch.
+     * @param targetAssignment      All members' target assignments.
+     * @param members               The members the target assignment is keyed by: the group's members with the
+     *                              unwritten membership change of the heartbeat applied.
+     */
+    private record UpdateStreamsTargetAssignmentResult(
+        int targetAssignmentEpoch,
+        Map<String, TasksTuple> targetAssignment,
+        Map<String, StreamsGroupMember> members
+    ) { }
+

Review Comment:
   that would be cleaner, thanks!



-- 
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: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org