je-ik commented on code in PR #39273:
URL: https://github.com/apache/beam/pull/39273#discussion_r3570349058
##########
runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ExecutableStageTranslator.java:
##########
@@ -82,10 +82,22 @@ public void translate(
String inputPCollectionId = stagePayload.getInput();
String parentProcessor =
context.getProcessorNameForPCollection(inputPCollectionId);
+ // A stage has at most one output (multi-output rejected above). Stamp its
watermark with that
+ // output's global producer id (0 if no Flatten consumes it), always as a
single source (1 of
+ // 1):
+ // a downstream Flatten tells its branches apart by this id and holds
using its own input count,
+ // while a single-input consumer just sees one source.
+ int watermarkSourceId =
+ transform.getOutputsMap().isEmpty()
+ ? 0
+ : context.getProducerWatermarkId(
+ Iterables.getOnlyElement(transform.getOutputsMap().values()));
+
Topology topology = context.getTopology();
topology.addProcessor(
transformId,
- () -> new ExecutableStageProcessor(stagePayload, context.getJobInfo()),
+ () ->
+ new ExecutableStageProcessor(stagePayload, context.getJobInfo(),
watermarkSourceId, 1),
Review Comment:
We are definitely confusing something here. We cannot pass
"watermarkSourceId" to sourcePartition. These are distinct things.
--
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]
