mjsax commented on code in PR #23096:
URL: https://github.com/apache/kafka/pull/23096#discussion_r3725443777
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -288,44 +289,44 @@ private static boolean assignStandbyToMemberWithLeastLoad(
/**
* Finds the previous member with the least load for a given task.
*
- * @param localState The state of the assignment in progress.
- * @param members The list of previous members owning the task.
- * @param taskId The taskId, to check if the previous member already has
the task. Can be null, if we assign it
- * for the first time (e.g., during active task assignment).
+ * @param localState
+ * The state of the assignment in progress.
+ * @param members
+ * The list of previous members owning the task.
+ * @param standbyTaskId
+ * The taskId, to check if the previous member already has the task.
*
* @return Previous member with the least load that does not have the
task, or null if no such member exists.
*/
private static Member findPrevMemberWithLeastLoad(
final LocalState localState,
final ArrayList<Member> members,
- final TaskId taskId
+ final Optional<TaskId> standbyTaskId
) {
if (members == null || members.isEmpty()) {
return null;
}
- Member candidate = members.get(0);
- final ProcessState candidateProcessState =
localState.processIdToState.get(candidate.processId);
- double candidateProcessLoad = candidateProcessState.load();
- double candidateMemberLoad =
candidateProcessState.memberToTaskCounts().get(candidate.memberId);
- for (int i = 1; i < members.size(); i++) {
- final Member member = members.get(i);
+ Member candidate = null;
+ double candidateProcessLoad = Double.MAX_VALUE;
+ double candidateMemberLoad = Double.MAX_VALUE;
+ for (final Member member : members) {
final ProcessState processState =
localState.processIdToState.get(member.processId);
+ // A process that already owns a standby task (either as active or
standby) cannot take it again,
Review Comment:
```suggestion
// A process that already owns a standby task (either as active
or standby) cannot take it again
```
--
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]
