[runners-spark] Cover Scala-array fallback and EvaluationContext error paths

Address codecov/patch on PR #38255 by exercising the new branches added for
Scala 2.13 / null-safe error logging:

- Refactor SparkRunnerKryoRegistrator's nested Scala-array Class.forName
  fallback into a small @VisibleForTesting findFirstAvailableClass helper
  and add unit tests for first-hit, fallback, no-match, and empty-input
  paths.
- Add EvaluationContextTest covering the catch (RuntimeException) /
  catch (Exception) blocks in evaluate() and collect(), including the
  null-message path that motivated the String.valueOf wrap.

Author: tkaymak

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

@@ -28,8 +28,10 @@
 import org.apache.beam.sdk.transforms.windowing.PaneInfo;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.TupleTag;
+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.HashBasedTable;
 import org.apache.spark.serializer.KryoRegistrator;
+import org.checkerframework.checker.nullness.qual.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -65,18 +67,18 @@ public void registerClasses(Kryo kryo) {
     kryo.register(StateAndTimers.class);
     kryo.register(TupleTag.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) {
-        LOG.warn(
-            "Neither scala.collection.mutable.ArraySeq$ofRef (Scala 2.13) nor "
-                + "scala.collection.mutable.WrappedArray$ofRef (Scala 2.12) was found on the "
-                + "classpath. Kryo serialization of Scala wrapped arrays will fall back to Java "
-                + "serialization or fail at runtime if spark.kryo.registrationRequired is true.");
-      }
+    Class<?> scalaArrayClass =
+        findFirstAvailableClass(
+            "scala.collection.mutable.ArraySeq$ofRef",
+            "scala.collection.mutable.WrappedArray$ofRef");
+    if (scalaArrayClass != null) {
+      kryo.register(scalaArrayClass);
+    } else {
+      LOG.warn(
+          "Neither scala.collection.mutable.ArraySeq$ofRef (Scala 2.13) nor "
+              + "scala.collection.mutable.WrappedArray$ofRef (Scala 2.12) was found on the "
+              + "classpath. Kryo serialization of Scala wrapped arrays will fall back to Java "
+              + "serialization or fail at runtime if spark.kryo.registrationRequired is true.");
     }
 
     try {
@@ -90,4 +92,16 @@ public void registerClasses(Kryo kryo) {
       throw new IllegalStateException("Unable to register classes with kryo.", e);
     }
   }
+
+  @VisibleForTesting
+  static @Nullable Class<?> findFirstAvailableClass(String... classNames) {
+    for (String name : classNames) {
+      try {
+        return Class.forName(name);
+      } catch (ClassNotFoundException ignored) {
+        // try the next candidate
+      }
+    }
+    return null;
+  }
 }

runners/spark/src/test/java/org/apache/beam/runners/spark/coders/SparkRunnerKryoRegistratorTest.java

@@ -17,7 +17,10 @@
  */
 package org.apache.beam.runners.spark.coders;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
 import com.esotericsoftware.kryo.Kryo;
@@ -73,6 +76,54 @@ public void testDefaultSerializerNotCallingKryo() {
     }
   }
 
