default KeyValue<Boolean, Integer> assignTaskToClient(
        final Graph<Integer> graph,
        final TaskId taskId,
        final int taskNodeId,
        final int topicGroupIndex,
        final Map<ProcessId, T> clientStates,
        final List<ProcessId> clientList,
        final List<TaskId> taskIdList,
        final Map<TaskId, ProcessId> taskClientMap,
        final BiConsumer<T, TaskId> assignTask,
        final BiConsumer<T, TaskId> unAssignTask
    ) {
        int tasksAssigned = 0;
        boolean taskMoved = false;
        final Map<Integer, Graph<Integer>.Edge> edges = graph.edges(taskNodeId);
        for (final Graph<Integer>.Edge edge : edges.values()) {
            if (edge.flow > 0) {
                tasksAssigned++;
                final int clientIndex = getClientIndex(edge.destination, taskIdList, clientList, topicGroupIndex);
                final ProcessId processId = clientList.get(clientIndex);
                final ProcessId originalProcessId = taskClientMap.get(taskId);

                // Don't need to assign this task to other client
                if (processId.equals(originalProcessId)) {
                    break;
                }

                unAssignTask.accept(clientStates.get(originalProcessId), taskId);
                assignTask.accept(clientStates.get(processId), taskId);
                taskMoved = true;
            }
        }
        return KeyValue.pair(taskMoved, tasksAssigned);
    }
Construct graph for rack aware task assignor
@param <T> represents a KafkaStreams client and its currently-assigned tasks
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/RackAwareGraphConstructor.java