Refactor primitive read translators to reduce code duplication

- Add translateUnboundedSource() helper method for common unbounded source translation logic
- Add translateBoundedSource() helper method for common bounded source translation logic
- Simplify UnboundedReadSourceTranslator to use helper method
- Simplify PrimitiveUnboundedReadTranslator to use helper method
- Simplify BoundedReadSourceTranslator to use helper method
- Simplify PrimitiveBoundedReadTranslator to use helper method
- Add missing try-catch, .returns(), and batch mode slot sharing to PrimitiveBoundedReadTranslator

This refactoring reduces ~100 lines of duplicated code and ensures
consistent behavior across all read translators.

Author: dnamaz

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java

@@ -97,6 +97,7 @@
 import org.apache.beam.sdk.values.WindowingStrategy;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.functions.FlatMapFunction;
@@ -195,29 +196,127 @@ public static String getCurrentTransformName(FlinkStreamingTranslationContext co
   //  Transformation Implementations
   // --------------------------------------------------------------------------------------------
 
+  /** Common translation logic for unbounded sources. */
+  @SuppressWarnings("unchecked")
+  private static <T> void translateUnboundedSource(
+      UnboundedSource<T, ?> rawSource,
+      String transformName,
+      FlinkStreamingTranslationContext context) {
+
+    PCollection<T> output =
+        (PCollection<T>)
+            Iterables.getOnlyElement(context.getCurrentTransform().getOutputs().values());
+
+    DataStream<WindowedValue<T>> source;
+    DataStream<WindowedValue<ValueWithRecordId<T>>> nonDedupSource;
+    TypeInformation<WindowedValue<T>> outputTypeInfo = context.getTypeInfo(output);
+
+    Coder<T> coder = output.getCoder();
+
+    TypeInformation<WindowedValue<ValueWithRecordId<T>>> withIdTypeInfo =
+        new CoderTypeInformation<>(
+            WindowedValues.getFullCoder(
+                ValueWithRecordId.ValueWithRecordIdCoder.of(coder),
+                output.getWindowingStrategy().getWindowFn().windowCoder()),
+            context.getPipelineOptions());
+
+    String fullName = getCurrentTransformName(context);
+    try {
+      int parallelism =
+          context.getExecutionEnvironment().getMaxParallelism() > 0
+              ? context.getExecutionEnvironment().getMaxParallelism()
+              : context.getExecutionEnvironment().getParallelism();
+
+      FlinkUnboundedSource<T> unboundedSource =
+          FlinkSource.unbounded(
+              transformName,
+              rawSource,
+              new SerializablePipelineOptions(context.getPipelineOptions()),
+              parallelism);
+      nonDedupSource =
+          context
+              .getExecutionEnvironment()
+              .fromSource(
+                  unboundedSource, WatermarkStrategy.noWatermarks(), fullName, withIdTypeInfo)
+              .uid(fullName);
+
+      if (rawSource.requiresDeduping()) {
+        source =
+            nonDedupSource
+                .keyBy(new ValueWithRecordIdKeySelector<>())
+                .transform(
+                    "deduping",
+                    outputTypeInfo,
+                    new DedupingOperator<>(context.getPipelineOptions()))
+                .uid(format("%s/__deduplicated__", fullName));
+      } else {
+        source =
+            nonDedupSource
+                .flatMap(new StripIdsMap<>(context.getPipelineOptions()))
+                .returns(outputTypeInfo);
+      }
+    } catch (Exception e) {
+      throw new RuntimeException("Error while translating UnboundedSource: " + rawSource, e);
+    }
+
+    context.setOutputDataStream(output, source);
+  }
+
+  /** Common translation logic for bounded sources. */
+  @SuppressWarnings("unchecked")
+  private static <T> void translateBoundedSource(
+      BoundedSource<T> rawSource, String transformName, FlinkStreamingTranslationContext context) {
+
+    PCollection<T> output =
+        (PCollection<T>)
+            Iterables.getOnlyElement(context.getCurrentTransform().getOutputs().values());
+
+    TypeInformation<WindowedValue<T>> outputTypeInfo = context.getTypeInfo(output);
+
+    String fullName = getCurrentTransformName(context);
+    int parallelism =
+        context.getExecutionEnvironment().getMaxParallelism() > 0
+            ? context.getExecutionEnvironment().getMaxParallelism()
+            : context.getExecutionEnvironment().getParallelism();
+
+    FlinkBoundedSource<T> flinkBoundedSource =
+        FlinkSource.bounded(
+            transformName,
+            rawSource,
+            new SerializablePipelineOptions(context.getPipelineOptions()),
+            parallelism);
+
+    SingleOutputStreamOperator<WindowedValue<T>> source;
+    try {
+      source =
+          context
+              .getExecutionEnvironment()
+              .fromSource(
+                  flinkBoundedSource, WatermarkStrategy.noWatermarks(), fullName, outputTypeInfo)
+              .uid(fullName)
+              .returns(outputTypeInfo);
+
+      if (!context.isStreaming()
+          && context
+              .getPipelineOptions()
+              .as(FlinkPipelineOptions.class)
+              .getForceSlotSharingGroup()) {
+        source = source.slotSharingGroup(FORCED_SLOT_GROUP);
+      }
+    } catch (Exception e) {
+      throw new RuntimeException("Error while translating BoundedSource: " + rawSource, e);
+    }
+
+    context.setOutputDataStream(output, source);
+  }
+
   private static class UnboundedReadSourceTranslator<T>
       extends FlinkStreamingPipelineTranslator.StreamTransformTranslator<
           PTransform<PBegin, PCollection<T>>> {
 
     @Override
     public void translateNode(
         PTransform<PBegin, PCollection<T>> transform, FlinkStreamingTranslationContext context) {
-      PCollection<T> output = context.getOutput(transform);
-
-      DataStream<WindowedValue<T>> source;
-      DataStream<WindowedValue<ValueWithRecordId<T>>> nonDedupSource;
-      TypeInformation<WindowedValue<T>> outputTypeInfo =
-          context.getTypeInfo(context.getOutput(transform));
-
-      Coder<T> coder = context.getOutput(transform).getCoder();
-
-      TypeInformation<WindowedValue<ValueWithRecordId<T>>> withIdTypeInfo =
-          new CoderTypeInformation<>(
-              WindowedValues.getFullCoder(
-                  ValueWithRecordId.ValueWithRecordIdCoder.of(coder),
-                  output.getWindowingStrategy().getWindowFn().windowCoder()),
-              context.getPipelineOptions());
-
       UnboundedSource<T, ?> rawSource;
       try {
         rawSource =
@@ -227,47 +326,7 @@ public void translateNode(
       } catch (IOException e) {
         throw new RuntimeException(e);
       }
-
-      String fullName = getCurrentTransformName(context);
-      try {
-        int parallelism =
-            context.getExecutionEnvironment().getMaxParallelism() > 0
-                ? context.getExecutionEnvironment().getMaxParallelism()
-                : context.getExecutionEnvironment().getParallelism();
-
-        FlinkUnboundedSource<T> unboundedSource =
-            FlinkSource.unbounded(
-                transform.getName(),
-                rawSource,
-                new SerializablePipelineOptions(context.getPipelineOptions()),
-                parallelism);
-        nonDedupSource =
-            context
-                .getExecutionEnvironment()
-                .fromSource(
-                    unboundedSource, WatermarkStrategy.noWatermarks(), fullName, withIdTypeInfo)
-                .uid(fullName);
-
-        if (rawSource.requiresDeduping()) {
-          source =
-              nonDedupSource
-                  .keyBy(new ValueWithRecordIdKeySelector<>())
-                  .transform(
-                      "deduping",
-                      outputTypeInfo,
-                      new DedupingOperator<>(context.getPipelineOptions()))
-                  .uid(format("%s/__deduplicated__", fullName));
-        } else {
-          source =
-              nonDedupSource
-                  .flatMap(new StripIdsMap<>(context.getPipelineOptions()))
-                  .returns(outputTypeInfo);
-        }
-      } catch (Exception e) {
-        throw new RuntimeException("Error while translating UnboundedSource: " + rawSource, e);
-      }
-
-      context.setOutputDataStream(output, source);
+      translateUnboundedSource(rawSource, transform.getName(), context);
     }
   }
 
@@ -285,66 +344,7 @@ private static class PrimitiveUnboundedReadTranslator<T>
     public void translateNode(
         SplittableParDo.PrimitiveUnboundedRead<T> transform,
         FlinkStreamingTranslationContext context) {
-
-      PCollection<T> output = context.getOutput(transform);
-
-      DataStream<WindowedValue<T>> source;
-      DataStream<WindowedValue<ValueWithRecordId<T>>> nonDedupSource;
-      TypeInformation<WindowedValue<T>> outputTypeInfo =
-          context.getTypeInfo(context.getOutput(transform));
-
-      Coder<T> coder = context.getOutput(transform).getCoder();
-
-      TypeInformation<WindowedValue<ValueWithRecordId<T>>> withIdTypeInfo =
-          new CoderTypeInformation<>(
-              WindowedValues.getFullCoder(
-                  ValueWithRecordId.ValueWithRecordIdCoder.of(coder),
-                  output.getWindowingStrategy().getWindowFn().windowCoder()),
-              context.getPipelineOptions());
-
-      // Get source directly from PrimitiveUnboundedRead (not via ReadTranslation)
-      UnboundedSource<T, ?> rawSource = transform.getSource();
-
-      String fullName = getCurrentTransformName(context);
-      try {
-        int parallelism =
-            context.getExecutionEnvironment().getMaxParallelism() > 0
-                ? context.getExecutionEnvironment().getMaxParallelism()
-                : context.getExecutionEnvironment().getParallelism();
-
-        FlinkUnboundedSource<T> unboundedSource =
-            FlinkSource.unbounded(
-                transform.getName(),
-                rawSource,
-                new SerializablePipelineOptions(context.getPipelineOptions()),
-                parallelism);
-        nonDedupSource =
-            context
-                .getExecutionEnvironment()
-                .fromSource(
-                    unboundedSource, WatermarkStrategy.noWatermarks(), fullName, withIdTypeInfo)
-                .uid(fullName);
-
-        if (rawSource.requiresDeduping()) {
-          source =
-              nonDedupSource
-                  .keyBy(new ValueWithRecordIdKeySelector<>())
-                  .transform(
-                      "deduping",
-                      outputTypeInfo,
-                      new DedupingOperator<>(context.getPipelineOptions()))
-                  .uid(format("%s/__deduplicated__", fullName));
-        } else {
-          source =
-              nonDedupSource
-                  .flatMap(new StripIdsMap<>(context.getPipelineOptions()))
-                  .returns(outputTypeInfo);
-        }
-      } catch (Exception e) {
-        throw new RuntimeException("Error while translating UnboundedSource: " + rawSource, e);
-      }
-
-      context.setOutputDataStream(output, source);
+      translateUnboundedSource(transform.getSource(), transform.getName(), context);
     }
   }
 
@@ -362,35 +362,7 @@ private static class PrimitiveBoundedReadTranslator<T>
     public void translateNode(
         SplittableParDo.PrimitiveBoundedRead<T> transform,
         FlinkStreamingTranslationContext context) {
-
-      PCollection<T> output = context.getOutput(transform);
-      TypeInformation<WindowedValue<T>> outputTypeInfo =
-          context.getTypeInfo(context.getOutput(transform));
-
-      // Get source directly from PrimitiveBoundedRead (not via ReadTranslation)
-      BoundedSource<T> rawSource = transform.getSource();
-
-      String fullName = getCurrentTransformName(context);
-      int parallelism =
-          context.getExecutionEnvironment().getMaxParallelism() > 0
-              ? context.getExecutionEnvironment().getMaxParallelism()
-              : context.getExecutionEnvironment().getParallelism();
-
-      FlinkBoundedSource<T> flinkBoundedSource =
-          FlinkSource.bounded(
-              fullName,
-              rawSource,
-              new SerializablePipelineOptions(context.getPipelineOptions()),
-              parallelism);
-
-      DataStream<WindowedValue<T>> source =
-          context
-              .getExecutionEnvironment()
-              .fromSource(
-                  flinkBoundedSource, WatermarkStrategy.noWatermarks(), fullName, outputTypeInfo)
-              .uid(fullName);
-
-      context.setOutputDataStream(output, source);
+      translateBoundedSource(transform.getSource(), transform.getName(), context);
     }
   }
 