+  /** Unit tests for the {@link SparkRunnerKryoRegistrator#findFirstAvailableClass} helper. */
+  public static class FindFirstAvailableClassTest {
+
+    @Test
+    public void returnsFirstWhenAvailable() {
+      Class<?> result =
+          SparkRunnerKryoRegistrator.findFirstAvailableClass(
+              "java.lang.String", "java.lang.Integer");
+      assertSame(String.class, result);
+    }
+
+    @Test
+    public void fallsBackWhenFirstMissing() {
+      Class<?> result =
+          SparkRunnerKryoRegistrator.findFirstAvailableClass(
+              "does.not.Exist", "java.lang.Integer");
+      assertSame(Integer.class, result);
+    }
+
+    @Test
+    public void returnsNullWhenNoneAvailable() {
+      Class<?> result =
+          SparkRunnerKryoRegistrator.findFirstAvailableClass(
+              "does.not.Exist1", "does.not.Exist2");
+      assertNull(result);
+    }
+
+    @Test
+    public void returnsNullForEmptyInput() {
+      assertNull(SparkRunnerKryoRegistrator.findFirstAvailableClass());
+    }
+
+    @Test
+    public void resolvesScalaWrappedArrayClassOnRealClasspath() {
+      // On any supported Scala version (2.12 ArraySeq$ofRef does not exist; 2.13 it does), at
+      // least one of the two wrapped-array class names must resolve. This is the production call
+      // the registrator makes.
+      Class<?> result =
+          SparkRunnerKryoRegistrator.findFirstAvailableClass(
+              "scala.collection.mutable.ArraySeq$ofRef",
+              "scala.collection.mutable.WrappedArray$ofRef");
+      assertEquals(
+          "expected one of the Scala wrapped-array classes to be on the classpath",
+          true,
+          result != null);
+    }
+  }
+
   // Hide TestKryoRegistrator from the Enclosed JUnit runner
   interface Others {
     class TestKryoRegistrator extends SparkRunnerKryoRegistrator {

runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/EvaluationContextTest.java

@@ -0,0 +1,84 @@
+/*
+ * 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.spark.structuredstreaming.translation;
+
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+
+import org.apache.spark.sql.Dataset;
+import org.junit.Test;
+
+/**
+ * Unit tests for the static error-path branches of {@link EvaluationContext}. The happy-path
+ * branches are covered end-to-end by the structured-streaming translation tests; these tests
+ * specifically exercise the {@code catch} blocks that wrap and rethrow underlying Spark failures.
+ */
+public class EvaluationContextTest {
+
+  @Test
+  public void evaluateWrapsAndRethrowsRuntimeException() {
+    @SuppressWarnings("unchecked")
+    Dataset<Object> ds = mock(Dataset.class);
+    RuntimeException underlying = new RuntimeException("boom");
+    doThrow(underlying).when(ds).write();
+
+    RuntimeException thrown =
+        assertThrows(RuntimeException.class, () -> EvaluationContext.evaluate("test-ds", ds));
+    assertSame(underlying, thrown.getCause());
+  }
+
+  @Test
+  public void evaluateHandlesNullExceptionMessage() {
+    // Reproduces the original NPE motivation for the String.valueOf wrap: a RuntimeException
+    // whose root cause carries a null message must not crash the error logger.
+    @SuppressWarnings("unchecked")
+    Dataset<Object> ds = mock(Dataset.class);
+    RuntimeException underlying = new RuntimeException((String) null);
+    doThrow(underlying).when(ds).write();
+
+    RuntimeException thrown =
+        assertThrows(RuntimeException.class, () -> EvaluationContext.evaluate("test-ds", ds));
+    assertSame(underlying, thrown.getCause());
+  }
+
+  @Test
+  public void collectWrapsAndRethrowsException() {
+    @SuppressWarnings("unchecked")
+    Dataset<Object> ds = mock(Dataset.class);
+    RuntimeException underlying = new RuntimeException("boom");
+    doThrow(underlying).when(ds).collect();
+
+    RuntimeException thrown =
+        assertThrows(RuntimeException.class, () -> EvaluationContext.collect("test-ds", ds));
+    assertSame(underlying, thrown.getCause());
+  }
+
+  @Test
+  public void collectHandlesNullExceptionMessage() {
+    @SuppressWarnings("unchecked")
+    Dataset<Object> ds = mock(Dataset.class);
+    RuntimeException underlying = new RuntimeException((String) null);
+    doThrow(underlying).when(ds).collect();
+
+    RuntimeException thrown =
+        assertThrows(RuntimeException.class, () -> EvaluationContext.collect("test-ds", ds));
+    assertSame(underlying, thrown.getCause());
+  }
+}