vpapavas commented on code in PR #12555:
URL: https://github.com/apache/kafka/pull/12555#discussion_r960474042
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java:
##########
@@ -270,17 +276,36 @@ private void maybeAddNodeForOptimizationMetadata(final
GraphNode node) {
// use this method for testing only
public void buildAndOptimizeTopology() {
- buildAndOptimizeTopology(false);
+ buildAndOptimizeTopology(null);
}
- public void buildAndOptimizeTopology(final boolean optimizeTopology) {
+ public void buildAndOptimizeTopology(final Properties props) {
+ // Vicky: Do we need to verify props?
+ final List<String> optimizationConfigs;
+ if (props == null) {
+ optimizationConfigs = new ArrayList<>();
+ optimizationConfigs.add(StreamsConfig.NO_OPTIMIZATION);
+ } else {
+ final StreamsConfig config = new StreamsConfig(props);
+ optimizationConfigs =
config.getList(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG);
+ }
mergeDuplicateSourceNodes();
- if (optimizeTopology) {
- LOG.debug("Optimizing the Kafka Streams graph for repartition
nodes");
+ if (optimizationConfigs.contains(StreamsConfig.OPTIMIZE)
Review Comment:
A user can optimize either by providing the config `all` which is the
`StreamsConfig.OPTIMIZE` which will apply all optimization rules or by
specifying a list of optimization rules. With your suggestion, an optimization
will be applied only if `contains(OPTIMIZE)` is true which is not correct. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
