@Override
    @Deprecated
    public void process(final org.apache.kafka.streams.processor.ProcessorSupplier<? super K, ? super V> processorSupplier,
                        final String... stateStoreNames) {
        process(processorSupplier, Named.as(builder.newProcessorName(PROCESSOR_NAME)), stateStoreNames);
    }
@Test
    public void shouldProcessFromSourceThatMatchPattern() {
        final KStream<String, String> pattern2Source = builder.stream(Pattern.compile("topic-\\d"));

        pattern2Source.process(processorSupplier);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            final TestInputTopic<String, String> inputTopic3 =
                driver.createInputTopic("topic-3", new StringSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
            final TestInputTopic<String, String> inputTopic4 =
                driver.createInputTopic("topic-4", new StringSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
            final TestInputTopic<String, String> inputTopic5 =
                driver.createInputTopic("topic-5", new StringSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
            final TestInputTopic<String, String> inputTopic6 =
                driver.createInputTopic("topic-6", new StringSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
            final TestInputTopic<String, String> inputTopic7 =
                driver.createInputTopic("topic-7", new StringSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);

            inputTopic3.pipeInput("A", "aa", 1L);
            inputTopic4.pipeInput("B", "bb", 5L);
            inputTopic5.pipeInput("C", "cc", 10L);
            inputTopic6.pipeInput("D", "dd", 8L);
            inputTopic7.pipeInput("E", "ee", 3L);
        }

        assertEquals(asList(new KeyValueTimestamp<>("A", "aa", 1),
            new KeyValueTimestamp<>("B", "bb", 5),
            new KeyValueTimestamp<>("C", "cc", 10),
            new KeyValueTimestamp<>("D", "dd", 8),
            new KeyValueTimestamp<>("E", "ee", 3)),
            processorSupplier.theCapturedProcessor().processed());
    }