[Flink Runner] Add translators for PrimitiveUnboundedRead and PrimitiveBoundedRead

This commit adds explicit translators for SplittableParDo.PrimitiveUnboundedRead
and SplittableParDo.PrimitiveBoundedRead to the Flink streaming transform
translators.

The Flink classic runner calls convertReadBasedSplittableDoFnsToPrimitiveReadsIfNecessary()
when NOT using beam_fn_api experiment, which converts SDF-wrapped reads to
PrimitiveUnboundedRead and PrimitiveBoundedRead. Without translators for these
transforms, pipelines using unbounded sources like KinesisIO.read() fail with:
'No translator known for PrimitiveUnboundedRead'

Changes:
- Add PrimitiveUnboundedReadTranslator class
- Add PrimitiveBoundedReadTranslator class
- Modify getTranslator() to handle these transforms before URN lookup
- Add unit tests for the new translators
- Update CHANGES.md with bugfix entry

Related to #20530

Author: dnamaz

CHANGES.md

@@ -84,6 +84,7 @@
 
 ## Bugfixes
 
+* Fixed Flink classic runner failing with "No translator known for PrimitiveUnboundedRead" when using unbounded source connectors like KinesisIO after SDF-to-primitive-read conversion (Java) ([#XXXXX](https://github.com/apache/beam/issues/XXXXX)).
 * Fixed FirestoreV1 Beam connectors allow configuring inconsistent project/database IDs between RPC requests and routing headers #36895 (Java) ([#36895](https://github.com/apache/beam/issues/36895)).
  Logical type and coder registry are saved for pipelines in the case of default pickler. This fixes a side effect of switching to cloudpickle as default pickler in Beam 2.65.0 (Python) ([#35738](https://github.com/apache/beam/issues/35738)).
 

runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslatorsTest.java

@@ -196,7 +196,8 @@ public void primitiveUnboundedReadTranslatorProducesCorrectSource() {
 
     FlinkSource<?, ?> source =
         (FlinkSource<?, ?>)
-            ((SourceTransformation<?, ?, ?>) Iterables.getOnlyElement(oneInputTransform.getInputs()))
+            ((SourceTransformation<?, ?, ?>)
+                    Iterables.getOnlyElement(oneInputTransform.getInputs()))
                 .getSource();
 
     assertEquals(maxParallelism, source.getNumSplits());
@@ -208,7 +209,8 @@ public void primitiveBoundedReadTranslatorProducesCorrectSource() {
     final int parallelism = 2;
 
     SplittableParDo.PrimitiveBoundedRead<String> transform =
-        new SplittableParDo.PrimitiveBoundedRead<>(Read.from(new TestBoundedSource(maxParallelism)));
+        new SplittableParDo.PrimitiveBoundedRead<>(
+            Read.from(new TestBoundedSource(maxParallelism)));
     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
     env.setParallelism(parallelism);
     env.setMaxParallelism(maxParallelism);
@@ -288,8 +290,7 @@ private Object applyReadSourceTransformWithTranslator(
             Pipeline.create());
 
     ctx.setCurrentTransform(appliedTransform);
-    ((FlinkStreamingPipelineTranslator.StreamTransformTranslator<PTransform<?, ?>>)
-            translator)
+    ((FlinkStreamingPipelineTranslator.StreamTransformTranslator<PTransform<?, ?>>) translator)
         .translateNode(transform, ctx);
 
     return ctx.getInputDataStream(pc).getTransformation();