public long optimizeActiveTasks(final SortedSet<TaskId> activeTasks,
                                    final SortedMap<ProcessId, ClientState> clientStates,
                                    final int trafficCost,
                                    final int nonOverlapCost) {
        if (activeTasks.isEmpty()) {
            return 0;
        }

        log.info("Assignment before active task optimization is {}\n with cost {}", clientStates,
            activeTasksCost(activeTasks, clientStates, trafficCost, nonOverlapCost));

        final long startTime = time.milliseconds();
        final List<ProcessId> clientList = new ArrayList<>(clientStates.keySet());
        final List<TaskId> taskIdList = new ArrayList<>(activeTasks);
        final Map<TaskId, ProcessId> taskClientMap = new HashMap<>();
        final Map<ProcessId, Integer> originalAssignedTaskNumber = new HashMap<>();
        final RackAwareGraphConstructor<ClientState> graphConstructor = RackAwareGraphConstructorFactory.create(assignmentConfigs, tasksForTopicGroup);
        final Graph<Integer> graph = graphConstructor.constructTaskGraph(
            clientList,
            taskIdList,
            clientStates,
            taskClientMap,
            originalAssignedTaskNumber,
            ClientState::hasActiveTask,
            this::getCost,
            trafficCost,
            nonOverlapCost,
            false,
            false
        );

        graph.solveMinCostFlow();
        final long cost = graph.totalCost();

        graphConstructor.assignTaskFromMinCostFlow(graph, clientList, taskIdList, clientStates, originalAssignedTaskNumber,
            taskClientMap, ClientState::assignActive, ClientState::unassignActive, ClientState::hasActiveTask);

        final long duration = time.milliseconds() - startTime;
        log.info("Assignment after {} milliseconds for active task optimization is {}\n with cost {}", 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