feat(java): deserialize one pojo into another type

java/fury-core/src/main/java/org/apache/fury/Fury.java

@@ -1122,7 +1122,7 @@ public <T> T deserializeJavaObject(MemoryBuffer buffer, Class<T> cls) {
       if (nextReadRefId >= NOT_NULL_VALUE_FLAG) {
         ClassInfo classInfo;
         if (shareMeta) {
-          classInfo = classResolver.readClassInfo(buffer);
+          classInfo = classResolver.readClassInfo(buffer, cls);
         } else {
           classInfo = classResolver.getClassInfo(cls);
         }

java/fury-core/src/main/java/org/apache/fury/meta/ClassDef.java

@@ -48,6 +48,7 @@
 import org.apache.fury.memory.Platform;
 import org.apache.fury.reflect.ReflectionUtils;
 import org.apache.fury.reflect.TypeRef;
+import org.apache.fury.resolver.ClassInfo;
 import org.apache.fury.resolver.ClassResolver;
 import org.apache.fury.serializer.CompatibleSerializer;
 import org.apache.fury.serializer.NonexistentClass;
@@ -787,4 +788,12 @@ public static ClassDef buildClassDef(
       ClassResolver classResolver, Class<?> type, List<Field> fields, boolean isObjectType) {
     return ClassDefEncoder.buildClassDef(classResolver, type, fields, isObjectType);
   }
+
+  public static ClassDef replaceRootClassTo(
+      ClassResolver classResolver,
+      ClassInfo targetCls
+  ) {
+    return ClassDefEncoder.buildClassDefWithFieldInfos(
+        classResolver, targetCls.getCls(), targetCls.classDef.getFieldsInfo(), targetCls.classDef.isObjectType);
+  }
 }

java/fury-core/src/main/java/org/apache/fury/meta/ClassDefEncoder.java

@@ -105,7 +105,11 @@ static List<FieldInfo> buildFieldsInfo(ClassResolver resolver, List<Field> field
   /** Build class definition from fields of class. */
   static ClassDef buildClassDef(
       ClassResolver classResolver, Class<?> type, List<Field> fields, boolean isObjectType) {
-    List<FieldInfo> fieldInfos = buildFieldsInfo(classResolver, fields);
+    return buildClassDefWithFieldInfos(classResolver, type, buildFieldsInfo(classResolver, fields), isObjectType);
+  }
+
+  static ClassDef buildClassDefWithFieldInfos(
+      ClassResolver classResolver, Class<?> type, List<ClassDef.FieldInfo> fieldInfos, boolean isObjectType) {
     Map<String, List<FieldInfo>> classLayers = getClassFields(type, fieldInfos);
     fieldInfos = new ArrayList<>(fieldInfos.size());
     classLayers.values().forEach(fieldInfos::addAll);

java/fury-core/src/main/java/org/apache/fury/meta/ClassSpec.java

@@ -22,7 +22,7 @@
 import org.apache.fury.type.TypeUtils;
 
 public class ClassSpec {
-  public final String entireClassName;
+  public String entireClassName;
 
   /** Whether current class is enum of component is enum if current class is array. */
   public final boolean isEnum;

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

@@ -48,7 +48,7 @@ public class ClassInfo {
   // use primitive to avoid boxing
   // class id must be less than Integer.MAX_VALUE/2 since we use bit 0 as class id flag.
   short classId;
-  ClassDef classDef;
+  public ClassDef classDef;
   boolean needToWriteClassDef;
 
   ClassInfo(

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

@@ -1674,6 +1674,33 @@ public ClassInfo readClassInfo(MemoryBuffer buffer) {
     return classInfo;
   }
 
+  public ClassInfo readClassInfo(MemoryBuffer buffer, Class<?> targetClass) {
+    if (metaContextShareEnabled) {
+      ClassInfo classInfo = readClassInfoWithMetaShare(buffer, fury.getSerializationContext().getMetaContext());
+
+      // replace target class if needed
+      if (!targetClass.getName().equals(classInfo.getCls().getName())) {
+        ClassInfo targetClassInfo = getClassInfo(targetClass);
+        buildClassDef(targetClassInfo);
+        classInfo.classDef = ClassDef.replaceRootClassTo(this, targetClassInfo);
+        classInfo.serializer = createSerializer(targetClass);
+      }
+
+      return classInfo;
+    }
+
+    int header = buffer.readVarUint32Small14();
+    ClassInfo classInfo;
+    if ((header & 0b1) != 0) {
+      classInfo = readClassInfoFromBytes(buffer, classInfoCache, header);
+      classInfoCache = classInfo;
+    } else {
+      classInfo = getOrUpdateClassInfo((short) (header >> 1));
+    }
+    currentReadClass = classInfo.cls;
+    return classInfo;
+  }
+
   /**
    * Read class info from java data <code>buffer</code>. `classInfoCache` is used as a cache to
    * reduce map lookup to load class from binary.

java/fury-core/src/test/java/org/apache/fury/serializer/compatible/DifferentPOJOCompatibleSerializerTest.java

@@ -61,9 +61,9 @@ void testTargetHasLessFieldComparedToSourceClass() throws InterruptedException {
     ClassCompleteField<String> subclass = new ClassCompleteField<>("subclass", "subclass2");
     ClassCompleteField<ClassCompleteField<String>> classCompleteField =
         new ClassCompleteField<>(subclass, subclass);
-    byte[] serialized = getFury(ClassCompleteField.class).serializeJavaObject(classCompleteField);
+    byte[] serialized = getFury().serializeJavaObject(classCompleteField);
     ClassMissingField<ClassMissingField<String>> classMissingField =
-        getFury(ClassMissingField.class).deserializeJavaObject(serialized, ClassMissingField.class);
+        getFury().deserializeJavaObject(serialized, ClassMissingField.class);
 
     assertEq(classCompleteField, classMissingField);
   }
@@ -73,10 +73,10 @@ void testTargetHasMoreFieldComparedToSourceClass() throws InterruptedException {
 
     ClassMissingField<String> subclass = new ClassMissingField<>("subclass");
     ClassMissingField classMissingField = new ClassMissingField(subclass);
-    byte[] serialized = getFury(ClassMissingField.class).serializeJavaObject(classMissingField);
+    byte[] serialized = getFury().serializeJavaObject(classMissingField);
 
     ClassCompleteField classCompleteField =
-        getFury(ClassCompleteField.class)
+        getFury()
             .deserializeJavaObject(serialized, ClassCompleteField.class);
 
     assertEq(classCompleteField, classMissingField);