@Override
    public KTable<K, Long> count() {
        return doCount(NamedInternal.empty(), Materialized.with(keySerde, Serdes.Long()));
    }
@Test
    public void shouldLogAndMeasureSkipsInAggregate() {
        groupedStream.count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("count").withKeySerde(Serdes.String()));

        try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(KStreamAggregate.class);
             final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {

            processData(driver);

            assertThat(
                appender.getMessages(),
                hasItem("Skipping record due to null key or value. topic=[topic] partition=[0] "
                    + "offset=[6]")
            );
        }
    }