junaiddshaukat commented on code in PR #39273:
URL: https://github.com/apache/beam/pull/39273#discussion_r3570958642
##########
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:
You're right, I was mixing two different things. Reworked it: the watermark
payload now carries all three fields, the producing transform's id (new
transform_id proto field), the source partition, and the partition count. Every
producer stamps its own transform id and its own partition (0 of 1 while
single-instance), without knowing who consumes it. On the consuming side
there's a new WatermarkAggregator used by ExecutableStage, GBK and Flatten
(CombinePerKey later): it gets the expected upstream transform ids from the
pipeline graph at translation time, tracks each upstream's partitions with its
own WatermarkManager, and advances to the min across upstreams once all have
reported. A report from an unexpected transform now throws, so a wiring mistake
like this one fails loudly instead of silently working.
On why tests didn't catch it: everything is single-instance (0 of 1), so the
two versions behaved identically in every topology we can currently build, it
would only have become observable with multi-instance transforms. The stage
watermark test now also asserts the forwarded report carries the stage's own
transform id.
One side finding: a self-flatten (Flatten.of(pc, pc)) is handled by the
fuser, it folds the Flatten into the consuming harness stage, which does the
duplication. The previous revision falsely rejected it; there's now a test
asserting each element appears twice.
--
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]
