### Question:
WindowedStreamPartitioner implements StreamPartitioner<Windowed<K>, V> { public Integer partition(final Windowed<K> windowedKey, final V value, final int numPartitions) { final byte[] keyBytes = serializer.serializeBaseKey(topic, windowedKey); return toPositive(Utils.murmur2(keyBytes)) % numPartitions; } WindowedStreamPartitioner(final String topic, final WindowedSerializer<K> serializer); Integer partition(final Windowed<K> windowedKey, final V value, final int numPartitions);  }### Answer:
@Test public void testCopartitioning() { Random rand = new Random(); DefaultPartitioner defaultPartitioner = new DefaultPartitioner(); WindowedSerializer<Integer> windowedSerializer = new WindowedSerializer<>(intSerializer); WindowedStreamPartitioner<Integer, String> streamPartitioner = new WindowedStreamPartitioner<>(topicName, windowedSerializer); for (int k = 0; k < 10; k++) { Integer key = rand.nextInt(); byte[] keyBytes = intSerializer.serialize(topicName, key); String value = key.toString(); byte[] valueBytes = stringSerializer.serialize(topicName, value); Integer expected = defaultPartitioner.partition("topic", key, keyBytes, value, valueBytes, cluster); for (int w = 1; w < 10; w++) { TimeWindow window = new TimeWindow(10 * w, 20 * w); Windowed<Integer> windowedKey = new Windowed<>(key, window); Integer actual = streamPartitioner.partition(windowedKey, value, infos.size()); assertEquals(expected, actual); } } }