@Test
    public void testRegexRecordsAreProcessedAfterNewTopicCreatedWithMultipleSubtopologies() throws Exception {
        final String topic1 = "TEST-TOPIC-1";
        final String topic2 = "TEST-TOPIC-2";

        try {
            CLUSTER.createTopic(topic1);

            final StreamsBuilder builder = new StreamsBuilder();
            final KStream<String, String> pattern1Stream = builder.stream(Pattern.compile("TEST-TOPIC-\\d"));
            final KStream<String, String> otherStream = builder.stream(Pattern.compile("not-a-match"));

            pattern1Stream
                .selectKey((k, v) -> k)
                .groupByKey()
                .aggregate(() -> "", (k, v, a) -> v)
                .toStream().to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));

            final Topology topology = builder.build();
            assertThat(topology.describe().subtopologies().size(), greaterThan(1));
            streams = new KafkaStreams(topology, streamsConfiguration);

            startApplicationAndWaitUntilRunning(streams);

            CLUSTER.createTopic(topic2);

            final KeyValue<String, String> record1 = new KeyValue<>("1", "1");
            final KeyValue<String, String> record2 = new KeyValue<>("2", "2");
            IntegrationTestUtils.produceKeyValuesSynchronously(
                topic1,
                Collections.singletonList(record1),
                TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class),
                CLUSTER.time
            );
            IntegrationTestUtils.produceKeyValuesSynchronously(
                topic2,
                Collections.singletonList(record2),
                TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class),
                CLUSTER.time
            );
            IntegrationTestUtils.waitUntilFinalKeyValueRecordsReceived(
                TestUtils.consumerConfig(CLUSTER.bootstrapServers(), StringDeserializer.class, StringDeserializer.class),
                outputTopic,
                Arrays.asList(record1, record2)
            );

            streams.close();
        } finally {
            CLUSTER.deleteTopicsAndWait(topic1, topic2);
        }
    }
End-to-end integration test based on using regex and named topics for creating sources, using
an embedded Kafka cluster.
streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java