junaiddshaukat commented on code in PR #39273:
URL: https://github.com/apache/beam/pull/39273#discussion_r3566196886
##########
runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsPipelineTranslator.java:
##########
@@ -126,6 +131,34 @@ public void translate(KafkaStreamsTranslationContext
context, RunnerApi.Pipeline
}
}
+ /**
+ * Records each Flatten input PCollection's branch identity {@code (i of N)}
with the context, so
+ * the producer of that PCollection stamps its watermark with a distinct
source partition. Without
+ * this every branch would report as the single source {@code (0 of 1)} and
the Flatten's {@link
+ * WatermarkManager} could not tell them apart, releasing its watermark
before every branch
+ * drains.
+ */
+ private static void registerFlattenSourceStamps(
+ KafkaStreamsTranslationContext context, RunnerApi.Pipeline pipeline) {
+ for (RunnerApi.PTransform transform :
pipeline.getComponents().getTransformsMap().values()) {
+ if
(!PTransformTranslation.FLATTEN_TRANSFORM_URN.equals(transform.getSpec().getUrn()))
{
+ continue;
+ }
+ // Sort the input PCollection ids so each branch's index is assigned
deterministically. Not
+ // required for correctness — a Flatten and its producers sit in one
task, so the indices only
+ // have to agree within a single topology build — but a stable
assignment is easier to reason
+ // about and reproduce. registerFlattenSourceStamp fails fast on a
duplicate (self-flatten).
+ List<String> inputPCollectionIds = new
ArrayList<>(transform.getInputsMap().values());
+ Collections.sort(inputPCollectionIds);
+ int totalPartitions = inputPCollectionIds.size();
Review Comment:
Good catch, thanks, you're right, per-Flatten (i of N) can't express a
PCollection that feeds two Flattens. Reworked it the way you described: each
Flatten-input PCollection gets one stable global producer id, its producer
stamps that id (always as a single source, 1 of 1), and each Flatten holds its
watermark using its own input count. So input2 stamps one id, l1 waits for
{1,2}, l2 for {2,3}, and the shared input just works. Nice side effect it also
removes the fan-out hazard, since any single-input consumer always sees 1 of 1.
Added a test for your exact case (input2 feeding both flattens produces both
unions). Self-flatten (Flatten.of(pc,pc)) is still rejected, since one producer
can't be two branches.
--
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]
