KAFKA-10847: Set StreamsConfig on InternalTopologyDriver before writing topology #10640
Conversation
| @@ -290,6 +290,10 @@ public void buildAndOptimizeTopology(final Properties props) { | |||
| } | |||
|
|
|||
| if (streamGraphNode.allParentsWrittenToTopology() && !streamGraphNode.hasWrittenToTopology()) { | |||
| if (props != null && !props.isEmpty()) { | |||
| internalTopologyBuilder.setStreamsConfig(new StreamsConfig(props)); | |||
There was a problem hiding this comment.
I think we should unify this can call this method only once.
Why not just call it with KafkaStreams constructor?
There was a problem hiding this comment.
Ksql calls the StreamsBuilder first, which calls the StreamStreamJoinNode.writeToTopology() at some point. When KafkaStreams is called later, the shared state store was already added to the join nodes.
| @@ -119,6 +120,11 @@ public Integer buildPriority() { | |||
|
|
|||
| public abstract void writeToTopology(final InternalTopologyBuilder topologyBuilder); | |||
There was a problem hiding this comment.
I don't know if removing this method and replace it with the below method with the properties parameter is better. There are several classes that implement GraphNode and that will need to accept the new parameter. It is an internal interface and impl, though. Do you think I should go with that approach instead?
There was a problem hiding this comment.
Yeah. Might be worth to just remove this one even if we need to update all implementations.
| @@ -98,12 +108,19 @@ public void writeToTopology(final InternalTopologyBuilder topologyBuilder) { | |||
| topologyBuilder.addStateStore(thisWindowStoreBuilder, thisWindowedStreamProcessorName, otherProcessorName); | |||
| topologyBuilder.addStateStore(otherWindowStoreBuilder, otherWindowedStreamProcessorName, thisProcessorName); | |||
|
|
|||
| final StreamsConfig streamsConfig = topologyBuilder.getStreamsConfig(); | |||
| if (streamsConfig == null || StreamsConfig.InternalConfig.getBoolean(streamsConfig.originals(), ENABLE_KSTREAMS_OUTER_JOIN_SPURIOUS_RESULTS_FIX, true)) { | |||
| if (props == null || StreamsConfig.InternalConfig.getBoolean(toMap(props), ENABLE_KSTREAMS_OUTER_JOIN_SPURIOUS_RESULTS_FIX, true)) { | |||
There was a problem hiding this comment.
Instead of using toMap, can't we just to new HashMap<String,String>(props)
There was a problem hiding this comment.
HashMap does not have a constructor that accepts Properties.
There was a problem hiding this comment.
This works (note that Properties implements Map<Object, Object>):
Properties p = new Properties();
Map<String, Object> foo = new HashMap(p);
So you should be able to do getBoolean(new HashMap(props), ...) (Need to omit the generics though...)
There was a problem hiding this comment.
Done. It wasn't working initially, I probably had a syntax error.
There was a problem hiding this comment.
Made a pass on the new commit, I do not have further comments. @mjsax please feel free to merge when it LGTY.
spena
force-pushed
the
set_config_internal_topology
branch
from
ca620bc
to
2d3b936
Compare
While running some tests, I noticed the KSTREAMS-OUTERSHARED store, used in left/outer joins, was still added in the list of state stores when the
StreamsConfig.InternalConfig.ENABLE_KSTREAMS_OUTER_JOIN_SPURIOUS_RESULTS_FIXflag was false. When this flag is false, the shared store should not be added to the join nodes.Testing
Committer Checklist (excluded from commit message)