MINOR: further InternalTopologyBuilder cleanup

streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java

@@ -118,9 +118,9 @@ public class InternalTopologyBuilder {
 
     private String applicationId = null;
 
-    private Pattern topicPattern = null;
+    private Pattern sourceTopicPattern = null;
 
-    private List<String> topicCollection = null;
+    private List<String> sourceTopicCollection = null;
 
     private Map<Integer, Set<String>> nodeGroups = null;
 
@@ -790,23 +790,18 @@ private int putNodeGroupName(final String nodeName,
     }
 
     public synchronized ProcessorTopology build() {
-        return build((Integer) null);
+        final Set<String> nodeGroup = new HashSet<>();
+        for (final Set<String> value : nodeGroups().values()) {
+            nodeGroup.addAll(value);
+        }
+        nodeGroup.removeAll(globalNodeGroups());
+
+        initializeSubscription();
+        return build(nodeGroup);
     }
 
-    public synchronized ProcessorTopology build(final Integer topicGroupId) {
-        final Set<String> nodeGroup;
-        if (topicGroupId != null) {
-            nodeGroup = nodeGroups().get(topicGroupId);
-        } else {
-            // when topicGroupId is null, we build the full topology minus the global groups
-            final Set<String> globalNodeGroups = globalNodeGroups();
-            final Collection<Set<String>> values = nodeGroups().values();
-            nodeGroup = new HashSet<>();
-            for (final Set<String> value : values) {
-                nodeGroup.addAll(value);
-            }
-            nodeGroup.removeAll(globalNodeGroups);
-        }
+    public synchronized ProcessorTopology build(final int topicGroupId) {
+        final Set<String> nodeGroup = nodeGroups().get(topicGroupId);
         return build(nodeGroup);
     }
 
@@ -1210,32 +1205,29 @@ private String decorateTopic(final String topic) {
         return applicationId + "-" + topic;
     }
 
+    void initializeSubscription() {
+        if (usesPatternSubscription()) {
+            log.debug("Found pattern subscribed source topics, initializing consumer's subscription pattern.");
+            final List<String> allSourceTopics = maybeDecorateInternalSourceTopics(sourceTopicNames);
+            Collections.sort(allSourceTopics);
+            sourceTopicPattern = buildPattern(allSourceTopics, nodeToSourcePatterns.values());
+        } else {
+            log.debug("No source topics using pattern subscription found, initializing consumer's subscription collection.");
+            sourceTopicCollection = maybeDecorateInternalSourceTopics(sourceTopicNames);
+            Collections.sort(sourceTopicCollection);
+        }
+    }
+
     boolean usesPatternSubscription() {
         return !nodeToSourcePatterns.isEmpty();
     }
 
     synchronized Collection<String> sourceTopicCollection() {
-        log.debug("No source topics using pattern subscription found, using regular subscription for the main consumer.");
-
-        if (topicCollection == null) {
-            topicCollection = maybeDecorateInternalSourceTopics(sourceTopicNames);
-            Collections.sort(topicCollection);
-        }
-
-        return topicCollection;
+        return sourceTopicCollection;
     }
 
     synchronized Pattern sourceTopicPattern() {
-        if (topicPattern == null) {
-            final List<String> allSourceTopics = maybeDecorateInternalSourceTopics(sourceTopicNames);
-            Collections.sort(allSourceTopics);
-            topicPattern = buildPattern(allSourceTopics, nodeToSourcePatterns.values());
-        }
-
-        log.debug("Found pattern subscribed source topics, falling back to pattern " +
-            "subscription for the main consumer: {}", topicPattern);
-
-        return topicPattern;
+        return sourceTopicPattern;
     }
 
     private boolean isGlobalSource(final String nodeName) {
@@ -1872,37 +1864,35 @@ private boolean hasSubscriptionUpdates() {
     }
 
     synchronized void addSubscribedTopicsFromAssignment(final List<TopicPartition> partitions, final String logPrefix) {
-        if (sourceTopicPattern() != null) {
+        if (usesPatternSubscription()) {
             final Set<String> assignedTopics = new HashSet<>();
             for (final TopicPartition topicPartition : partitions) {
                 assignedTopics.add(topicPartition.topic());
             }
-
-            final Collection<String> existingTopics = subscriptionUpdates();
-            if (!existingTopics.containsAll(assignedTopics)) {
-                assignedTopics.addAll(existingTopics);
-                updateSubscribedTopics(assignedTopics, logPrefix);
-            }
+            updateSubscribedTopics(assignedTopics, logPrefix);
         }
     }
 
     synchronized void addSubscribedTopicsFromMetadata(final Set<String> topics, final String logPrefix) {
-        if (sourceTopicPattern() != null) {
-            final Collection<String> existingTopics = subscriptionUpdates();
-            if (!existingTopics.equals(topics)) {
-                topics.addAll(existingTopics);
-                updateSubscribedTopics(topics, logPrefix);
-            }
+        if (usesPatternSubscription()) {
+            updateSubscribedTopics(topics, logPrefix);
         }
     }
 
     private void updateSubscribedTopics(final Set<String> topics, final String logPrefix) {
-        log.debug("{}found {} topics possibly matching subscription", logPrefix, topics.size());
-        subscriptionUpdates.clear();
-        subscriptionUpdates.addAll(topics);
+        final Collection<String> existingTopics = subscriptionUpdates();
+
+        if  (!existingTopics.equals(topics)) {
+            topics.addAll(existingTopics);
 
-        setRegexMatchedTopicsToSourceNodes();
-        setRegexMatchedTopicToStateStore();
+            subscriptionUpdates.clear();
+            subscriptionUpdates.addAll(topics);
+
+            log.debug("{}found {} topics possibly matching subscription", logPrefix, topics.size());
+
+            setRegexMatchedTopicsToSourceNodes();
+            setRegexMatchedTopicToStateStore();
+        }
     }
 
     // following functions are for test only

streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java

@@ -132,7 +132,7 @@ public void shouldNotMaterializeSourceKTableIfNotRequired() {
         builder.buildAndOptimizeTopology();
         final ProcessorTopology topology = builder.internalTopologyBuilder
             .rewriteTopology(new StreamsConfig(StreamsTestUtils.getStreamsConfig(APP_ID)))
-            .build(null);
+            .build();
 
         assertEquals(0, topology.stateStores().size());
         assertEquals(0, topology.storeToChangelogTopic().size());
@@ -352,7 +352,7 @@ public void shouldHaveNullTimestampExtractorWhenNoneSupplied() {
         builder.stream(Collections.singleton("topic"), consumed);
         builder.buildAndOptimizeTopology();
         builder.internalTopologyBuilder.rewriteTopology(new StreamsConfig(StreamsTestUtils.getStreamsConfig(APP_ID)));
-        final ProcessorTopology processorTopology = builder.internalTopologyBuilder.build(null);
+        final ProcessorTopology processorTopology = builder.internalTopologyBuilder.build();
         assertNull(processorTopology.source("topic").getTimestampExtractor());
     }
 
@@ -363,7 +363,7 @@ public void shouldUseProvidedTimestampExtractor() {
         builder.buildAndOptimizeTopology();
         final ProcessorTopology processorTopology = builder.internalTopologyBuilder
             .rewriteTopology(new StreamsConfig(StreamsTestUtils.getStreamsConfig(APP_ID)))
-            .build(null);
+            .build();
         assertThat(processorTopology.source("topic").getTimestampExtractor(), instanceOf(MockTimestampExtractor.class));
     }
 
@@ -373,7 +373,7 @@ public void ktableShouldHaveNullTimestampExtractorWhenNoneSupplied() {
         builder.buildAndOptimizeTopology();
         final ProcessorTopology processorTopology = builder.internalTopologyBuilder
             .rewriteTopology(new StreamsConfig(StreamsTestUtils.getStreamsConfig(APP_ID)))
-            .build(null);
+            .build();
         assertNull(processorTopology.source("topic").getTimestampExtractor());
     }
 
@@ -384,7 +384,7 @@ public void ktableShouldUseProvidedTimestampExtractor() {
         builder.buildAndOptimizeTopology();
         final ProcessorTopology processorTopology = builder.internalTopologyBuilder
             .rewriteTopology(new StreamsConfig(StreamsTestUtils.getStreamsConfig(APP_ID)))
-            .build(null);
+            .build();
         assertThat(processorTopology.source("topic").getTimestampExtractor(), instanceOf(MockTimestampExtractor.class));
     }
 

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java

@@ -1313,7 +1313,7 @@ public void testNumProcesses() {
                 1 + // to
                 2 + // through
                 1, // process
-            TopologyWrapper.getInternalTopologyBuilder(builder.build()).setApplicationId("X").build(null).processors().size());
+            TopologyWrapper.getInternalTopologyBuilder(builder.build()).setApplicationId("X").build().processors().size());
     }
 
     @SuppressWarnings("rawtypes")
@@ -1417,7 +1417,7 @@ public void shouldUseRecordMetadataTimestampExtractorWithThrough() {
         stream1.to("topic-5");
         stream2.through("topic-6");
 
-        final ProcessorTopology processorTopology = TopologyWrapper.getInternalTopologyBuilder(builder.build()).setApplicationId("X").build(null);
+        final ProcessorTopology processorTopology = TopologyWrapper.getInternalTopologyBuilder(builder.build()).setApplicationId("X").build();
         assertThat(processorTopology.source("topic-6").getTimestampExtractor(), instanceOf(FailOnInvalidTimestamp.class));
         assertNull(processorTopology.source("topic-4").getTimestampExtractor());
         assertNull(processorTopology.source("topic-3").getTimestampExtractor());

streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java

@@ -72,6 +72,7 @@ public void shouldAddSourceWithOffsetReset() {
 
         builder.addSource(Topology.AutoOffsetReset.EARLIEST, "source", null, null, null, earliestTopic);
         builder.addSource(Topology.AutoOffsetReset.LATEST, "source2", null, null, null, latestTopic);
+        builder.initializeSubscription();
 
         assertTrue(builder.earliestResetTopicsPattern().matcher(earliestTopic).matches());
         assertTrue(builder.latestResetTopicsPattern().matcher(latestTopic).matches());
@@ -84,18 +85,18 @@ public void shouldAddSourcePatternWithOffsetReset() {
 
         builder.addSource(Topology.AutoOffsetReset.EARLIEST, "source", null, null, null, Pattern.compile(earliestTopicPattern));
         builder.addSource(Topology.AutoOffsetReset.LATEST, "source2", null, null, null,  Pattern.compile(latestTopicPattern));
+        builder.initializeSubscription();
 
         assertTrue(builder.earliestResetTopicsPattern().matcher("earliestTestTopic").matches());
         assertTrue(builder.latestResetTopicsPattern().matcher("latestTestTopic").matches());
     }
 
     @Test
     public void shouldAddSourceWithoutOffsetReset() {
-        final Pattern expectedPattern = Pattern.compile("test-topic");
-
         builder.addSource(null, "source", null, stringSerde.deserializer(), stringSerde.deserializer(), "test-topic");
+        builder.initializeSubscription();
 
-        assertEquals(expectedPattern.pattern(), builder.sourceTopicPattern().pattern());
+        assertEquals(Arrays.asList("test-topic"), builder.sourceTopicCollection());
         assertEquals(builder.earliestResetTopicsPattern().pattern(), "");
         assertEquals(builder.latestResetTopicsPattern().pattern(), "");
     }
@@ -105,6 +106,7 @@ public void shouldAddPatternSourceWithoutOffsetReset() {
         final Pattern expectedPattern = Pattern.compile("test-.*");
 
         builder.addSource(null, "source", null, stringSerde.deserializer(), stringSerde.deserializer(), Pattern.compile("test-.*"));
+        builder.initializeSubscription();
 
         assertEquals(expectedPattern.pattern(), builder.sourceTopicPattern().pattern());
         assertEquals(builder.earliestResetTopicsPattern().pattern(), "");
@@ -237,29 +239,35 @@ public void testOnlyTopicNameSourceTopics() {
         builder.addSource(null, "source-2", null, null, null, "topic-2");
         builder.addSource(null, "source-3", null, null, null, "topic-3");
         builder.addInternalTopic("topic-3");
+        builder.initializeSubscription();
 
         assertFalse(builder.usesPatternSubscription());
         assertEquals(Arrays.asList("X-topic-3", "topic-1", "topic-2"), builder.sourceTopicCollection());
     }
 
     @Test
-    public void testPatternSourceTopics() {
+    public void testPatternAndNameSourceTopics() {
+        final Pattern sourcePattern = Pattern.compile("topic-4|topic-5");
+
         builder.setApplicationId("X");
         builder.addSource(null, "source-1", null, null, null, "topic-1");
         builder.addSource(null, "source-2", null, null, null, "topic-2");
         builder.addSource(null, "source-3", null, null, null, "topic-3");
+        builder.addSource(null, "source-4", null, null, null, sourcePattern);
+
         builder.addInternalTopic("topic-3");
+        builder.initializeSubscription();
 
-        final Pattern expectedPattern = Pattern.compile("X-topic-3|topic-1|topic-2");
+        final Pattern expectedPattern = Pattern.compile("X-topic-3|topic-1|topic-2|topic-4|topic-5");
 
         assertEquals(expectedPattern.pattern(), builder.sourceTopicPattern().pattern());
     }
 
     @Test
-    public void testSourceTopicsWithGlobalTopics() {
+    public void testPatternSourceTopicsWithGlobalTopics() {
         builder.setApplicationId("X");
-        builder.addSource(null, "source-1", null, null, null, "topic-1");
-        builder.addSource(null, "source-2", null, null, null, "topic-2");
+        builder.addSource(null, "source-1", null, null, null, Pattern.compile("topic-1"));
+        builder.addSource(null, "source-2", null, null, null, Pattern.compile("topic-2"));
         builder.addGlobalStore(
                 new MockKeyValueStoreBuilder("global-store", false).withLoggingDisabled(),
                 "globalSource",
@@ -269,16 +277,37 @@ public void testSourceTopicsWithGlobalTopics() {
                 "globalTopic",
                 "global-processor",
                 new MockProcessorSupplier());
+        builder.initializeSubscription();
 
         final Pattern expectedPattern = Pattern.compile("topic-1|topic-2");
 
         assertThat(builder.sourceTopicPattern().pattern(), equalTo(expectedPattern.pattern()));
     }
 
+    @Test
+    public void testNameSourceTopicsWithGlobalTopics() {
+        builder.setApplicationId("X");
+        builder.addSource(null, "source-1", null, null, null, "topic-1");
+        builder.addSource(null, "source-2", null, null, null, "topic-2");
+        builder.addGlobalStore(
+            new MockKeyValueStoreBuilder("global-store", false).withLoggingDisabled(),
+            "globalSource",
+            null,
+            null,
+            null,
+            "globalTopic",
+            "global-processor",
+            new MockProcessorSupplier());
+        builder.initializeSubscription();
+
+        assertThat(builder.sourceTopicCollection(), equalTo(asList("topic-1", "topic-2")));
+    }
+
     @Test
     public void testPatternSourceTopic() {
         final Pattern expectedPattern = Pattern.compile("topic-\\d");
         builder.addSource(null, "source-1", null, null, null, expectedPattern);
+        builder.initializeSubscription();
         assertEquals(expectedPattern.pattern(), builder.sourceTopicPattern().pattern());
     }
 
@@ -287,6 +316,7 @@ public void testAddMoreThanOnePatternSourceNode() {
         final Pattern expectedPattern = Pattern.compile("topics[A-Z]|.*-\\d");
         builder.addSource(null, "source-1", null, null, null, Pattern.compile("topics[A-Z]"));
         builder.addSource(null, "source-2", null, null, null, Pattern.compile(".*-\\d"));
+        builder.initializeSubscription();
         assertEquals(expectedPattern.pattern(), builder.sourceTopicPattern().pattern());
     }
 
@@ -295,6 +325,7 @@ public void testSubscribeTopicNameAndPattern() {
         final Pattern expectedPattern = Pattern.compile("topic-bar|topic-foo|.*-\\d");
         builder.addSource(null, "source-1", null, null, null, "topic-foo", "topic-bar");
         builder.addSource(null, "source-2", null, null, null, Pattern.compile(".*-\\d"));
+        builder.initializeSubscription();
         assertEquals(expectedPattern.pattern(), builder.sourceTopicPattern().pattern());
     }
 
@@ -356,11 +387,11 @@ public void testAddStateStore() {
         builder.addSource(null, "source-1", null, null, null, "topic-1");
         builder.addProcessor("processor-1", new MockProcessorSupplier(), "source-1");
 
-        assertEquals(0, builder.build(null).stateStores().size());
+        assertEquals(0, builder.build().stateStores().size());
 
         builder.connectProcessorAndStateStores("processor-1", storeBuilder.name());
 
-        final List<StateStore> suppliers = builder.build(null).stateStores();
+        final List<StateStore> suppliers = builder.build().stateStores();
         assertEquals(1, suppliers.size());
         assertEquals(storeBuilder.name(), suppliers.get(0).name());
     }
@@ -703,15 +734,15 @@ public void shouldSetCorrectSourceNodesWithRegexUpdatedTopics() {
     @Test
     public void shouldAddTimestampExtractorPerSource() {
         builder.addSource(null, "source", new MockTimestampExtractor(), null, null, "topic");
-        final ProcessorTopology processorTopology = builder.rewriteTopology(new StreamsConfig(StreamsTestUtils.getStreamsConfig())).build(null);
+        final ProcessorTopology processorTopology = builder.rewriteTopology(new StreamsConfig(StreamsTestUtils.getStreamsConfig())).build();
         assertThat(processorTopology.source("topic").getTimestampExtractor(), instanceOf(MockTimestampExtractor.class));
     }
 
     @Test
     public void shouldAddTimestampExtractorWithPatternPerSource() {
         final Pattern pattern = Pattern.compile("t.*");
         builder.addSource(null, "source", new MockTimestampExtractor(), null, null, pattern);
-        final ProcessorTopology processorTopology = builder.rewriteTopology(new StreamsConfig(StreamsTestUtils.getStreamsConfig())).build(null);
+        final ProcessorTopology processorTopology = builder.rewriteTopology(new StreamsConfig(StreamsTestUtils.getStreamsConfig())).build();
         assertThat(processorTopology.source(pattern.pattern()).getTimestampExtractor(), instanceOf(MockTimestampExtractor.class));
     }
 

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java

@@ -188,6 +188,8 @@ private StreamThread createStreamThread(@SuppressWarnings("SameParameterValue")
             config.getString(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG)
         );
 
+        internalTopologyBuilder.build();
+
         return StreamThread.create(
             internalTopologyBuilder,
             config,

streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java

@@ -292,7 +292,7 @@ private TopologyTestDriver(final InternalTopologyBuilder builder,
         internalTopologyBuilder = builder;
         internalTopologyBuilder.rewriteTopology(streamsConfig);
 
-        processorTopology = internalTopologyBuilder.build(null);
+        processorTopology = internalTopologyBuilder.build();
         globalTopology = internalTopologyBuilder.buildGlobalStateTopology();
         final boolean createStateDirectory = processorTopology.hasPersistentLocalStore() ||
             (globalTopology != null && globalTopology.hasPersistentGlobalStore());