fix(java): fix jdk proxy serialization when proxy writePlace method (#…

…1857)

## What does this PR do?

fix jdk proxy serialization when proxy writePlace method
## Related issues

Closes #1854 

## Does this PR introduce any user-facing change?

<!--
If any user-facing interface changes, please [open an
issue](https://github.com/apache/fury/issues/new/choose) describing the
need to do so and update the document if necessary.
-->

- [ ] Does this PR introduce any public API change?
- [ ] Does this PR introduce any binary protocol compatibility change?

## Benchmark

<!--
When the PR has an impact on performance (if you don't know whether the
PR will have an impact on performance, you can submit the PR first, and
if it will have impact on performance, the code reviewer will explain
it), be sure to attach a benchmark data here.
-->

java/fury-core/src/main/java/org/apache/fury/resolver/ClassInfo.java

@@ -114,7 +114,7 @@ public class ClassInfo {
     this.classId = classId;
     if (cls != null) {
       boolean isLambda = Functions.isLambda(cls);
-      boolean isProxy = ReflectionUtils.isJdkProxy(cls);
+      boolean isProxy = classId != ClassResolver.REPLACE_STUB_ID && ReflectionUtils.isJdkProxy(cls);
       this.isDynamicGeneratedClass = isLambda || isProxy;
       if (isLambda) {
         this.classId = ClassResolver.LAMBDA_STUB_ID;

java/fury-core/src/main/java/org/apache/fury/resolver/ClassResolver.java

@@ -853,7 +853,11 @@ public Class<? extends Serializer> getSerializerClass(Class<?> cls, boolean code
       } else if (Functions.isLambda(cls)) {
         return LambdaSerializer.class;
       } else if (ReflectionUtils.isJdkProxy(cls)) {
-        return JdkProxySerializer.class;
+        if (JavaSerializer.getWriteReplaceMethod(cls) != null) {
+          return ReplaceResolveSerializer.class;
+        } else {
+          return JdkProxySerializer.class;
+        }
       } else if (Calendar.class.isAssignableFrom(cls)) {
         return TimeSerializers.CalendarSerializer.class;
       } else if (ZoneId.class.isAssignableFrom(cls)) {

java/fury-core/src/test/java/org/apache/fury/serializer/JdkProxySerializerTest.java

@@ -21,7 +21,9 @@
 
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotSame;
+import static org.testng.Assert.assertTrue;
 
+import java.io.ObjectStreamException;
 import java.io.Serializable;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Method;
@@ -30,6 +32,7 @@
 import org.apache.fury.Fury;
 import org.apache.fury.FuryTestBase;
 import org.apache.fury.config.Language;
+import org.apache.fury.reflect.ReflectionUtils;
 import org.testng.annotations.Test;
 
 @SuppressWarnings({"unchecked", "rawtypes"})
@@ -129,4 +132,91 @@ public void testJdkProxyRef(Fury fury) {
         (RefTestInvocationHandler) Proxy.getInvocationHandler(copy);
     assertEquals(copyHandler.getProxy(), copy);
   }
+
+  @Test
+  public void testSerializeProxyWriteReplace() {
+    final Fury fury =
+        Fury.builder().withLanguage(Language.JAVA).requireClassRegistration(false).build();
+
+    final Object o = ProxyFactory.createProxy(TestInterface.class);
+    final byte[] s = fury.serialize(o);
+    assertTrue(ReflectionUtils.isJdkProxy(fury.deserialize(s).getClass()));
+  }
+
+  interface TestInterface {
+    void test();
+  }
+
+  static class ProxyFactory {
+
+    static <T> T createProxy(final Class<T> type) {
+      return new JdkProxyFactory().createProxy(type);
+    }
+
+    public interface IWriteReplace {
+      Object writeReplace() throws ObjectStreamException;
+    }
+
+    static final class JdkProxyFactory {
+
+      @SuppressWarnings("unchecked")
+      <T> T createProxy(final Class<T> type) {
+        final JdkHandler handler = new JdkHandler(type);
+        try {
+          final ClassLoader cl = Thread.currentThread().getContextClassLoader();
+          return (T)
+              Proxy.newProxyInstance(
+                  cl, new Class[] {type, IWriteReplace.class, Serializable.class}, handler);
+        } catch (IllegalArgumentException e) {
+          throw new RuntimeException("Could not create proxy for type [" + type.getName() + "]", e);
+        }
+      }
+
+      static class JdkHandler implements InvocationHandler, IWriteReplace, Serializable {
+
+        private final String typeName;
+
+        private JdkHandler(Class<?> type) {
+          typeName = type.getName();
+        }
+
+        @Override
+        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
+          if (isWriteReplaceMethod(method)) {
+            return writeReplace();
+          }
+          return null;
+        }
+
+        public Object writeReplace() throws ObjectStreamException {
+          return new ProxyReplacement(typeName);
+        }
+
+        static boolean isWriteReplaceMethod(final Method method) {
+          return (method.getReturnType() == Object.class)
+              && (method.getParameterTypes().length == 0)
+              && method.getName().equals("writeReplace");
+        }
+      }
+
+      public static final class ProxyReplacement implements Serializable {
+
+        private final String type;
+
+        public ProxyReplacement(final String type) {
+          this.type = type;
+        }
+
+        private Object readResolve() throws ObjectStreamException {
+          try {
+            final Class<?> clazz =
+                Class.forName(type, false, Thread.currentThread().getContextClassLoader());
+            return ProxyFactory.createProxy(clazz);
+          } catch (ClassNotFoundException ex) {
+            throw new RuntimeException(ex);
+          }
+        }
+      }
+    }
+  }
 }