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

Code before:

    private String memberWithLeastLoad(final ProcessState processWithLeastLoad) {
        final Map<String, Integer> members = processWithLeastLoad.memberToTaskCounts();
        if (members.isEmpty()) {
            return null;
        }
        if (members.size() == 1) {
            return members.keySet().iterator().next();
        }
        final Optional<String> memberWithLeastLoad = processWithLeastLoad.memberToTaskCounts().entrySet().stream()
            .min(Map.Entry.comparingByValue())
            .map(Map.Entry::getKey);
        return memberWithLeastLoad.orElse(null);
    }

    private boolean hasUnfulfilledQuota(final Member member) {
        return localState.processIdToState.get(member.processId).memberToTaskCounts().get(member.memberId) < localState.tasksPerMember;
    }

    private void assignStandby(final LinkedList<TaskId> standbyTasks, int numStandbyReplicas) {
        final ArrayList<StandbyToAssign> toLeastLoaded = new ArrayList<>(standbyTasks.size() * numStandbyReplicas);

        // Assuming our current assignment is range-based, we want to sort by partition first.
        standbyTasks.sort(Comparator.comparing(TaskId::partition).thenComparing(TaskId::subtopologyId).reversed());

        for (Iterator<TaskId> it = standbyTasks.descendingIterator(); it.hasNext(); ) {
            TaskId task = it.next();
            for (int i = 0; i < numStandbyReplicas; i++) {

                // prev active task
                final Member prevMember = localState.activeTaskToPrevMember.get(task);
                if (prevMember != null) {
                    final ProcessState prevMemberProcessState = localState.processIdToState.get(prevMember.processId);
                    if (!prevMemberProcessState.hasTask(task) && isLoadBalanced(prevMemberProcessState)) {
                        prevMemberProcessState.addTask(prevMember.memberId, task, false);
                        continue;
                    }
                }

                // prev standby tasks
                final ArrayList<Member> prevMembers = localState.standbyTaskToPrevMember.get(task);
                if (prevMembers != null && !prevMembers.isEmpty()) {
                    final Member prevMember2 = findPrevMemberWithLeastLoad(prevMembers, task);
                    if (prevMember2 != null) {
                        final ProcessState prevMemberProcessState = localState.processIdToState.get(prevMember2.processId);
                        if (isLoadBalanced(prevMemberProcessState)) {
                            prevMemberProcessState.addTask(prevMember2.memberId, task, false);
                            continue;
                        }
                    }
                }

Review comment (suggestion):
The logic sorts standbyTasks in reverse order and then iterates using descendingIterator(), which effectively processes tasks in the original order. This double reversal is confusing and could be simplified by removing both the .reversed() and using a regular iterator.
```suggestion
        standbyTasks.sort(Comparator.comparing(TaskId::partition).thenComparing(TaskId::subtopologyId));

        for (Iterator<TaskId> it = standbyTasks.iterator(); it.hasNext(); ) {
```

Code after:

    private String memberWithLeastLoad(final ProcessState processWithLeastLoad) {
        final Map<String, Integer> members = processWithLeastLoad.memberToTaskCounts();
        if (members.isEmpty()) {
            return null;
        }
        if (members.size() == 1) {
            return members.keySet().iterator().next();
        }
        final Optional<String> memberWithLeastLoad = processWithLeastLoad.memberToTaskCounts().entrySet().stream()
            .min(Map.Entry.comparingByValue())
            .map(Map.Entry::getKey);
        return memberWithLeastLoad.orElse(null);
    }

    private boolean hasUnfulfilledQuota(final Member member) {
        return localState.processIdToState.get(member.processId).memberToTaskCounts().get(member.memberId) < localState.tasksPerMember;
    }

    private void assignStandby(final LinkedList<TaskId> standbyTasks, int numStandbyReplicas) {
        final ArrayList<StandbyToAssign> toLeastLoaded = new ArrayList<>(standbyTasks.size() * numStandbyReplicas);

        // Assuming our current assignment is range-based, we want to sort by partition first.
        standbyTasks.sort(Comparator.comparing(TaskId::partition).thenComparing(TaskId::subtopologyId).reversed());

        for (TaskId task : standbyTasks) {
            for (int i = 0; i < numStandbyReplicas; i++) {

                // prev active task
                final Member prevMember = localState.activeTaskToPrevMember.get(task);
                if (prevMember != null) {
                    final ProcessState prevMemberProcessState = localState.processIdToState.get(prevMember.processId);
                    if (!prevMemberProcessState.hasTask(task) && isLoadBalanced(prevMemberProcessState)) {
                        prevMemberProcessState.addTask(prevMember.memberId, task, false);
                        continue;
                    }
                }

                // prev standby tasks
                final ArrayList<Member> prevMembers = localState.standbyTaskToPrevMember.get(task);
                if (prevMembers != null && !prevMembers.isEmpty()) {
                    final Member prevMember2 = findPrevMemberWithLeastLoad(prevMembers, task);
                    if (prevMember2 != null) {
                        final ProcessState prevMemberProcessState = localState.processIdToState.get(prevMember2.processId);
                        if (isLoadBalanced(prevMemberProcessState)) {
                            prevMemberProcessState.addTask(prevMember2.memberId, task, false);
                            continue;
                        }
                    }
                }