@Test
    public void testRegexMatchesTopicsAWhenDeleted() throws Exception {
        final Serde<String> stringSerde = Serdes.String();
        final List<String> expectedFirstAssignment = Arrays.asList("TEST-TOPIC-A", "TEST-TOPIC-B");
        final List<String> expectedSecondAssignment = Collections.singletonList("TEST-TOPIC-B");
        final List<String> assignedTopics = new CopyOnWriteArrayList<>();

        try {
            CLUSTER.createTopics("TEST-TOPIC-A", "TEST-TOPIC-B");

            final StreamsBuilder builder = new StreamsBuilder();

            final KStream<String, String> pattern1Stream = builder.stream(Pattern.compile("TEST-TOPIC-[A-Z]"));

            pattern1Stream.to(outputTopic, Produced.with(stringSerde, stringSerde));

            streams = new KafkaStreams(builder.build(), streamsConfiguration, new DefaultKafkaClientSupplier() {
                @Override
                public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
                    return new KafkaConsumer<byte[], byte[]>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer()) {
                        @Override
                        public void subscribe(final Pattern topics, final ConsumerRebalanceListener listener) {
                            super.subscribe(topics, new TheConsumerRebalanceListener(assignedTopics, listener));
                        }
                    };
                }
            });


            streams.start();
            TestUtils.waitForCondition(() -> assignedTopics.equals(expectedFirstAssignment), STREAM_TASKS_NOT_UPDATED);
        } finally {
            CLUSTER.deleteTopicAndWait("TEST-TOPIC-A");
        }

        TestUtils.waitForCondition(() -> assignedTopics.equals(expectedSecondAssignment), STREAM_TASKS_NOT_UPDATED);
    }
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