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