code: @Override
  	public final int[] selectChannels(SerializationDelegate<T> record, int numberOfChannels) {
  		switch (strategy) {
  		case FORWARD:
  		case PARTITION_RANDOM:
  			return robin(numberOfChannels);
  		case PARTITION_HASH:
  		case PARTITION_LOCAL_HASH:
  			return hashPartitionDefault(record.getInstance(), numberOfChannels);
  		case PARTITION_RANGE:
  			return rangePartition(record.getInstance(), numberOfChannels);
  		case BROADCAST:
  			return broadcast(numberOfChannels);
  		default:
  			throw new UnsupportedOperationException("Unsupported distribution strategy: " + strategy.name());
  		}
  	}
docs: ------------------------------------------------------------------------