@Override
  public Map<TaskName, Set<SystemStreamPartition>> group(Set<SystemStreamPartition> ssps) {
    Map<TaskName, Set<SystemStreamPartition>> groupedMap = new HashMap<TaskName, Set<SystemStreamPartition>>();

    for (SystemStreamPartition ssp : ssps) {
      if (broadcastStreams.contains(ssp)) {
        continue;
      }

      // if elasticity factor > 1 then elasticity is enabled
      // for each ssp create ElasticityFactor number of tasks
      // i.e; result will have number of tasks =  ElasticityFactor X number of SSP
      // each task will have name SSP[system,stream,partition,keyBucket] keyBucket <= elasticityFactor
      // each task portion corresponding to keyBucket of the SSP.
      for (int i = 0; i < elasticityFactor; i++) {
        int keyBucket = elasticityFactor == 1 ? -1 : i;
        SystemStreamPartition sspWithKeyBucket = new SystemStreamPartition(ssp, keyBucket);
        HashSet<SystemStreamPartition> sspSet = new HashSet<SystemStreamPartition>();
        sspSet.add(sspWithKeyBucket);
        groupedMap.put(new TaskName(sspWithKeyBucket.toString()), sspSet);
      }
    }

    // assign the broadcast streams to all the taskNames
    if (!broadcastStreams.isEmpty()) {
      for (Set<SystemStreamPartition> value : groupedMap.values()) {
        value.addAll(broadcastStreams);
      }
    }

    return groupedMap;
  }
An implementation of {@link SystemStreamPartitionGrouper} that assigns each input {@link SystemStreamPartition} to a separate task.
Provides increased parallelism in message processing within a container. Useful in message processing scenarios involving remote I/O.
samza-core/src/main/java/org/apache/samza/container/grouper/stream/GroupBySystemStreamPartition.java