MINOR: Move KTable source topic for changelog to optimization framework

streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java

@@ -61,6 +61,7 @@ public class InternalStreamsBuilder implements InternalNameProvider {
     private final AtomicInteger buildPriorityIndex = new AtomicInteger(0);
     private final LinkedHashMap<StreamsGraphNode, LinkedHashSet<OptimizableRepartitionNode>> keyChangingOperationsToOptimizableRepartitionNodes = new LinkedHashMap<>();
     private final LinkedHashSet<StreamsGraphNode> mergeNodes = new LinkedHashSet<>();
+    private final LinkedHashSet<StreamsGraphNode> tableSourceNodes = new LinkedHashSet<>();
 
     private static final String TOPOLOGY_ROOT = "root";
     private static final Logger LOG = LoggerFactory.getLogger(InternalStreamsBuilder.class);
@@ -254,6 +255,8 @@ private void maybeAddNodeForOptimizationMetadata(final StreamsGraphNode node) {
             }
         } else if (node.isMergeNode()) {
             mergeNodes.add(node);
+        } else if (node instanceof TableSourceNode) {
+            tableSourceNodes.add(node);
         }
     }
 
@@ -292,10 +295,16 @@ private void maybePerformOptimizations(final Properties props) {
 
         if (props != null && StreamsConfig.OPTIMIZE.equals(props.getProperty(StreamsConfig.TOPOLOGY_OPTIMIZATION))) {
             LOG.debug("Optimizing the Kafka Streams graph for repartition nodes");
+            optimizeKTableSourceTopics();
             maybeOptimizeRepartitionOperations();
         }
     }
 
