KeyGroupStreamPartitioner extends StreamPartitioner<T> implements ConfigurableStreamPartitioner { @Override public int[] selectChannels( SerializationDelegate<StreamRecord<T>> record, int numberOfOutputChannels) { K 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] = KeyGroupRangeAssignment.assignKeyToParallelOperator(key, maxParallelism, numberOfOutputChannels); return returnArray; } KeyGroupStreamPartitioner(KeySelector<T, K> keySelector, int maxParallelism); int getMaxParallelism(); @Override int[] selectChannels(
		SerializationDelegate<StreamRecord<T>> record,
		int numberOfOutputChannels); @Override StreamPartitioner<T> copy(); @Override String toString(); @Override void configure(int maxParallelism);  }
@Test public void testSelectChannelsLength() { sd1.setInstance(streamRecord1); assertEquals(1, keyGroupPartitioner.selectChannels(sd1, 1).length); assertEquals(1, keyGroupPartitioner.selectChannels(sd1, 2).length); assertEquals(1, keyGroupPartitioner.selectChannels(sd1, 1024).length); }
@Test public void testSelectChannelsGrouping() { sd1.setInstance(streamRecord1); sd2.setInstance(streamRecord2); assertArrayEquals(keyGroupPartitioner.selectChannels(sd1, 1), keyGroupPartitioner.selectChannels(sd2, 1)); assertArrayEquals(keyGroupPartitioner.selectChannels(sd1, 2), keyGroupPartitioner.selectChannels(sd2, 2)); assertArrayEquals(keyGroupPartitioner.selectChannels(sd1, 1024), keyGroupPartitioner.selectChannels(sd2, 1024)); }