Nikita-Shupletsov commented on code in PR #19995:
URL: https://github.com/apache/kafka/pull/19995#discussion_r3753697182
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java:
##########
@@ -300,22 +300,30 @@ private void
maybeAddNodeForVersionedSemanticsMetadata(final GraphNode node) {
}
}
- // use this method for testing only
public void buildAndOptimizeTopology() {
- buildAndOptimizeTopology(null);
+ buildAndOptimizeTopologyInternal(null);
}
+ @Deprecated
public void buildAndOptimizeTopology(final Properties props) {
+ buildAndOptimizeTopologyInternal(props);
+ }
+
+ private void buildAndOptimizeTopologyInternal(final Properties props) {
mergeDuplicateSourceNodes();
- optimizeTopology(props);
+ if (props != null) {
+ optimizeTopology(props);
+ } else {
+ optimizeTopology();
+ }
enableVersionedSemantics();
- final PriorityQueue<GraphNode> graphNodePriorityQueue = new
PriorityQueue<>(5, Comparator.comparing(GraphNode::buildPriority));
-
+ final PriorityQueue<GraphNode> graphNodePriorityQueue =
Review Comment:
nit: the PR is already pretty big, I would suggest keeping reformatting as a
separate change if needed.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java:
##########
@@ -342,12 +349,36 @@ public void buildAndOptimizeTopology(final Properties
props) {
*/
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));
+ (String) props.get(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG)
+ );
}
+
+ applyOptimizations(optimizationConfigs);
+ }
+
+ /**
+ * This method is called after the topology has been built, and it applies
the optimizations
+ * based on the configuration set in the topology configs.
+ */
+ private void optimizeTopology() {
+ final StreamsConfig topicSpecificConfigs =
internalTopologyBuilder.topologySpecificConfigs();
Review Comment:
nit: topologySpecificConfigs?
##########
streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java:
##########
@@ -591,6 +619,7 @@ public synchronized Topology build() {
* @param props the {@link Properties} used for building possibly
optimized topology
* @return the {@link Topology} that represents the specified processing
logic
*/
+ @Deprecated
Review Comment:
it looks like right now it's possible to pass properties into the
constructor and then pass them here as well.
Which will silently override what we passed into the constructor.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java:
##########
@@ -342,12 +349,36 @@ public void buildAndOptimizeTopology(final Properties
props) {
*/
private void optimizeTopology(final Properties props) {
final Set<String> optimizationConfigs;
+
if (props == null ||
!props.containsKey(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG)) {
Review Comment:
right now it looks like this branch is redundant, as the check moved up to
buildAndOptimizeTopologyInternal
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/StreamJoinedInternal.java:
##########
@@ -40,9 +42,14 @@ public StreamJoinedInternal(
super(streamJoined);
passedInDslStoreSuppliers = dslStoreSuppliers;
if (dslStoreSuppliers == null) {
- final TopologyConfig topologyConfig =
builder.internalTopologyBuilder().topologyConfigs();
- if (topologyConfig != null) {
- dslStoreSuppliers =
topologyConfig.resolveDslStoreSuppliers().orElse(null);
+ final StreamsConfig streamsConfig =
builder.internalTopologyBuilder.topologySpecificConfigs();
Review Comment:
maybe this selection logic should live in internalTopologyBuilder for the
time being so we don't need to duplicate it?
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java:
##########
@@ -300,22 +300,30 @@ private void
maybeAddNodeForVersionedSemanticsMetadata(final GraphNode node) {
}
}
- // use this method for testing only
public void buildAndOptimizeTopology() {
- buildAndOptimizeTopology(null);
+ buildAndOptimizeTopologyInternal(null);
}
+ @Deprecated
public void buildAndOptimizeTopology(final Properties props) {
+ buildAndOptimizeTopologyInternal(props);
+ }
+
+ private void buildAndOptimizeTopologyInternal(final Properties props) {
mergeDuplicateSourceNodes();
- optimizeTopology(props);
+ if (props != null) {
+ optimizeTopology(props);
+ } else {
+ optimizeTopology();
+ }
enableVersionedSemantics();
- final PriorityQueue<GraphNode> graphNodePriorityQueue = new
PriorityQueue<>(5, Comparator.comparing(GraphNode::buildPriority));
-
+ final PriorityQueue<GraphNode> graphNodePriorityQueue =
+ new PriorityQueue<>(5,
Comparator.comparing(GraphNode::buildPriority));
graphNodePriorityQueue.offer(root);
while (!graphNodePriorityQueue.isEmpty()) {
- final GraphNode streamGraphNode = graphNodePriorityQueue.remove();
+ final GraphNode streamGraphNode = graphNodePriorityQueue.poll();
Review Comment:
I am not 100% sure if this change is really needed.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -3521,10 +3519,9 @@ private StreamTask
createStatelessTaskWithForwardingTopology(final SourceNode<In
singleton(partition1),
topology,
consumer,
- new TopologyConfig(null, config, new
Properties()).getTaskConfig(),
+ config.getTaskConfig(),
new StreamsMetricsImpl(metrics, "test", time),
- stateDirectory,
- cache,
+ stateDirectory, cache,
Review Comment:
nit: fotmatting
--
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]