+    private void optimizeKTableSourceTopics() {
+        LOG.debug("Marking KTable source nodes to optimize using source topic for changelogs ");
+        tableSourceNodes.forEach(node -> ((TableSourceNode) node).reuseSourceTopicForChangeLog(true));
+    }
+
     @SuppressWarnings("unchecked")
     private void maybeOptimizeRepartitionOperations() {
         maybeUpdateKeyChangingRepartitionNodeMap();

streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableSourceNode.java

@@ -38,6 +38,7 @@ public class TableSourceNode<K, V> extends StreamSourceNode<K, V> {
     private final ProcessorParameters<K, V> processorParameters;
     private final String sourceName;
     private final boolean isGlobalKTable;
+    private boolean shouldReuseSourceTopicForChangelog = false;
 
     private TableSourceNode(final String nodeName,
                             final String sourceName,
@@ -57,6 +58,11 @@ private TableSourceNode(final String nodeName,
         this.materializedInternal = materializedInternal;
     }
 
+
+    public void reuseSourceTopicForChangeLog(final boolean shouldReuseSourceTopicForChangelog) {
+        this.shouldReuseSourceTopicForChangelog = shouldReuseSourceTopicForChangelog;
+    }
+
     @Override
     public String toString() {
         return "TableSourceNode{" +
@@ -104,7 +110,11 @@ public void writeToTopology(final InternalTopologyBuilder topologyBuilder) {
             final KTableSource<K, V> ktableSource = (KTableSource<K, V>) processorParameters.processorSupplier();
             if (ktableSource.queryableName() != null) {
                 topologyBuilder.addStateStore(storeBuilder, nodeName());
-                topologyBuilder.markSourceStoreAndTopic(storeBuilder, topicName);
+
+                if (shouldReuseSourceTopicForChangelog) {
+                    storeBuilder.withLoggingDisabled();
+                    topologyBuilder.connectSourceStoreAndTopic(storeBuilder.name(), topicName);
+                }
             }
         }
 

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

@@ -123,9 +123,6 @@ public class InternalTopologyBuilder {
 
     private Map<Integer, Set<String>> nodeGroups = null;
 
-    // TODO: this is only temporary for 2.0 and should be removed
-    private final Map<StoreBuilder, String> storeToSourceChangelogTopic = new HashMap<>();
-
     public static class StateStoreFactory {
         private final StoreBuilder builder;
         private final Set<String> users = new HashSet<>();
@@ -359,10 +356,6 @@ public synchronized final InternalTopologyBuilder rewriteTopology(final StreamsC
             globalStateStores.put(storeBuilder.name(), storeBuilder.build());
         }
 
-        // adjust the topology if optimization is turned on.
-        // TODO: to be removed post 2.0
-        adjust(config);
-
         return this;
     }
 
@@ -606,22 +599,14 @@ public final void connectProcessorAndStateStores(final String processorName,
         nodeGroups = null;
     }
 
-    private void connectSourceStoreAndTopic(final String sourceStoreName,
+    public void connectSourceStoreAndTopic(final String sourceStoreName,
                                             final String topic) {
         if (storeToChangelogTopic.containsKey(sourceStoreName)) {
             throw new TopologyException("Source store " + sourceStoreName + " is already added.");
         }
         storeToChangelogTopic.put(sourceStoreName, topic);
     }
 
-    public final void markSourceStoreAndTopic(final StoreBuilder storeBuilder,
-                                              final String topic) {
-        if (storeToSourceChangelogTopic.containsKey(storeBuilder)) {
-            throw new TopologyException("Source store " + storeBuilder.name() + " is already used.");
-        }
-        storeToSourceChangelogTopic.put(storeBuilder, topic);
-    }
-
     public final void addInternalTopic(final String topicName) {
         Objects.requireNonNull(topicName, "topicName can't be null");
         internalTopicNames.add(topicName);
@@ -1071,25 +1056,6 @@ public synchronized Map<Integer, TopicsInfo> topicGroups() {
         return Collections.unmodifiableMap(topicGroups);
     }
 
-    // Adjust the generated topology based on the configs.
-    // Not exposed as public API and should be removed post 2.0
-    private void adjust(final StreamsConfig config) {
-        final boolean enableOptimization20 =
-            config.getString(StreamsConfig.TOPOLOGY_OPTIMIZATION).equals(StreamsConfig.OPTIMIZE);
-
-        if (enableOptimization20) {
-            for (final Map.Entry<StoreBuilder, String> entry : storeToSourceChangelogTopic.entrySet()) {
-                final StoreBuilder storeBuilder = entry.getKey();
-                final String topicName = entry.getValue();
-
-                // update store map to disable logging for this store
-                storeBuilder.withLoggingDisabled();
-                addStateStore(storeBuilder, true);
-                connectSourceStoreAndTopic(storeBuilder.name(), topicName);
-            }
-        }
-    }
-
     private void setRegexMatchedTopicsToSourceNodes() {
         if (subscriptionUpdates.hasUpdates()) {
             for (final Map.Entry<String, Pattern> stringPatternEntry : nodeToSourcePatterns.entrySet()) {

streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java

@@ -349,9 +349,9 @@ public void shouldNotMaterializeStoresIfNotRequired() {
     public void shouldReuseSourceTopicAsChangelogsWithOptimization20() {
         final String topic = "topic";
         builder.table(topic, Materialized.<Long, String, KeyValueStore<Bytes, byte[]>>as("store"));
-        final Topology topology = builder.build();
         final Properties props = StreamsTestUtils.getStreamsConfig();
         props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION, StreamsConfig.OPTIMIZE);
+        final Topology topology = builder.build(props);
 
         final InternalTopologyBuilder internalTopologyBuilder = TopologyWrapper.getInternalTopologyBuilder(topology);
         internalTopologyBuilder.rewriteTopology(new StreamsConfig(props));

streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java

@@ -146,7 +146,7 @@ public void shouldRestoreStateFromSourceTopic() throws Exception {
                     }
                 });
 
-        kafkaStreams = new KafkaStreams(builder.build(), props);
+        kafkaStreams = new KafkaStreams(builder.build(props), props);
         kafkaStreams.setStateListener((newState, oldState) -> {
             if (newState == KafkaStreams.State.RUNNING && oldState == KafkaStreams.State.REBALANCING) {
                 startupLatch.countDown();