gabriellefu commented on code in PR #23591:
URL: https://github.com/apache/kafka/pull/23591#discussion_r4262987583
##########
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:
Since findPrevMemberWithLeastLoad() already return the least-loaded
pre-member, if it's null already meaning all the premember don't have room, why
we need to add member -> hasUnfulfilledTaskQuota(localState,
localState.processIdToState.get(member.processId), member) here?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -308,13 +309,16 @@ private static void assignActive(final LocalState
localState, final LinkedList<T
for (final Iterator<TaskId> it = activeTasks.iterator();
it.hasNext();) {
final TaskId task = it.next();
final ArrayList<Member> prevMembers =
localState.standbyTaskToPrevMember.get(task);
- final Member prevMember = findPrevMemberWithLeastLoad(localState,
prevMembers, Optional.empty());
+ final Member prevMember = findPrevMemberWithLeastLoad(
+ localState,
+ prevMembers,
+ Optional.empty(),
+ member -> hasUnfulfilledActiveTaskQuota(localState,
localState.processIdToState.get(member.processId), member, stateful)
Review Comment:
Since findPrevMemberWithLeastLoad() already return the least-loaded
pre-member, if it's null already meaning all the premember don't have room, why
we need to add member -> hasUnfulfilledTaskQuota(localState,
localState.processIdToState.get(member.processId), member) 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]
