@Test
public void shouldMeterOnSkippedRecordsWithNullValue() {

    StreamsBuilder builder = new StreamsBuilder();
    ConsumerRecordFactory<String, String> recordFactory = new ConsumerRecordFactory<>(
        new StringSerializer(),
        new StringSerializer()
    );
    KStream<String, String> stream = builder.stream(TOPIC_TEST_1);
    builder.addStateStore(nfaStateStore);
    builder.addStateStore(eventBufferStore);
    builder.addStateStore(aggregateStateStores);
    final String[] stateStoreNames = new String[]{
        nfaStateStore.name(),
        eventBufferStore.name(),
        aggregateStateStores.name()
    };
    stream.process(() -> new CEPProcessor<>(TEST_QUERY, pattern), stateStoreNames);
    try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), this.props)) {
        driver.pipeInput(recordFactory.create(TOPIC_TEST_1, "A", (String)null));
        Assert.assertEquals(1.0D,
        StreamsTestUtils.getMetricByName(
            driver.metrics(),
            "skipped-records-total", "stream-metrics")
        .metricValue());
    }
}