@Override
    public KStream<K, V> filter(final Predicate<? super K, ? super V> predicate) {
        return filter(predicate, NamedInternal.empty());
    }
@Test
    public void shouldNotAllowNullPredicateOnFilterWithNamed() {
        final NullPointerException exception = assertThrows(
            NullPointerException.class,
            () -> testStream.filter(null, Named.as("filter")));
        assertThat(exception.getMessage(), equalTo("predicate can't be null"));
    }