default void validateAssignedTask(
        final List<TaskId> taskIdList,
        final int tasksAssigned,
        final Map<ProcessId, T> clientStates,
        final Map<ProcessId, Integer> originalAssignedTaskNumber,
        final BiPredicate<T, TaskId> hasAssignedTask
    ) {
        // Validate task assigned
        if (tasksAssigned != taskIdList.size()) {
            throw new IllegalStateException("Computed active task assignment number "
                + tasksAssigned + " is different size " + taskIdList.size());
        }

        // Validate original assigned task number matches
        final Map<ProcessId, Integer> assignedTaskNumber = new HashMap<>();
        for (final TaskId taskId : taskIdList) {
            for (final Entry<ProcessId, T> clientState : clientStates.entrySet()) {
                if (hasAssignedTask.test(clientState.getValue(), taskId)) {
                    assignedTaskNumber.merge(clientState.getKey(), 1, Integer::sum);
                }
            }
        }

        if (originalAssignedTaskNumber.size() != assignedTaskNumber.size()) {
            throw new IllegalStateException("There are " + originalAssignedTaskNumber.size() + " clients have "
                + " active tasks before assignment, but " + assignedTaskNumber.size() + " clients have"
                + " active tasks after assignment");
        }

        for (final Entry<ProcessId, Integer> originalCapacity : originalAssignedTaskNumber.entrySet()) {
            final int capacity = assignedTaskNumber.getOrDefault(originalCapacity.getKey(), 0);
            if (!Objects.equals(originalCapacity.getValue(), capacity)) {
                throw new IllegalStateException("There are " + originalCapacity.getValue() + " tasks assigned to"
                    + " client " + originalCapacity.getKey() + " before assignment, but " + capacity + " tasks "
                    + " are assigned to it after assignment");
            }
        }
    }
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