Repository: apache/kafka
File: group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java (Java)

Code before:
        if (!processWithLeastLoad.hasTask(taskId)) {
            final String memberId = memberWithLeastLoad(processWithLeastLoad);
            if (memberId != null) {
                processWithLeastLoad.addTask(memberId, taskId, false);
                found = true;
            }
        } else if (!queue.isEmpty()) {
            found = assignStandbyToMemberWithLeastLoad(queue, taskId);
        }
        queue.add(processWithLeastLoad); // Add it back to the queue after updating its state
        return found;
    }

    /**
     * Finds the previous member with the least load for a given task.
     *
     * @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).
     *
     * @return Previous member with the least load that deoes not have the task, or null if no such member exists.
     */
    private Member findPrevMemberWithLeastLoad(final ArrayList<Member> members, final TaskId taskId) {
        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);
            final ProcessState processState = localState.processIdToState.get(member.processId);
            final double newProcessLoad = processState.load();
            if (newProcessLoad < candidateProcessLoad && (taskId == null || !processState.hasTask(taskId))) {
                final double newMemberLoad = processState.memberToTaskCounts().get(member.memberId);
                if (newMemberLoad < candidateMemberLoad) {
                    candidateProcessLoad = newProcessLoad;
                    candidateMemberLoad = newMemberLoad;
                    candidate = member;
                }
            }
        }

        if (taskId == null || !candidateProcessState.hasTask(taskId)) {
            return candidate;
        }
        return null;
    }


Review comment (suggestion):
There's a typo in the documentation comment: 'deoes' should be 'does'.
```suggestion
     * @return Previous member with the least load that does not have the task, or null if no such member exists.
```

Code after:
        final ProcessState processWithLeastLoad = queue.poll();
        if (processWithLeastLoad == null) {
            return false;
        }
        boolean found = false;
        if (!processWithLeastLoad.hasTask(taskId)) {
            final String memberId = memberWithLeastLoad(processWithLeastLoad);
            if (memberId != null) {
                processWithLeastLoad.addTask(memberId, taskId, false);
                found = true;
            }
        } else if (!queue.isEmpty()) {
            found = assignStandbyToMemberWithLeastLoad(queue, taskId);
        }
        queue.add(processWithLeastLoad); // Add it back to the queue after updating its state
        return found;
    }

    /**
     * Finds the previous member with the least load for a given task.
     *
     * @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).
     *
     * @return Previous member with the least load that deoes not have the task, or null if no such member exists.
     */
    private Member findPrevMemberWithLeastLoad(final ArrayList<Member> members, final TaskId taskId) {
        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);
            final ProcessState processState = localState.processIdToState.get(member.processId);
            final double newProcessLoad = processState.load();
            if (newProcessLoad < candidateProcessLoad && (taskId == null || !processState.hasTask(taskId))) {
                final double newMemberLoad = processState.memberToTaskCounts().get(member.memberId);
                if (newMemberLoad < candidateMemberLoad) {
                    candidateProcessLoad = newProcessLoad;
                    candidateMemberLoad = newMemberLoad;
                    candidate = member;
                }
            }
        }

        if (taskId == null || !candidateProcessState.hasTask(taskId)) {