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

Code before:
                        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) {
            for (int i = 0; i < toAssign.remainingReplicas; i++) {
                if (!assignStandbyToMemberWithLeastLoad(processByLoad, toAssign.taskId)) {
                    log.warn("{} There is not enough available capacity. " +
                            "You should increase the number of threads and/or application instances to maintain the requested number of standby replicas.",
                        errorMessage(numStandbyReplicas, i, toAssign.taskId));
                    break;
                }
            }
        }
    }

    private String errorMessage(final int numStandbyReplicas, final int i, final TaskId task) {
        return "Unable to assign " + (numStandbyReplicas - i) +
            " of " + numStandbyReplicas + " standby tasks for task [" + task + "].";
    }

    private boolean isLoadBalanced(final ProcessState process) {
        final double load = process.load();
        final boolean isLeastLoadedProcess = localState.processIdToState.values().stream()
            .allMatch(p -> p.load() >= load);
        return process.hasCapacity() || isLeastLoadedProcess;
    }

    private static int computeTasksPerMember(final int numberOfTasks, final int numberOfMembers) {
        if (numberOfMembers == 0) {
            return 0;
        }
        int tasksPerMember = numberOfTasks / numberOfMembers;
        if (numberOfTasks % numberOfMembers > 0) {
            tasksPerMember++;
        }
        return tasksPerMember;
    }

    static class StandbyToAssign {

Review comment (suggestion):
Similar to the previous sorting, `.reversed()` is applied to the entire comparator chain. This reverses both subtopology and partition ordering, which may not be the intended behavior for achieving range-based assignment.
```suggestion
            .thenComparing(x -> x.taskId.partition(), Comparator.reverseOrder()));
```

Code after:
                        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;
            }
        }

        // To achieve a range-based assignment, sort by subtopology
        toLeastLoaded.sort(Comparator.<StandbyToAssign, String>comparing(x -> x.taskId.subtopologyId())
            .thenComparing(x -> x.taskId.partition()).reversed());

        final PriorityQueue<ProcessState> processByLoad = new PriorityQueue<>(Comparator.comparingDouble(ProcessState::load));
        processByLoad.addAll(localState.processIdToState.values());
        for (final StandbyToAssign toAssign : toLeastLoaded) {
            for (int i = 0; i < toAssign.remainingReplicas; i++) {
                if (!assignStandbyToMemberWithLeastLoad(processByLoad, toAssign.taskId)) {
                    log.warn("{} There is not enough available capacity. " +
                            "You should increase the number of threads and/or application instances to maintain the requested number of standby replicas.",
                        errorMessage(numStandbyReplicas, i, toAssign.taskId));
                    break;
                }
            }
        }
    }

    private String errorMessage(final int numStandbyReplicas, final int i, final TaskId task) {
        return "Unable to assign " + (numStandbyReplicas - i) +
            " of " + numStandbyReplicas + " standby tasks for task [" + task + "].";
    }

    private boolean isLoadBalanced(final ProcessState process) {
        final double load = process.load();
        final boolean isLeastLoadedProcess = localState.processIdToState.values().stream()
            .allMatch(p -> p.load() >= load);
        return process.hasCapacity() || isLeastLoadedProcess;