class HashPartitioner extends StreamPartitioner<T> {

	@Override
	public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
			int numberOfOutputChannels) {
		Object key;
		try {
			key = keySelector.getKey(record.getInstance().getValue());
		} catch (Exception e) {
			throw new RuntimeException("Could not extract key from " + record.getInstance().getValue(), e);
		}
		returnArray[0] = MathUtils.murmurHash(key.hashCode()) % numberOfOutputChannels;

		return returnArray;
	}

	public  HashPartitioner(KeySelector<T, ?> keySelector);

}