gabriellefu commented on code in PR #23588:
URL: https://github.com/apache/kafka/pull/23588#discussion_r4389945242
##########
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);
Review Comment:
thanks, I actually file a ticket about this issue and was thinking about
doing a follow up and follow the pattern with trunk. But I can update it here.
--
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]
