Repository: apache/kafka
File: streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java (Java)

Code before:
            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.deleteTopics(topic1, topic2);
        }
    }

    @Test
    public void test() 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"));
            builder.stream(Pattern.compile("not-a-match"));
            final List<String> assignedTopics = new CopyOnWriteArrayList<>();

            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(builder.build(), streamsConfiguration, new DefaultKafkaClientSupplier() {
                @Override
                public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
                    return new KafkaConsumer<>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer()) {
                        @Override
                        public void subscribe(final Pattern topics, final ConsumerRebalanceListener listener) {

Review comment (bug):
```suggestion
    public void shouldNotCrashIfPatternMatchesTopicHasNoData() throws Exception {
```

Code after:
            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.deleteTopics(topic1, topic2);
        }
    }

    @Test
    public void shouldNotCrashIfPatternMatchesTopicHasNoData() 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"));
            builder.stream(Pattern.compile("not-a-match"));
            final List<String> assignedTopics = new CopyOnWriteArrayList<>();

            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(builder.build(), streamsConfiguration, new DefaultKafkaClientSupplier() {
                @Override
                public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
                    return new KafkaConsumer<>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer()) {
                        @Override
                        public void subscribe(final Pattern topics, final ConsumerRebalanceListener listener) {