class DefaultBalancedAssignor implements BalancedAssignor {

    @Override
    public Map<UUID, List<TaskId>> assign(final SortedSet<UUID> clients,
                                          final SortedSet<TaskId> tasks,
                                          final Map<UUID, Integer> clientsToNumberOfStreamThreads) {
        final Map<UUID, List<TaskId>> assignment = new HashMap<>();
        clients.forEach(client -> assignment.put(client, new ArrayList<>()));
        distributeTasksEvenlyOverClients(assignment, clients, tasks);
        balanceTasksOverStreamThreads(assignment, clients, clientsToNumberOfStreamThreads);
        return assignment;
    }

}