private static AssignmentGraph buildTaskGraph(final Map<ProcessId, KafkaStreamsAssignment> assignments,
                                                  final Map<ProcessId, KafkaStreamsState> clientStates,
                                                  final List<TaskId> taskIds,
                                                  final List<ProcessId> clientList,
                                                  final Map<TaskId, Set<TaskTopicPartition>> topicPartitionsByTaskId,
                                                  final int crossRackTrafficCost,
                                                  final int nonOverlapCost,
                                                  final boolean hasReplica,
                                                  final boolean isStandby,
                                                  final RackAwareGraphConstructor<KafkaStreamsAssignment> graphConstructor) {
        // Intentionally passed in empty -- these are actually outputs of the graph
        final Map<TaskId, ProcessId> clientByTask = new HashMap<>();
        final Map<ProcessId, Integer> taskCountByClient = new HashMap<>();

        final AssignedTask.Type taskType = isStandby ? AssignedTask.Type.STANDBY : AssignedTask.Type.ACTIVE;
        final Graph<Integer> graph = graphConstructor.constructTaskGraph(
            clientList,
            taskIds,
            assignments,
            clientByTask,
            taskCountByClient,
            (assignment, taskId) -> assignment.tasks().containsKey(taskId) && assignment.tasks().get(taskId).type() == taskType,
            (taskId, processId, inCurrentAssignment, unused0, unused1, unused2) -> {
                final String clientRack = clientStates.get(processId).rackId().get();
                final int assignmentChangeCost = !inCurrentAssignment ? nonOverlapCost : 0;
                final int trafficCost = getCrossRackTrafficCost(topicPartitionsByTaskId.get(taskId), clientRack, crossRackTrafficCost);
                return assignmentChangeCost + trafficCost;
            },
            crossRackTrafficCost,
            nonOverlapCost,
            hasReplica,
            isStandby
        );
        return new AssignmentGraph(graph, clientByTask, taskCountByClient);
    }
Optimize standby 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 standby tasks only. For active task optimization,
use {@link #optimizeRackAwareActiveTasks}.

@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