[SPARK-36627][CORE] Fix java deserialization of proxy classes

core/src/main/scala/org/apache/spark/serializer/JavaSerializer.scala

@@ -71,6 +71,13 @@ private[spark] class JavaDeserializationStream(in: InputStream, loader: ClassLoa
         case e: ClassNotFoundException =>
           JavaDeserializationStream.primitiveMappings.getOrElse(desc.getName, throw e)
       }
+
+    override def resolveProxyClass(ifaces: Array[String]): Class[_] = {
+        // scalastyle:off classforname
+        val resolved = ifaces.map(iface => Class.forName(iface, false, loader))
+        // scalastyle:on classforname
+        java.lang.reflect.Proxy.getProxyClass(loader, resolved: _*)
+      }
   }
 
   def readObject[T: ClassTag](): T = objIn.readObject().asInstanceOf[T]

core/src/test/java/org/apache/spark/serializer/ProxySerializerTest.java

@@ -0,0 +1,33 @@
+package org.apache.spark.serializer;
+
+import java.io.Serializable;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+
+class ContainsProxyClass implements Serializable {
+    final MyInterface proxy = (MyInterface) Proxy.newProxyInstance(
+            MyInterface.class.getClassLoader(),
+            new Class[]{MyInterface.class},
+            new MyInvocationHandler());
+
+    // Interface needs to be public as classloaders will mismatch - see ObjectInputStream#resolveProxyClass for details.
+    public interface MyInterface {
+        void myMethod();
+    }
+
+    static class MyClass implements MyInterface, Serializable {
+        @Override
+        public void myMethod() {
+        }
+    }
+
+    class MyInvocationHandler implements InvocationHandler, Serializable {
+        private final MyClass real = new MyClass();
+
+        @Override
+        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
+            return method.invoke(real, args);
+        }
+    }
+}

core/src/test/scala/org/apache/spark/serializer/JavaSerializerSuite.scala

@@ -31,11 +31,33 @@ class JavaSerializerSuite extends SparkFunSuite {
   test("Deserialize object containing a primitive Class as attribute") {
     val serializer = new JavaSerializer(new SparkConf())
     val instance = serializer.newInstance()
-    val obj = instance.deserialize[ContainsPrimitiveClass](instance.serialize(
-      new ContainsPrimitiveClass()))
+    val obj = instance.deserialize[ContainsPrimitiveClass](
+      instance.serialize(new ContainsPrimitiveClass()))
     // enforce class cast
     obj.getClass
   }
+
+  test("SPARK-36627: Deserialize object containing a proxy Class as attribute") {
+    var classesLoaded = Set[String]()
+    val outer = Thread.currentThread.getContextClassLoader
+    val inner = new ClassLoader() {
+      override def loadClass(name: String): Class[_] = {
+        classesLoaded = classesLoaded + name
+        outer.loadClass(name)
+      }
+    }
+    Thread.currentThread.setContextClassLoader(inner)
+
+    val serializer = new JavaSerializer(new SparkConf())
+    val instance = serializer.newInstance()
+    val obj =
+      instance.deserialize[ContainsProxyClass](instance.serialize(new ContainsProxyClass()))
+    // enforce class cast
+    obj.getClass
+
+    // check that serializer's loader is used to resolve proxied interface.
+    assert(classesLoaded.exists(klass => klass.contains("MyInterface")))
+  }
 }
 
 private class ContainsPrimitiveClass extends Serializable {