[runners-spark] Use robust constructor resolution in EncoderFactory (#38271)

* [runners-spark] Use robust constructor resolution in EncoderFactory

Replace `(Constructor<X>) X.class.getConstructors()[0]` for
StaticInvoke, Invoke, and NewInstance with a `primaryConstructor()`
helper that picks the constructor with the most parameters.
Class.getConstructors() returns constructors in JVM-defined order
that is not guaranteed stable, so resolving the widest constructor
explicitly makes the lookup robust to future Spark releases that
add overloaded constructors.

Today this is a no-op: StaticInvoke / Invoke / NewInstance only
have one public constructor each in Spark 3.1.x through 3.5.x, so
getConstructors()[0] and the widest constructor resolve to the
same one. The change is purely defensive.

Same fix has already landed in the Spark 4 override in PR #38255
(commit 9c071c5, flagged by Gemini Code Assist round 1). Porting
it to the shared base keeps both code paths consistent.

* Address Gemini review: guard primaryConstructor against empty ctors

If Class.getConstructors() returns empty (e.g. all public constructors
removed in a future Spark version), throw an IllegalStateException with
the class name instead of letting ctors[0] surface as a cryptic
ArrayIndexOutOfBoundsException during static initialization.

Author: tkaymak

runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderFactory.java

@@ -31,15 +31,32 @@
 import scala.reflect.ClassTag;
 
 public class EncoderFactory {
-  // default constructor to reflectively create static invoke expressions
+  // Resolve the Scala case-class primary constructor (the one with the most parameters).
+  // Constructor ordering returned by Class.getConstructors() is JVM-defined and not stable
+  // across Spark versions, so we pick the widest constructor explicitly and then dispatch on
+  // parameter count below to pick the right argument shape per Spark version.
   private static final Constructor<StaticInvoke> STATIC_INVOKE_CONSTRUCTOR =
-      (Constructor<StaticInvoke>) StaticInvoke.class.getConstructors()[0];
+      primaryConstructor(StaticInvoke.class);
 
-  private static final Constructor<Invoke> INVOKE_CONSTRUCTOR =
-      (Constructor<Invoke>) Invoke.class.getConstructors()[0];
+  private static final Constructor<Invoke> INVOKE_CONSTRUCTOR = primaryConstructor(Invoke.class);
 
   private static final Constructor<NewInstance> NEW_INSTANCE_CONSTRUCTOR =
-      (Constructor<NewInstance>) NewInstance.class.getConstructors()[0];
+      primaryConstructor(NewInstance.class);
+
+  @SuppressWarnings("unchecked")
+  private static <T> Constructor<T> primaryConstructor(Class<T> cls) {
+    Constructor<?>[] ctors = cls.getConstructors();
+    if (ctors.length == 0) {
+      throw new IllegalStateException("No public constructors found for " + cls.getName());
+    }
+    Constructor<?> widest = ctors[0];
+    for (int i = 1; i < ctors.length; i++) {
+      if (ctors[i].getParameterCount() > widest.getParameterCount()) {
+        widest = ctors[i];
+      }
+    }
+    return (Constructor<T>) widest;
+  }
 
   static <T> ExpressionEncoder<T> create(
       Expression serializer, Expression deserializer, Class<? super T> clazz) {