CEPProcessor implements Processor<K, V> { @Override public void process(final K key, final V value) { List<Sequence<K, V>> sequences = match(key, value); sequences.forEach(seq -> context().forward(key, seq)); } CEPProcessor(final String queryName,
                        final Pattern<K, V> pattern);   CEPProcessor(final String queryName,
                 final Stages<K, V> stages); @Override @SuppressWarnings("unchecked") void init(final ProcessorContext context); @Override void process(final K key, final V value); List<Sequence<K, V>> match(final K key, final V value); @Override void close();  }
@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()); } }
@Test public void shouldForwardMatchingSequencesToDownStreamsProcessors() { StreamsBuilder builder = new StreamsBuilder(); KStream<String, String> stream = builder.stream(Arrays.asList(TOPIC_TEST_1, TOPIC_TEST_2)); builder.addStateStore(nfaStateStore); builder.addStateStore(eventBufferStore); builder.addStateStore(aggregateStateStores); ConsumerRecordFactory<String, String> recordFactory = new ConsumerRecordFactory<>( new StringSerializer(), new StringSerializer() ); final String[] stateStoreNames = new String[]{ nfaStateStore.name(), eventBufferStore.name(), aggregateStateStores.name() }; CapturedCEProcessor<String, String> processor = new CapturedCEProcessor<>(TEST_QUERY, pattern); stream.process(() -> processor, stateStoreNames); try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), this.props)) { driver.pipeInput(recordFactory.create(TOPIC_TEST_1, TEST_RECORD_KEY, TEST_RECORD_VALUE)); driver.pipeInput(recordFactory.create(TOPIC_TEST_2, TEST_RECORD_KEY, TEST_RECORD_VALUE)); } List<CapturedProcessorContext.CapturedForward> capturedForward = processor.context.capturedForward; Assert.assertEquals(2, capturedForward.size()); }