@Override
    public <KR> KStream<KR, V> selectKey(final KeyValueMapper<? super K, ? super V, ? extends KR> mapper) {
        return selectKey(mapper, NamedInternal.empty());
    }
@Test
    public void shouldNotAllowNullMapperOnSelectKey() {
        final NullPointerException exception = assertThrows(
            NullPointerException.class,
            () -> testStream.selectKey(null));
        assertThat(exception.getMessage(), equalTo("mapper can't be null"));
    }