@Override
    public <VR> KStream<K, VR> flatMapValues(final ValueMapper<? super V, ? extends Iterable<? extends VR>> mapper) {
        return flatMapValues(withKey(mapper));
    }
@Test
    public void shouldNotAllowNullMapperOnFlatMapValuesWithKey() {
        final NullPointerException exception = assertThrows(
            NullPointerException.class,
            () -> testStream.flatMapValues((ValueMapperWithKey<Object, Object, ? extends Iterable<Object>>) null));
        assertThat(exception.getMessage(), equalTo("valueMapper can't be null"));
    }