BroadcastPartitioner extends StreamPartitioner<T> { @Override public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record, int numberOfOutputChannels) { if (set && setNumber == numberOfOutputChannels) { return returnArray; } else { this.returnArray = new int[numberOfOutputChannels]; for (int i = 0; i < numberOfOutputChannels; i++) { returnArray[i] = i; } set = true; setNumber = numberOfOutputChannels; return returnArray; } }  @Override int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
			int numberOfOutputChannels); @Override StreamPartitioner<T> copy(); @Override String toString();  }
@Test public void testSelectChannels() { int[] first = new int[] { 0 }; int[] second = new int[] { 0, 1 }; int[] sixth = new int[] { 0, 1, 2, 3, 4, 5 }; sd.setInstance(streamRecord); assertArrayEquals(first, broadcastPartitioner1.selectChannels(sd, 1)); assertArrayEquals(second, broadcastPartitioner2.selectChannels(sd, 2)); assertArrayEquals(sixth, broadcastPartitioner3.selectChannels(sd, 6)); }