public static void optimizeRackAwareActiveTasks(final RackAwareOptimizationParams optimizationParams,
                                                    final Map<ProcessId, KafkaStreamsAssignment> kafkaStreamsAssignments) {
        final ApplicationState applicationState = optimizationParams.applicationState;
        final SortedSet<TaskId> activeTasksToOptimize = getTasksToOptimize(kafkaStreamsAssignments, optimizationParams, AssignedTask.Type.ACTIVE);
        if (activeTasksToOptimize.isEmpty()) {
            return;
        }

        if (!canPerformRackAwareOptimization(applicationState, optimizationParams, AssignedTask.Type.ACTIVE)) {
            return;
        }

        initializeAssignmentsForAllClients(applicationState, kafkaStreamsAssignments);

        final int crossRackTrafficCost =
            optimizationParams.trafficCostOverride.orElseGet(() -> applicationState.assignmentConfigs()
                .rackAwareTrafficCost()
                .getAsInt());
        final int nonOverlapCost =
            optimizationParams.nonOverlapCostOverride.orElseGet(() -> applicationState.assignmentConfigs()
                .rackAwareNonOverlapCost()
                .getAsInt());

        final Map<ProcessId, KafkaStreamsState> kafkaStreamsStates = applicationState.kafkaStreamsStates(false);
        final List<TaskId> taskIds = new ArrayList<>(activeTasksToOptimize);

        final Map<TaskId, Set<TaskTopicPartition>> topicPartitionsByTaskId = applicationState.allTasks().values().stream()
            .filter(taskInfo -> activeTasksToOptimize.contains(taskInfo.id()))
            .collect(Collectors.toMap(TaskInfo::id, TaskInfo::topicPartitions));

        final List<ProcessId> clientIds = new ArrayList<>(kafkaStreamsStates.keySet());
        final long initialCost = computeTotalAssignmentCost(
            topicPartitionsByTaskId,
            taskIds,
            clientIds,
            kafkaStreamsAssignments,
            kafkaStreamsStates,
            crossRackTrafficCost,
            nonOverlapCost,
            false,
            false
        );

        LOG.info("Assignment before active task optimization has cost {}", initialCost);

        final RackAwareGraphConstructor<KafkaStreamsAssignment> graphConstructor = RackAwareGraphConstructorFactory.create(
            applicationState.assignmentConfigs().rackAwareAssignmentStrategy(), taskIds);

        final AssignmentGraph assignmentGraph = buildTaskGraph(
            kafkaStreamsAssignments,
            kafkaStreamsStates,
            taskIds,
            clientIds,
            topicPartitionsByTaskId,
            crossRackTrafficCost,
            nonOverlapCost,
            false,
            false,
            graphConstructor
        );

        assignmentGraph.graph.solveMinCostFlow();

        graphConstructor.assignTaskFromMinCostFlow(
            assignmentGraph.graph,
            clientIds,
            taskIds,
            kafkaStreamsAssignments,
            assignmentGraph.taskCountByClient,
            assignmentGraph.clientByTask,
            (assignment, taskId) -> assignment.assignTask(new AssignedTask(taskId, AssignedTask.Type.ACTIVE)),
            (assignment, taskId) -> assignment.removeTask(new AssignedTask(taskId, AssignedTask.Type.ACTIVE)),
            (assignment, taskId) -> assignment.tasks().containsKey(taskId) && assignment.tasks().get(taskId).type() == AssignedTask.Type.ACTIVE
        );
    }
Optimize active task assignment for rack awareness. This optimization is based on the
{@link StreamsConfig#RACK_AWARE_ASSIGNMENT_TRAFFIC_COST_CONFIG trafficCost}
and {@link StreamsConfig#RACK_AWARE_ASSIGNMENT_NON_OVERLAP_COST_CONFIG nonOverlapCost}
configs which balance cross rack traffic minimization and task movement.
Setting {@code trafficCost} to a larger number reduces the overall cross rack traffic of the resulting
assignment, but can increase the number of tasks shuffled around between clients.
Setting {@code nonOverlapCost} to a larger number increases the affinity of tasks to their intended client
and reduces the amount by which the rack-aware optimization can shuffle tasks around, at the cost of higher
cross-rack traffic.
In an extreme case, if we set {@code nonOverlapCost} to 0 and @{code trafficCost} to a positive value,
the resulting assignment will have an absolute minimum of cross rack traffic. If we set {@code trafficCost} to 0,
and {@code nonOverlapCost} to a positive value, the resulting assignment will be identical to the input assignment.
<p>
Note: this method will modify the input {@link KafkaStreamsAssignment} objects and return the same map.
It does not make a copy of the map or the KafkaStreamsAssignment objects.
<p>
This method optimizes cross-rack traffic for active tasks only. For standby task optimization,
use {@link #optimizeRackAwareStandbyTasks}.
<p>
It is recommended to run this optimization before assigning any standby tasks, especially if you have configured
your KafkaStreams clients with assignment tags via the rack.aware.assignment.tags config since this method may
shuffle around active tasks without considering the client tags and can result in a violation of the original
client tag assignment's constraints.

@param optimizationParams      optional configuration parameters to apply
@param kafkaStreamsAssignments the current assignment of tasks to KafkaStreams clients
streams/src/main/java/org/apache/kafka/streams/processor/assignment/TaskAssignmentUtils.java