public static <T> Serde<Windowed<T>> timeWindowedSerdeFrom(final Class<T> type, final long windowSize) {
        return new TimeWindowedSerde<>(Serdes.serdeFrom(type), windowSize);
    }
@Test
    public void testTimeWindowSerdeFrom() {
        final Windowed<Integer> timeWindowed = new Windowed<>(10, new TimeWindow(0, Long.MAX_VALUE));
        final Serde<Windowed<Integer>> timeWindowedSerde = WindowedSerdes.timeWindowedSerdeFrom(Integer.class, Long.MAX_VALUE);
        final byte[] bytes = timeWindowedSerde.serializer().serialize(topic, timeWindowed);
        final Windowed<Integer> windowed = timeWindowedSerde.deserializer().deserialize(topic, bytes);
        assertEquals(timeWindowed, windowed);
    }