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; } HashPartitioner(KeySelector<T, ?> keySelector); @Override int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
			int numberOfOutputChannels); @Override StreamPartitioner<T> copy(); @Override String toString();  }
@Test public void testSelectChannelsLength() { sd1.setInstance(streamRecord1); assertEquals(1, hashPartitioner.selectChannels(sd1, 1).length); assertEquals(1, hashPartitioner.selectChannels(sd1, 2).length); assertEquals(1, hashPartitioner.selectChannels(sd1, 1024).length); }
@Test public void testSelectChannelsGrouping() { sd1.setInstance(streamRecord1); sd2.setInstance(streamRecord2); assertArrayEquals(hashPartitioner.selectChannels(sd1, 1), hashPartitioner.selectChannels(sd2, 1)); assertArrayEquals(hashPartitioner.selectChannels(sd1, 2), hashPartitioner.selectChannels(sd2, 2)); assertArrayEquals(hashPartitioner.selectChannels(sd1, 1024), hashPartitioner.selectChannels(sd2, 1024)); }