@Override
    public <W extends Window> TimeWindowedKStream<K, V> windowedBy(final Windows<W> windows) {

        return new TimeWindowedKStreamImpl<>(
            windows,
            builder,
            subTopologySourceNodes,
            name,
            keySerde,
            valueSerde,
            aggregateBuilder,
            graphNode
        );
    }
@Test
    public void shouldNotAcceptNullSessionWindowsReducingSessionWindows() {
        assertThrows(NullPointerException.class, () ->  groupedStream.windowedBy((SessionWindows) null));
    }