junaiddshaukat commented on code in PR #39273:
URL: https://github.com/apache/beam/pull/39273#discussion_r3580571776
##########
runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsPipelineTranslator.java:
##########
@@ -126,6 +128,28 @@ 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;
+ }
+ int totalPartitions = transform.getInputsMap().size();
+ int sourcePartition = 0;
+ for (String inputPCollectionId : transform.getInputsMap().values()) {
+ context.registerFlattenSourceStamp(inputPCollectionId,
sourcePartition, totalPartitions);
+ sourcePartition++;
+ }
Review Comment:
Superseded by the rework: the per-Flatten branch numbering (and the pre-pass
this commented on) is gone — producers stamp their own transform id, so there's
no index assignment left to order or dedupe.
##########
runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/FlattenTranslator.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.kafka.streams.translation;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
+import org.apache.kafka.streams.Topology;
+
+/**
+ * Translates Beam's {@code Flatten} primitive ({@code
beam:transform:flatten:v1}): the union of N
+ * input PCollections into one output PCollection.
+ *
+ * <p>Wires a single {@link FlattenProcessor} node to the producer of every
input PCollection (Kafka
+ * Streams lets a processor have many parents), so the parents' data streams
merge into it, and
+ * registers it as the producer of the flattened output so downstream
translators wire to it. The
+ * processor forwards data through and owns its output watermark via a {@link
WatermarkManager},
+ * exactly like GroupByKey — see {@link FlattenProcessor}.
+ *
+ * <p>The {@code (i of N)} branch identity the {@link WatermarkManager} needs
is stamped upstream,
+ * by the parent transform that produces each branch, because Kafka Streams
does not tell a
+ * processor which parent forwarded a record.
+ */
+class FlattenTranslator implements PTransformTranslator {
+
+ @Override
+ public void translate(
+ String transformId, RunnerApi.Pipeline pipeline,
KafkaStreamsTranslationContext context) {
+ RunnerApi.PTransform transform =
pipeline.getComponents().getTransformsOrThrow(transformId);
+ // Flatten produces exactly one output PCollection, fed by all of its
input PCollections.
+ String outputPCollectionId =
Iterables.getOnlyElement(transform.getOutputsMap().values());
+
+ List<String> parentProcessors = new ArrayList<>();
+ for (String inputPCollectionId : transform.getInputsMap().values()) {
+
parentProcessors.add(context.getProcessorNameForPCollection(inputPCollectionId));
+ }
Review Comment:
Same as above, with producer-stamped transform ids there's no numbering to
keep consistent. A duplicate input (self-flatten) is folded into the harness
stage by the fuser and never reaches the runner's Flatten; the translator keeps
a defensive check.
--
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]
