refactor: make shared Spark source compatible with Scala 2.12 and 2.13

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: tkaymak

runners/spark/src/main/java/org/apache/beam/runners/spark/coders/SparkRunnerKryoRegistrator.java

@@ -30,7 +30,6 @@
 import org.apache.beam.sdk.values.TupleTag;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.HashBasedTable;
 import org.apache.spark.serializer.KryoRegistrator;
-import scala.collection.mutable.WrappedArray;
 
 /**
  * Custom {@link KryoRegistrator}s for Beam's Spark runner needs and registering used class in spark
@@ -61,7 +60,16 @@ public void registerClasses(Kryo kryo) {
     kryo.register(PaneInfo.class);
     kryo.register(StateAndTimers.class);
     kryo.register(TupleTag.class);
-    kryo.register(WrappedArray.ofRef.class);
+    // Scala 2.12 uses WrappedArray$ofRef, Scala 2.13 renamed it to ArraySeq$ofRef
+    try {
+      kryo.register(Class.forName("scala.collection.mutable.ArraySeq$ofRef"));
+    } catch (ClassNotFoundException e) {
+      try {
+        kryo.register(Class.forName("scala.collection.mutable.WrappedArray$ofRef"));
+      } catch (ClassNotFoundException ignored) {
+        // Neither class found; skip registration
+      }
+    }
 
     try {
       kryo.register(

runners/spark/src/main/java/org/apache/beam/runners/spark/io/SourceRDD.java

@@ -50,7 +50,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import scala.Option;
-import scala.collection.JavaConversions;
+import scala.collection.JavaConverters;
 
 /** Classes implementing Beam {@link Source} {@link RDD}s. */
 @SuppressWarnings({
@@ -75,7 +75,7 @@ public static class Bounded<T> extends RDD<WindowedValue<T>> {
 
     // to satisfy Scala API.
     private static final scala.collection.immutable.Seq<Dependency<?>> NIL =
-        JavaConversions.asScalaBuffer(Collections.<Dependency<?>>emptyList()).toList();
+        JavaConverters.asScalaBuffer(Collections.<Dependency<?>>emptyList()).toList();
 
     public Bounded(
         SparkContext sc,
@@ -148,7 +148,7 @@ public scala.collection.Iterator<WindowedValue<T>> compute(
       final Iterator<WindowedValue<T>> readerIterator =
           new ReaderToIteratorAdapter<>(metricsContainer, reader);
 
-      return new InterruptibleIterator<>(context, JavaConversions.asScalaIterator(readerIterator));
+      return new InterruptibleIterator<>(context, JavaConverters.asScalaIterator(readerIterator));
     }
 
     /**
@@ -299,7 +299,7 @@ public static class Unbounded<T, CheckpointMarkT extends UnboundedSource.Checkpo
 
     // to satisfy Scala API.
     private static final scala.collection.immutable.List<Dependency<?>> NIL =
-        JavaConversions.asScalaBuffer(Collections.<Dependency<?>>emptyList()).toList();
+        JavaConverters.asScalaBuffer(Collections.<Dependency<?>>emptyList()).toList();
 
     public Unbounded(
         SparkContext sc,
@@ -344,7 +344,7 @@ public scala.collection.Iterator<scala.Tuple2<Source<T>, CheckpointMarkT>> compu
           (CheckpointableSourcePartition<T, CheckpointMarkT>) split;
       scala.Tuple2<Source<T>, CheckpointMarkT> tuple2 =
           new scala.Tuple2<>(partition.getSource(), partition.checkpointMark);
-      return JavaConversions.asScalaIterator(Collections.singleton(tuple2).iterator());
+      return JavaConverters.asScalaIterator(Collections.singleton(tuple2).iterator());
     }
   }
 

runners/spark/src/main/java/org/apache/beam/runners/spark/io/SparkUnboundedSource.java

@@ -186,7 +186,7 @@ public Duration slideDuration() {
 
     @Override
     public scala.collection.immutable.List<DStream<?>> dependencies() {
-      return scala.collection.JavaConversions.asScalaBuffer(
+      return scala.collection.JavaConverters.asScalaBuffer(
               Collections.<DStream<?>>singletonList(parent))
           .toList();
     }

runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkGroupAlsoByWindowViaWindowSet.java

@@ -73,7 +73,7 @@
 import scala.Tuple2;
 import scala.Tuple3;
 import scala.collection.Iterator;
-import scala.collection.JavaConversions;
+import scala.collection.JavaConverters;
 import scala.collection.Seq;
 import scala.runtime.AbstractFunction1;
 
@@ -238,7 +238,7 @@ private Collection<TimerInternals.TimerData> filterTimersEligibleForProcessing(
             // new input for key.
             try {
               final Iterable<WindowedValue<InputT>> elements =
-                  FluentIterable.from(JavaConversions.asJavaIterable(encodedElements))
+                  FluentIterable.from(JavaConverters.asJavaIterable(encodedElements))
                       .transform(bytes -> CoderHelpers.fromByteArray(bytes, wvCoder));
 
               LOG.trace("{}: input elements: {}", logPrefix, elements);
@@ -410,7 +410,7 @@ private Collection<TimerInternals.TimerData> filterTimersEligibleForProcessing(
         droppedDueToClosedWindow.inc(-droppedDueToClosedWindow.getCumulative());
       }
 
-      return scala.collection.JavaConversions.asScalaIterator(
+      return JavaConverters.asScalaIterator(
           new UpdateStateByKeyOutputIterator(input, reduceFn, droppedDueToLateness));
     }
   }
@@ -522,7 +522,9 @@ JavaDStream<WindowedValue<KV<K, Iterable<InputT>>>> groupByKeyAndWindow(
             Tuple2</*K*/ ByteArray, Tuple2<StateAndTimers, /*WV<KV<K, Itr<I>>>*/ List<byte[]>>>>
         firedStream =
             pairDStream.updateStateByKey(
-                updateFunc,
+                // Raw cast to AbstractFunction1 suppresses Scala 2.12 (collection.Seq) vs
+                // Scala 2.13 (immutable.Seq) type difference — safe at runtime due to erasure.
+                (scala.runtime.AbstractFunction1) updateFunc,
                 pairDStream.defaultPartitioner(pairDStream.defaultPartitioner$default$1()),
                 true,
                 JavaSparkContext$.MODULE$.fakeClassTag());

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java

@@ -330,7 +330,8 @@ private static <T> void translateFlatten(
         }
       }
       // Unify streams into a single stream.
-      unifiedStreams = context.getStreamingContext().union(JavaConverters.asScalaBuffer(dStreams));
+      unifiedStreams =
+          context.getStreamingContext().union(JavaConverters.asScalaBuffer(dStreams).toList());
     }
 
     context.pushDataset(

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/ParDoStateUpdateFn.java

@@ -19,6 +19,7 @@
 
 import java.io.Serializable;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -62,7 +63,6 @@
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.sparkproject.guava.collect.Iterators;
 import scala.Option;
 import scala.Tuple2;
 import scala.runtime.AbstractFunction3;
@@ -236,7 +236,7 @@ public TimerInternals timerInternals() {
     final byte[] byteValue = serializedValue.get();
     @Nullable WindowedValue<ValueT> windowedValue;
     @Nullable WindowedValue<KV<KeyT, ValueT>> keyedWindowedValue;
-    Iterator<WindowedValue<KV<KeyT, ValueT>>> iterator = Iterators.emptyIterator();
+    Iterator<WindowedValue<KV<KeyT, ValueT>>> iterator = Collections.emptyIterator();
     if (byteValue.length > 0) {
       windowedValue = CoderHelpers.fromByteArray(byteValue, this.wvCoder);
       keyedWindowedValue = windowedValue.withValue(KV.of(key, windowedValue.getValue()));

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/StreamingTransformTranslator.java

@@ -306,7 +306,7 @@ public void evaluate(Flatten.PCollections<T> transform, EvaluationContext contex
         }
         // start by unifying streams into a single stream.
         JavaDStream<WindowedValue<T>> unifiedStreams =
-            context.getStreamingContext().union(JavaConverters.asScalaBuffer(dStreams));
+            context.getStreamingContext().union(JavaConverters.asScalaBuffer(dStreams).toList());
         context.putDataset(transform, new UnboundedDataset<>(unifiedStreams, streamingSources));
       }