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