fix(java): fix find constructor error in generated serializer class c…

…aused by duplicated class classloading for Fury (#1948)

## What does this PR do?

fix duplicate classloading in parent classloader.

Some classloader such as flink classloader can load class from children
classloader. If fury is located in children classloader, but we are
serializing a class in such parent class, the parent class will load
Fury class again, which caused two Fury clases loaded.

## Related issues
Closes #1947 

## 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/builder/CodecUtils.java

@@ -28,6 +28,7 @@
 import org.apache.fury.resolver.ClassResolver;
 import org.apache.fury.resolver.FieldResolver;
 import org.apache.fury.serializer.Serializer;
+import org.apache.fury.util.ClassLoaderUtils;
 import org.apache.fury.util.Preconditions;
 
 /** Codec util to create and load jit serializer class. */
@@ -83,9 +84,26 @@ static <T> Class<? extends Serializer<T>> loadOrGenCodecClass(
       beanClassClassLoader = fury.getClass().getClassLoader();
     }
     ClassResolver classResolver = fury.getClassResolver();
+    codeGenerator = getCodeGenerator(fury, beanClassClassLoader, classResolver);
+    ClassLoader classLoader =
+        codeGenerator.compile(
+            Collections.singletonList(compileUnit), compileState -> compileState.lock.lock());
+    String className = codecBuilder.codecQualifiedClassName(beanClass);
+    try {
+      return (Class<? extends Serializer<T>>) classLoader.loadClass(className);
+    } catch (ClassNotFoundException e) {
+      throw new IllegalStateException("Impossible because we just compiled class", e);
+    }
+  }
+
+  private static CodeGenerator getCodeGenerator(
+      Fury fury, ClassLoader beanClassClassLoader, ClassResolver classResolver) {
+    CodeGenerator codeGenerator;
     try {
       // generated code imported fury classes.
-      beanClassClassLoader.loadClass(Fury.class.getName());
+      if (beanClassClassLoader.loadClass(Fury.class.getName()) != Fury.class) {
+        throw new ClassNotFoundException();
+      }
       codeGenerator = classResolver.getCodeGenerator(beanClassClassLoader);
       if (codeGenerator == null) {
         codeGenerator = CodeGenerator.getSharedCodeGenerator(beanClassClassLoader);
@@ -100,20 +118,12 @@ static <T> Class<? extends Serializer<T>> loadOrGenCodecClass(
       if (codeGenerator == null) {
         codeGenerator =
             CodeGenerator.getSharedCodeGenerator(
-                beanClassClassLoader, fury.getClass().getClassLoader());
+                ClassLoaderUtils.FuryJarClassLoader.getInstance(), beanClassClassLoader);
         // Hold strong reference of {@link CodeGenerator}, so the referent of `DelayedRef`
         // won't be null.
         classResolver.setCodeGenerator(loaders, codeGenerator);
       }
     }
-    ClassLoader classLoader =
-        codeGenerator.compile(
-            Collections.singletonList(compileUnit), compileState -> compileState.lock.lock());
-    String className = codecBuilder.codecQualifiedClassName(beanClass);
-    try {
-      return (Class<? extends Serializer<T>>) classLoader.loadClass(className);
-    } catch (ClassNotFoundException e) {
-      throw new IllegalStateException("Impossible because we just compiled class", e);
-    }
+    return codeGenerator;
   }
 }

java/fury-core/src/main/java/org/apache/fury/util/ClassLoaderUtils.java

@@ -33,6 +33,7 @@
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
+import org.apache.fury.Fury;
 import org.apache.fury.logging.Logger;
 import org.apache.fury.logging.LoggerFactory;
 import org.apache.fury.util.unsafe.DefineClass;
@@ -168,6 +169,33 @@ public URL getResource(String name) {
     }
   }
 
+  /** A classloader to load Fury jar classes only. */
+  public static class FuryJarClassLoader extends URLClassLoader {
+    static {
+      ClassLoader.registerAsParallelCapable();
+    }
+
+    private static final ParentClassLoader LOADER =
+        new ParentClassLoader(Fury.class.getClassLoader());
+    private static final FuryJarClassLoader FURY_JAR_LOADER = new FuryJarClassLoader();
+
+    private FuryJarClassLoader() {
+      super(new URL[0]);
+    }
+
+    @Override
+    public Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException {
+      if (name.startsWith(Fury.class.getPackage().getName()) && !name.contains("test")) {
+        return LOADER.loadClass(name, resolve);
+      }
+      return null;
+    }
+
+    public static FuryJarClassLoader getInstance() {
+      return FURY_JAR_LOADER;
+    }
+  }
+
   /**
    * A parallel loadable {@link ClassLoader} which make defineClass public for using in JDK17+.
    * {@link MethodHandle} can also be used for access `defineClass`.

java/fury-core/src/test/java/org/apache/fury/codegen/CodeGeneratorTest.java

@@ -38,6 +38,7 @@
 import org.apache.fury.test.bean.Foo;
 import org.apache.fury.util.ClassLoaderUtils;
 import org.apache.fury.util.ClassLoaderUtils.ByteArrayClassLoader;
+import org.apache.fury.util.ClassLoaderUtils.FuryJarClassLoader;
 import org.apache.fury.util.DelayedRef;
 import org.testng.Assert;
 import org.testng.annotations.Test;
@@ -71,6 +72,13 @@ public void testGetSharedCodeGenerator() {
         sharedCodeGenerator2
             .get(new Object[] {getClass().getClassLoader(), Fury.class.getClassLoader()})
             .get());
+    CodeGenerator.getSharedCodeGenerator(
+        FuryJarClassLoader.getInstance(), getClass().getClassLoader());
+    System.gc();
+    assertNotNull(
+        sharedCodeGenerator2
+            .get(new Object[] {FuryJarClassLoader.getInstance(), getClass().getClassLoader()})
+            .get());
   }
 
   @Test