@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());
}