@Override
    @Deprecated
    public Integer partition(final String topic, final Windowed<K> windowedKey, final V value, final int numPartitions) {
        // for windowed key, the key bytes should never be null
        final byte[] keyBytes = serializer.serializeBaseKey(topic, windowedKey);

        // stick with the same built-in partitioner util functions that producer used
        // to make sure its behavior is consistent with the producer
        return BuiltInPartitioner.partitionForKey(keyBytes, numPartitions);
    }
@Test
    public void testCopartitioning() {
        final Random rand = new Random();
        @SuppressWarnings("deprecation")
        final org.apache.kafka.clients.producer.internals.DefaultPartitioner defaultPartitioner = new org.apache.kafka.clients.producer.internals.DefaultPartitioner();
        final WindowedSerializer<Integer> timeWindowedSerializer = new TimeWindowedSerializer<>(intSerializer);
        final WindowedStreamPartitioner<Integer, String> streamPartitioner = new WindowedStreamPartitioner<>(timeWindowedSerializer);

        for (int k = 0; k < 10; k++) {
            final Integer key = rand.nextInt();
            final byte[] keyBytes = intSerializer.serialize(topicName, key);

            final String value = key.toString();
            final byte[] valueBytes = stringSerializer.serialize(topicName, value);

            final Integer expected = defaultPartitioner.partition("topic", key, keyBytes, value, valueBytes, cluster);

            for (int w = 1; w < 10; w++) {
                final TimeWindow window = new TimeWindow(10 * w, 20 * w);

                final Windowed<Integer> windowedKey = new Windowed<>(key, window);
                @SuppressWarnings("deprecation")
                final Integer actual = streamPartitioner.partition(topicName, windowedKey, value, infos.size());

                assertEquals(expected, actual);
            }
        }

        defaultPartitioner.close();
    }