gabriellefu commented on code in PR #23588:
URL: https://github.com/apache/kafka/pull/23588#discussion_r4415499011
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -468,38 +485,59 @@ private static boolean hasUnfulfilledTaskQuota(
return process.memberToTaskCounts().get(member.memberId) <
localState.totalTasksPerMember;
}
+ /**
+ * The previous member to take a standby of {@code task}: the previous
active member, else the least-loaded previous
+ * standby member, each only while below the quota and on a process that
meets {@code prevMemberCondition}. Null if
+ * neither.
+ */
+ private static Member findPrevMemberForStandby(
+ final LocalState localState,
+ final TaskId task,
+ final Predicate<ProcessState> prevMemberCondition
+ ) {
+ // prev active task
+ final Member prevActiveMember =
localState.activeTaskToPrevMember.get(task);
+ if (prevActiveMember != null) {
+ final ProcessState prevActiveMemberProcessState =
localState.processIdToState.get(prevActiveMember.processId);
+ if (prevMemberCondition.test(prevActiveMemberProcessState) &&
!prevActiveMemberProcessState.hasTask(task) &&
hasUnfulfilledTaskQuota(localState, prevActiveMemberProcessState,
prevActiveMember)) {
+ return prevActiveMember;
+ }
+ }
+
+ // prev standby tasks
+ final Member prevStandbyMember =
findPrevMemberWithLeastLoad(localState,
localState.standbyTaskToPrevMember.get(task), Optional.of(task),
prevMemberCondition);
+ if (prevStandbyMember != null && hasUnfulfilledTaskQuota(localState,
localState.processIdToState.get(prevStandbyMember.processId),
prevStandbyMember)) {
+ return prevStandbyMember;
+ }
+ return null;
+ }
+
+ /**
+ * Assigns the standby tasks. Rack diversity ranks above stickiness: when
{@code rack.aware.assignment.tags} is
+ * set, steps 1 and 2 place every standby that still makes its task more
rack-diverse, using stickiness only to
+ * break ties among the most diverse processes; steps 3 and 4 are the
tag-blind assignment of the rest. Each pair
+ * runs its sticky step for all tasks before its least-loaded step, so
that a pick that leaves its previous members
+ * cannot take the room a later task needs to stay with one of its own.
Steps 3 and 4 run after step 2, so a
Review Comment:
I will address this in the follow up
--
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]
