@Override
  public Set<ContainerModel> group(Set<TaskModel> tasks, List<String> containerIds) {
    if (containerIds == null)
      return this.group(tasks);

    if (containerIds.isEmpty())
      throw new IllegalArgumentException("Must have at least one container");

    if (tasks.isEmpty())
      throw new IllegalArgumentException("cannot group an empty set. containerIds=" + Arrays
          .toString(containerIds.toArray()));

    if (containerIds.size() > tasks.size()) {
      LOG.warn("Number of containers: {} is greater than number of tasks: {}.",  containerIds.size(), tasks.size());
      /**
       * Choose lexicographically least `x` containerIds(where x = tasks.size()).
       */
      containerIds = containerIds.stream()
                                   .sorted()
                                   .limit(tasks.size())
                                   .collect(Collectors.toList());
      LOG.info("Generating containerModel with containers: {}.", containerIds);
    }

    int containerCount = containerIds.size();

    // Sort tasks by taskName.
    List<TaskModel> sortedTasks = new ArrayList<>(tasks);
    Collections.sort(sortedTasks);

    // Map every task to a container in round-robin fashion.
    Map<TaskName, TaskModel>[] taskGroups = new Map[containerCount];
    for (int i = 0; i < containerCount; i++) {
      taskGroups[i] = new HashMap<>();
    }
    for (int i = 0; i < sortedTasks.size(); i++) {
      TaskModel tm = sortedTasks.get(i);
      taskGroups[i % containerCount].put(tm.getTaskName(), tm);
    }

    // Convert to a Set of ContainerModel
    Set<ContainerModel> containerModels = new HashSet<>();
    for (int i = 0; i < containerCount; i++) {
      containerModels.add(new ContainerModel(containerIds.get(i), taskGroups[i]));
    }

    return Collections.unmodifiableSet(containerModels);
  }
{@inheritDoc}

When number of taskModels are less than number of available containerIds,
then chooses then selects the lexicographically least `x` containerIds.

Otherwise, assigns the tasks to the available containerIds in a round robin fashion
preserving the containerId in the final assignment.
samza-core/src/main/java/org/apache/samza/container/grouper/task/GroupByContainerIds.java