@@ -503,11 +475,6 @@ private static class BoundedReadSourceTranslator<T>
     @Override
     public void translateNode(
         PTransform<PBegin, PCollection<T>> transform, FlinkStreamingTranslationContext context) {
-      PCollection<T> output = context.getOutput(transform);
-
-      TypeInformation<WindowedValue<T>> outputTypeInfo =
-          context.getTypeInfo(context.getOutput(transform));
-
       BoundedSource<T> rawSource;
       try {
         rawSource =
@@ -517,43 +484,7 @@ public void translateNode(
       } catch (IOException e) {
         throw new RuntimeException(e);
       }
-
-      String fullName = getCurrentTransformName(context);
-      int parallelism =
-          context.getExecutionEnvironment().getMaxParallelism() > 0
-              ? context.getExecutionEnvironment().getMaxParallelism()
-              : context.getExecutionEnvironment().getParallelism();
-
-      FlinkBoundedSource<T> flinkBoundedSource =
-          FlinkSource.bounded(
-              transform.getName(),
-              rawSource,
-              new SerializablePipelineOptions(context.getPipelineOptions()),
-              parallelism);
-
-      TypeInformation<WindowedValue<T>> typeInfo = context.getTypeInfo(output);
-
-      SingleOutputStreamOperator<WindowedValue<T>> source;
-      try {
-        source =
-            context
-                .getExecutionEnvironment()
-                .fromSource(
-                    flinkBoundedSource, WatermarkStrategy.noWatermarks(), fullName, outputTypeInfo)
-                .uid(fullName)
-                .returns(typeInfo);
-
-        if (!context.isStreaming()
-            && context
-                .getPipelineOptions()
-                .as(FlinkPipelineOptions.class)
-                .getForceSlotSharingGroup()) {
-          source = source.slotSharingGroup(FORCED_SLOT_GROUP);
-        }
-      } catch (Exception e) {
-        throw new RuntimeException("Error while translating BoundedSource: " + rawSource, e);
-      }
-      context.setOutputDataStream(output, source);
+      translateBoundedSource(rawSource, transform.getName(), context);
     }
   }