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

Code before:
            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 Set<TaskId> standbyTasks, final int numStandbyReplicas) {
        final ArrayList<StandbyToAssign> toLeastLoaded = new ArrayList<>(standbyTasks.size() * numStandbyReplicas);
        for (final 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;
                        }
                    }
                }

                toLeastLoaded.add(new StandbyToAssign(task, numStandbyReplicas - i));
                break;
            }
        }

        final PriorityQueue<ProcessState> processByLoad = new PriorityQueue<>(Comparator.comparingDouble(ProcessState::load));
        processByLoad.addAll(localState.processIdToState.values());
        for (final StandbyToAssign toAssign : toLeastLoaded) {

Review comment (suggestion):
The `.reversed()` is applied to the entire comparator chain, which reverses both partition and subtopology ordering. This may not be the intended behavior. Consider applying `.reversed()` only to the specific comparison that needs reversing, or clarify the intention with a comment.
```suggestion
        // Sort by partition ascending, then subtopology descending (if that is the intention)
```

Code after:
        return null;
    }

    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;
                        }
                    }