@Test
    public void testMultipleConsumersCanReadFromPartitionedTopic() throws Exception {
        KafkaStreams partitionedStreamsLeader = null;
        KafkaStreams partitionedStreamsFollower = null;
        try {
            final Serde<String> stringSerde = Serdes.String();
            final StreamsBuilder builderLeader = new StreamsBuilder();
            final StreamsBuilder builderFollower = new StreamsBuilder();
            final List<String> expectedAssignment = Arrays.asList(PARTITIONED_TOPIC_1, PARTITIONED_TOPIC_2);

            final KStream<String, String> partitionedStreamLeader = builderLeader.stream(Pattern.compile("partitioned-\\d"));
            final KStream<String, String> partitionedStreamFollower = builderFollower.stream(Pattern.compile("partitioned-\\d"));


            partitionedStreamLeader.to(outputTopic, Produced.with(stringSerde, stringSerde));
            partitionedStreamFollower.to(outputTopic, Produced.with(stringSerde, stringSerde));

            final List<String> leaderAssignment = new CopyOnWriteArrayList<>();
            final List<String> followerAssignment = new CopyOnWriteArrayList<>();

            partitionedStreamsLeader = new KafkaStreams(builderLeader.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(leaderAssignment, listener));
                        }
                    };

                }
            });
            partitionedStreamsFollower = new KafkaStreams(builderFollower.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(followerAssignment, listener));
                        }
                    };

                }
            });

            partitionedStreamsLeader.start();
            partitionedStreamsFollower.start();
            TestUtils.waitForCondition(() -> followerAssignment.equals(expectedAssignment) && leaderAssignment.equals(expectedAssignment), "topic assignment not completed");
        } finally {
            if (partitionedStreamsLeader != null) {
                partitionedStreamsLeader.close();
            }
            if (partitionedStreamsFollower != null) {
                partitionedStreamsFollower.close();
            }
        }
    }
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