gabriellefu commented on code in PR #23588:
URL: https://github.com/apache/kafka/pull/23588#discussion_r4174035759
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -467,38 +496,89 @@ 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 {@code
allowed} accepts. Null if neither.
+ */
+ private static Member findPrevMemberForStandby(
+ final LocalState localState,
+ final TaskId task,
+ final Predicate<ProcessState> allowed
+ ) {
+ // prev active task
+ final Member prevActiveMember =
localState.activeTaskToPrevMember.get(task);
+ if (prevActiveMember != null) {
+ final ProcessState prevActiveMemberProcessState =
localState.processIdToState.get(prevActiveMember.processId);
+ if (allowed.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), allowed);
+ if (prevStandbyMember != null && hasUnfulfilledTaskQuota(localState,
localState.processIdToState.get(prevStandbyMember.processId),
prevStandbyMember)) {
+ return prevStandbyMember;
+ }
+ return null;
+ }
+
+ /** Whether {@code task} had an active or standby member before, the only
members a sticky pick can go back to. */
+ private static boolean hasPrevMember(final LocalState localState, final
TaskId task) {
+ final ArrayList<Member> prevStandbyMembers =
localState.standbyTaskToPrevMember.get(task);
+ return localState.activeTaskToPrevMember.get(task) != null ||
prevStandbyMembers != null && !prevStandbyMembers.isEmpty();
+ }
+
+ /**
+ * 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
+ * rack-aware pick may still take the room a tag-blind sticky pick needs.
+ */
private static void assignStandby(final LocalState localState, final
LinkedList<TaskId> standbyTasks) {
Review Comment:
sounds good, I have moved steps 1 and 2 into assignRackAwareStandbys, and
assignStandby now only orders
the rack-aware and tag-blind steps, and steps 3–4 match trunk apart from the
start index.
--
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]
