@Override
    public KGroupedStream<K, V> groupByKey() {
        return groupByKey(Grouped.with(keySerde, valueSerde));
    }
@Test
    public void shouldNotAllowNullGroupedOnGroupByKey() {
        final NullPointerException exception = assertThrows(
            NullPointerException.class,
            () -> testStream.groupByKey((Grouped<String, String>) null));
        assertThat(exception.getMessage(), equalTo("grouped can't be null"));
    }