junaiddshaukat commented on code in PR #39273:
URL: https://github.com/apache/beam/pull/39273#discussion_r3580566267
##########
runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsTranslationContext.java:
##########
@@ -120,4 +129,44 @@ public String getReadBootstrapTopic(String transformId) {
+ "_"
+ sanitizedTransformId;
}
+
+ /**
+ * Records that the given PCollection is the {@code sourcePartition}-th of
{@code totalPartitions}
+ * input branches of a Flatten, so its producer stamps its watermark with
that identity instead of
+ * the default single source. Lets the Flatten's {@link WatermarkManager}
tell its branches apart
+ * (Kafka Streams does not tell a processor which parent forwarded a record).
+ */
+ public void registerFlattenSourceStamp(
+ String pCollectionId, int sourcePartition, int totalPartitions) {
+ pCollectionIdToSourceStamp.put(
+ pCollectionId, new SourceStamp(sourcePartition, totalPartitions));
Review Comment:
The rework made this case work instead of throwing: each Flatten waits for
its own set of upstream transform ids, so a shared input needs no special
handling. There's a test (pCollectionFeedingTwoFlattensIsSupported) covering
exactly this.
--
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]
