[spark] Fix memstream reflection picking wrong MemoryStream.apply overload

Author: kerwin-zk

paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/memstream.scala

@@ -48,11 +48,20 @@ object memstream {
   object MemoryStream {
     def apply[A: Encoder](implicit sqlContext: SQLContext): MemoryStream[A] = {
       val companion = loadCompanion()
+      // Spark 4.1 added a second 2-arg `apply(Encoder, SparkSession)` overload alongside the
+      // existing `apply(Encoder, SQLContext)`, so filtering only on `parameterCount == 2` is
+      // ambiguous — `Class#getMethods` ordering is JVM-dependent, and picking the wrong overload
+      // produces `IllegalArgumentException: argument type mismatch` when we pass a `SQLContext`.
+      // Pin the selection to the `(Encoder, SQLContext)` variant explicitly.
       val applyMethod = companion.getClass.getMethods
-        .filter(m => m.getName == "apply" && m.getParameterCount == 2)
-        .headOption
+        .find {
+          m =>
+            m.getName == "apply" &&
+            m.getParameterCount == 2 &&
+            m.getParameterTypes()(1) == classOf[SQLContext]
+        }
         .getOrElse(throw new NoSuchMethodError(
-          "No 2-arg MemoryStream.apply(Encoder, SQLContext) found on " + companion.getClass))
+          "No apply(Encoder, SQLContext) found on " + companion.getClass))
       val encoder = implicitly[Encoder[A]].asInstanceOf[AnyRef]
       val underlying = applyMethod.invoke(companion, encoder, sqlContext).asInstanceOf[AnyRef]
       new ReflectiveMemoryStream[A](underlying)