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