spena commented on a change in pull request #10462:
URL: https://github.com/apache/kafka/pull/10462#discussion_r614105673
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImplJoin.java
##########
@@ -118,20 +142,47 @@
final ProcessorGraphNode<K1, V2> otherWindowedStreamsNode = new
ProcessorGraphNode<>(otherWindowStreamProcessorName,
otherWindowStreamProcessorParams);
builder.addGraphNode(otherGraphNode, otherWindowedStreamsNode);
+ Optional<StoreBuilder<WindowStore<KeyAndJoinSide<K1>,
LeftOrRightValue<V1, V2>>>> outerJoinWindowStore = Optional.empty();
+ if (leftOuter || rightOuter) {
+ final String outerJoinSuffix = "-shared-outer-join-store";
+
+ // Get the suffix index of the joinThisGeneratedName to build the
outer join store name.
+ final String outerJoinStoreGeneratedName =
KStreamImpl.OUTERSHARED_NAME
+ + joinThisGeneratedName.substring(
+ rightOuter
+ ? KStreamImpl.OUTERTHIS_NAME.length()
+ : KStreamImpl.JOINTHIS_NAME.length());
Review comment:
I initially generated a name with a new index for the shared store.
However, seems this was going to cause incompatibilities in the topology
because the new indexed increasing. Instead, now I just get the index from one
of the current join stores. Why doesn't make sense? Is there another way to get
an index? Or, do I really need to append an index at the end of the shared
store?
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
