@Override
  public SystemStreamPartition getPreviousSSP(SystemStreamPartition currentSystemStreamPartition, int previousPartitionCount, int currentPartitionCount) {
    Preconditions.checkNotNull(currentSystemStreamPartition);
    Preconditions.checkArgument(currentPartitionCount % previousPartitionCount == 0,
        String.format("New partition count: %d should be a multiple of previous partition count: %d.", currentPartitionCount, previousPartitionCount));
    Partition partition = currentSystemStreamPartition.getPartition();
    Preconditions.checkNotNull(partition, String.format("SystemStreamPartition: %s cannot have null partition", currentSystemStreamPartition));
    int currentPartitionId = partition.getPartitionId();
    int previousPartitionId = currentPartitionId % previousPartitionCount;
    return new SystemStreamPartition(currentSystemStreamPartition.getSystemStream(), new Partition(previousPartitionId));
  }
A SystemStreamPartitionMapper that uses the hash partitioning function of the producer to map a {@link SystemStreamPartition} to
correct previous {@link SystemStreamPartition} after the stream expansion.
samza-core/src/main/java/org/apache/samza/container/grouper/stream/HashSystemStreamPartitionMapper.java