chickenchickenlove commented on code in PR #23591:
URL: https://github.com/apache/kafka/pull/23591#discussion_r4420952426
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -490,14 +500,17 @@ private static void assignStandby(final LocalState
localState, final LinkedList<
// prev standby tasks
final ArrayList<Member> prevStandbyMembers =
localState.standbyTaskToPrevMember.get(task);
if (prevStandbyMembers != null &&
!prevStandbyMembers.isEmpty()) {
- final Member prevStandbyMember =
findPrevMemberWithLeastLoad(localState, prevStandbyMembers, Optional.of(task));
+ final Member prevStandbyMember =
findPrevMemberWithLeastLoad(
+ localState,
+ prevStandbyMembers,
+ Optional.of(task),
+ member -> hasUnfulfilledTaskQuota(localState,
localState.processIdToState.get(member.processId), member)
+ );
if (prevStandbyMember != null) {
Review Comment:
@gabriellefu
Sorry for the delay and Thanks for taking a look! Previously,
`findPrevMemberWithLeastLoad()` did not check quota -- that check happened in
the caller after selecting a candidate. So a null result did not mean that all
previous members were at quota.
The subtle case is two members in the same process: the helper only replaces
its candidate when both process load and member load are strictly lower. Since
their process loads are equal, it can select a member already at quota and
overlook another member that still has room. The caller then rejects the
selected member and falls back to general load balancing.
--
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]
