class DefaultStateConstrainedBalancedAssignor implements StateConstrainedBalancedAssignor {

    @Override
    public Map<UUID, List<TaskId>> assign(final SortedMap<TaskId, SortedSet<RankedClient>> statefulTasksToRankedClients,
                                          final int balanceFactor,
                                          final Set<UUID> clients,
                                          final Map<UUID, Integer> clientsToNumberOfStreamThreads,
                                          final Map<TaskId, SortedSet<UUID>> tasksToCaughtUpClients) {
        checkClientsAndNumberOfStreamThreads(clientsToNumberOfStreamThreads, clients);
        final Map<UUID, List<TaskId>> assignment = initAssignment(clients);
        assignTasksWithCaughtUpClients(
            assignment,
            tasksToCaughtUpClients,
            statefulTasksToRankedClients
        );
        assignTasksWithoutCaughtUpClients(
            assignment,
            tasksToCaughtUpClients,
            statefulTasksToRankedClients
        );
        balance(
            assignment,
            balanceFactor,
            statefulTasksToRankedClients,
            tasksToCaughtUpClients,
            clientsToNumberOfStreamThreads
        );
        return assignment;
    }

}