public long optimizeStandbyTasks(final SortedMap<ProcessId, ClientState> clientStates,
                                     final int trafficCost,
                                     final int nonOverlapCost,
                                     final MoveStandbyTaskPredicate moveStandbyTask) {
        final BiFunction<ClientState, ClientState, List<TaskId>> getMovableTasks = (source, destination) -> source.standbyTasks().stream()
            .filter(task -> !destination.hasAssignedTask(task))
            .filter(task -> moveStandbyTask.canMove(source, destination, task, clientStates))
            .sorted()
            .collect(Collectors.toList());

        final long startTime = time.milliseconds();
        final List<ProcessId> clientList = new ArrayList<>(clientStates.keySet());
        final SortedSet<TaskId> standbyTasks = new TreeSet<>();
        clientStates.values().forEach(clientState -> standbyTasks.addAll(clientState.standbyTasks()));

        log.info("Assignment before standby task optimization is {}\n with cost {}", clientStates,
            standByTasksCost(standbyTasks, clientStates, trafficCost, nonOverlapCost));

        boolean taskMoved = true;
        int round = 0;
        final RackAwareGraphConstructor<ClientState> graphConstructor = new MinTrafficGraphConstructor<>();
        while (taskMoved && round < STANDBY_OPTIMIZER_MAX_ITERATION) {
            taskMoved = false;
            round++;
            for (int i = 0; i < clientList.size(); i++) {
                final ClientState clientState1 = clientStates.get(clientList.get(i));
                for (int j = i + 1; j < clientList.size(); j++) {
                    final ClientState clientState2 = clientStates.get(clientList.get(j));

                    final String rack1 = rackForProcess.get(clientState1.processId());
                    final String rack2 = rackForProcess.get(clientState2.processId());
                    // Cross rack traffic can not be reduced if racks are the same
                    if (rack1.equals(rack2)) {
                        continue;
                    }

                    final List<TaskId> movable1 = getMovableTasks.apply(clientState1, clientState2);
                    final List<TaskId> movable2 = getMovableTasks.apply(clientState2, clientState1);

                    // There's no needed to optimize if one is empty because the optimization
                    // can only swap tasks to keep the client's load balanced
                    if (movable1.isEmpty() || movable2.isEmpty()) {
                        continue;
                    }

                    final List<TaskId> taskIdList = Stream.concat(movable1.stream(),
                            movable2.stream())
                        .sorted()
                        .collect(Collectors.toList());

                    final List<ProcessId> clients = Stream.of(clientList.get(i), clientList.get(j))
                        .sorted().collect(
                            Collectors.toList());

                    final Map<TaskId, ProcessId> taskClientMap = new HashMap<>();
                    final Map<ProcessId, Integer> originalAssignedTaskNumber = new HashMap<>();
                    final Graph<Integer> graph = graphConstructor.constructTaskGraph(
                        clients,
                        taskIdList,
                        clientStates,
                        taskClientMap,
                        originalAssignedTaskNumber,
                        ClientState::hasStandbyTask,
                        this::getCost,
                        trafficCost,
                        nonOverlapCost,
                        true,
                        true
                    );
                    graph.solveMinCostFlow();

                    taskMoved |= graphConstructor.assignTaskFromMinCostFlow(graph, clients, taskIdList, clientStates,
                        originalAssignedTaskNumber,
                        taskClientMap, ClientState::assignStandby, ClientState::unassignStandby,
                        ClientState::hasStandbyTask);
                }
            }
        }
        final long cost = standByTasksCost(standbyTasks, clientStates, trafficCost, nonOverlapCost);

        final long duration = time.milliseconds() - startTime;
        log.info("Assignment after {} rounds and {} milliseconds for standby task optimization is {}\n with cost {}", round, duration, clientStates, cost);
        return cost;
    }
Optimize active task assignment for rack awareness. canEnableRackAwareAssignor must be called first.
{@code trafficCost} and {@code nonOverlapCost} balance cross rack traffic optimization and task movement.
If we set {@code trafficCost} to a larger number, we are more likely to compute an assignment with less
cross rack traffic. However, tasks may be shuffled a lot across clients. If we set {@code nonOverlapCost}
to a larger number, we are more likely to compute an assignment with similar to input assignment. However,
cross rack traffic can be higher. In extreme case, if we set {@code nonOverlapCost} to 0 and @{code trafficCost}
to a positive value, the computed assignment will be minimum for cross rack traffic. If we set {@code trafficCost} to 0,
and {@code nonOverlapCost} to a positive value, the computed assignment should be the same as input
@param activeTasks Tasks to reassign if needed. They must be assigned already in clientStates
@param clientStates Client states
@param trafficCost Cost of cross rack traffic for each TopicPartition
@param nonOverlapCost Cost of assign a task to a different client
@return Total cost after optimization
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/RackAwareTaskAssignor.java