private void optimizeTopology(final Properties props) {
        final Set<String> optimizationConfigs;
        if (props == null || !props.containsKey(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG)) {
            optimizationConfigs = Collections.emptySet();
        } else {
            optimizationConfigs = StreamsConfig.verifyTopologyOptimizationConfigs(
                (String) props.get(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG));
        }
        if (optimizationConfigs.contains(StreamsConfig.REUSE_KTABLE_SOURCE_TOPICS)) {
            LOG.debug("Optimizing the Kafka Streams graph for ktable source nodes");
            reuseKTableSourceTopics();
        }
        if (optimizationConfigs.contains(StreamsConfig.MERGE_REPARTITION_TOPICS)) {
            LOG.debug("Optimizing the Kafka Streams graph for repartition nodes");
            mergeRepartitionTopics();
        }
        if (optimizationConfigs.contains(StreamsConfig.SINGLE_STORE_SELF_JOIN)) {
            LOG.debug("Optimizing the Kafka Streams graph for self-joins");
            rewriteSingleStoreSelfJoin(root, new IdentityHashMap<>());
        }
        LOG.debug("Optimizing the Kafka Streams graph for null-key records");
        rewriteRepartitionNodes();
    }
A user can provide either the config OPTIMIZE which means all optimizations rules will be
applied or they can provide a list of optimization rules.
streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java