public static <K> byte[] toBinary(final Windowed<K> timeKey,
                                      final Serializer<K> serializer,
                                      final String topic) {
        final byte[] bytes = serializer.serialize(topic, timeKey.key());
        final ByteBuffer buf = ByteBuffer.allocate(bytes.length + TIMESTAMP_SIZE);
        buf.put(bytes);
        buf.putLong(timeKey.window().start());

        return buf.array();
    }
Safely construct a time window of the given size,
taking care of bounding endMs to Long.MAX_VALUE if necessary
streams/src/main/java/org/apache/kafka/streams/state/internals/WindowKeySchema.java