fix(java): Compatible mode on de/serialize api failed to deserialize

docs/guide/java_serialization_guide.md

@@ -102,7 +102,7 @@ public class Example {
 | `compressLong`                      | Enables or disables long compression for smaller size.                                                                                                                                                                                                                                                                                                                                                                                                                                                                            | `true`                                                         |
 | `compressString`                    | Enables or disables string compression for smaller size.                                                                                                                                                                                                                                                                                                                                                                                                                                                                          | `false`                                                        |
 | `classLoader`                       | The classloader should not be updated; Fury caches class metadata. Use `LoaderBinding` or `ThreadSafeFury` for classloader updates.                                                                                                                                                                                                                                                                                                                                                                                               | `Thread.currentThread().getContextClassLoader()`               |
-| `compatibleMode`                    | Type forward/backward compatibility config. Also Related to `checkClassVersion` config. `SCHEMA_CONSISTENT`: Class schema must be consistent between serialization peer and deserialization peer. `COMPATIBLE`: Class schema can be different between serialization peer and deserialization peer. They can add/delete fields independently.                                                                                                                                                                                      | `CompatibleMode.SCHEMA_CONSISTENT`                             |
+| `compatibleMode`                    | Type forward/backward compatibility config. Also Related to `checkClassVersion` config. `SCHEMA_CONSISTENT`: Class schema must be consistent between serialization peer and deserialization peer. `COMPATIBLE`: Class schema can be different between serialization peer and deserialization peer. They can add/delete fields independently. [See more](#class-inconsistency-and-class-version-check).                                                                                                                             | `CompatibleMode.SCHEMA_CONSISTENT`                             |
 | `checkClassVersion`                 | Determines whether to check the consistency of the class schema. If enabled, Fury checks, writes, and checks consistency using the `classVersionHash`. It will be automatically disabled when `CompatibleMode#COMPATIBLE` is enabled. Disabling is not recommended unless you can ensure the class won't evolve.                                                                                                                                                                                                                  | `false`                                                        |
 | `checkJdkClassSerializable`         | Enables or disables checking of `Serializable` interface for classes under `java.*`. If a class under `java.*` is not `Serializable`, Fury will throw an `UnsupportedOperationException`.                                                                                                                                                                                                                                                                                                                                         | `true`                                                         |
 | `registerGuavaTypes`                | Whether to pre-register Guava types such as `RegularImmutableMap`/`RegularImmutableList`. These types are not public API, but seem pretty stable.                                                                                                                                                                                                                                                                                                                                                                                 | `true`                                                         |
@@ -518,6 +518,12 @@ fury with
 `CompatibleMode.COMPATIBLE` has more performance and space cost, do not set it by default if your classes are always
 consistent between serialization and deserialization.
 
+### Different POJO deserialization
+Fury allows you to serialize one POJO and deserialize it into a different POJO. To achieve this, configure Fury with
+`CompatibleMode` set to `org.apache.fury.config.CompatibleMode.COMPATIBLE`. Additionally, you only need to register the
+specific classes you want to serialize or deserialize; there's no need to register any nested classes within them.
+[See example here](/java/fury-core/src/test/java/org/apache/fury/serializer/compatible/DifferentPOJOCompatibleSerializerTest.java)
+
 ### Use wrong API for deserialization
 
 If you serialize an object by invoking `Fury#serialize`, you should invoke `Fury#deserialize` for deserialization

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

@@ -1070,6 +1070,7 @@ public void serializeJavaObject(MemoryBuffer buffer, Object obj) {
         buffer.writeInt32(-1); // preserve 4-byte for meta start offsets.
         if (!refResolver.writeRefOrNull(buffer, obj)) {
           ClassInfo classInfo = classResolver.getOrUpdateClassInfo(obj.getClass());
+          classResolver.writeClass(buffer, classInfo);
           writeData(buffer, classInfo, obj);
           MetaContext metaContext = serializationContext.getMetaContext();
           if (metaContext != null && !metaContext.writingClassDefs.isEmpty()) {
@@ -1119,7 +1120,13 @@ public <T> T deserializeJavaObject(MemoryBuffer buffer, Class<T> cls) {
       T obj;
       int nextReadRefId = refResolver.tryPreserveRefId(buffer);
       if (nextReadRefId >= NOT_NULL_VALUE_FLAG) {
-        obj = (T) readDataInternal(buffer, classResolver.getClassInfo(cls));
+        ClassInfo classInfo;
+        if (shareMeta) {
+          classInfo = classResolver.readClassInfo(buffer);
+        } else {
+          classInfo = classResolver.getClassInfo(cls);
+        }
+        obj = (T) readDataInternal(buffer, classInfo);
         return obj;
       } else {
         return null;

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

@@ -140,8 +140,10 @@ public void testEnumSerializationAsString() {
             .build();
 
     // serialize enum "B"
+    furySerialization.register(cls1);
     byte[] bytes = furySerialization.serializeJavaObject(cls1.getEnumConstants()[1]);
 
+    furyDeserialize.register(cls2);
     Object data = furyDeserialize.deserializeJavaObject(bytes, cls2);
     assertEquals(cls2.getEnumConstants()[0], data);
   }
@@ -175,8 +177,10 @@ public void testEnumSerializationAsString_differentClass() {
             .build();
 
     // serialize enum "B"
+    furySerialization.register(cls1);
     byte[] bytes = furySerialization.serializeJavaObject(cls1.getEnumConstants()[1]);
 
+    furyDeserialize.register(cls2);
     Object data = furyDeserialize.deserializeJavaObject(bytes, cls2);
     assertEquals(cls2.getEnumConstants()[0], data);
   }
@@ -209,9 +213,11 @@ public void testEnumSerializationAsString_invalidEnum() {
             .withAsyncCompilation(false)
             .build();
 
+    furySerialization.register(cls1);
     byte[] bytes = furySerialization.serializeJavaObject(cls1.getEnumConstants()[0]);
 
     try {
+      furyDeserialize.register(cls2);
       furyDeserialize.deserializeJavaObject(bytes, cls2);
       fail("expected to throw exception");
     } catch (Exception e) {

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

@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.fury.serializer.compatible;
+
+import org.apache.fury.Fury;
+import org.apache.fury.config.CompatibleMode;
+import org.apache.fury.config.Language;
+import org.apache.fury.serializer.compatible.classes.ClassCompleteField;
+import org.apache.fury.serializer.compatible.classes.ClassMissingField;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+/**
+ * Test COMPATIBILITY mode that supports - same field type and name can be deserialized to other
+ * class with different name - scrambled field order to make sure it could handle different field
+ * order - missing or extra field from source class to target class - generic class
+ */
+public class DifferentPOJOCompatibleSerializerTest extends Assert {
+
+  Fury getFury(Class<?>... classes) {
+    Fury instance =
+        Fury.builder()
+            .withLanguage(Language.JAVA)
+            .withRefTracking(true)
+            .withCompatibleMode(CompatibleMode.COMPATIBLE)
+            .withMetaShare(true)
+            .withScopedMetaShare(true)
+            .requireClassRegistration(false)
+            .withAsyncCompilation(true)
+            .serializeEnumByName(true)
+            .build();
+    if (classes != null) {
+      for (Class<?> clazz : classes) {
+        instance.register(clazz);
+      }
+    }
+    ;
+    return instance;
+  }
+
+  @Test
+  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);
+    ClassMissingField<ClassMissingField<String>> classMissingField =
+        getFury(ClassMissingField.class).deserializeJavaObject(serialized, ClassMissingField.class);
+
+    assertEq(classCompleteField, classMissingField);
+  }
+
+  @Test
+  void testTargetHasMoreFieldComparedToSourceClass() throws InterruptedException {
+
+    ClassMissingField<String> subclass = new ClassMissingField<>("subclass");
+    ClassMissingField classMissingField = new ClassMissingField(subclass);
+    byte[] serialized = getFury(ClassMissingField.class).serializeJavaObject(classMissingField);
+
+    ClassCompleteField classCompleteField =
+        getFury(ClassCompleteField.class)
+            .deserializeJavaObject(serialized, ClassCompleteField.class);
+
+    assertEq(classCompleteField, classMissingField);
+  }
+
+  void assertEq(ClassCompleteField classCompleteField, ClassMissingField classMissingField) {
+    assertEqSubClass(
+        (ClassCompleteField) classCompleteField.getPrivateFieldSubClass(),
+        (ClassMissingField) classMissingField.getPrivateFieldSubClass());
+    assertEquals(classCompleteField.getPrivateMap(), classMissingField.getPrivateMap());
+    assertEquals(classCompleteField.getPrivateList(), classMissingField.getPrivateList());
+    assertEquals(classCompleteField.getPrivateString(), classMissingField.getPrivateString());
+    assertEquals(classCompleteField.getPrivateInt(), classMissingField.getPrivateInt());
+  }
+
+  void assertEqSubClass(
+      ClassCompleteField classCompleteField, ClassMissingField classMissingField) {
+    assertEquals(
+        classCompleteField.getPrivateFieldSubClass(), classMissingField.getPrivateFieldSubClass());
+    assertEquals(classCompleteField.getPrivateMap(), classMissingField.getPrivateMap());
+    assertEquals(classCompleteField.getPrivateList(), classMissingField.getPrivateList());
+    assertEquals(classCompleteField.getPrivateString(), classMissingField.getPrivateString());
+    assertEquals(classCompleteField.getPrivateInt(), classMissingField.getPrivateInt());
+  }
+}

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

@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.fury.serializer.compatible.classes;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+
+@EqualsAndHashCode
+@Getter
+public class ClassCompleteField<T> {
+  private boolean privateBoolean = true;
+  private int privateInt = 10;
+  private String privateString = "notNull";
+  private Map<String, String> privateMap;
+  private List<String> privateList;
+  private T privateFieldSubClass;
+
+  private boolean privateBoolean2 = true;
+  private int privateInt2 = 10;
+  private String privateString2 = "notNull";
+  private Map<String, String> privateMap2;
+  private List<String> privateList2;
+  private T privateFieldSubClass2;
+
+  public ClassCompleteField(T privateFieldSubClass, T privateFieldSubClass2) {
+    privateMap = new HashMap<>();
+    privateMap.put("a", "b");
+    privateList = new ArrayList<>();
+    privateList.add("a");
+
+    privateMap2 = new HashMap<>();
+    privateMap2.put("a", "b");
+    privateList2 = new ArrayList<>();
+    privateList2.add("a");
+
+    this.privateFieldSubClass = privateFieldSubClass;
+    this.privateFieldSubClass2 = privateFieldSubClass2;
+  }
+}

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

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.fury.serializer.compatible.classes;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+
+@EqualsAndHashCode
+@Getter
+public class ClassMissingField<T> {
+  private T privateFieldSubClass;
+  private List<String> privateList;
+  private Map<String, String> privateMap;
+  private String privateString = "missing";
+  private int privateInt = 999;
+  private boolean privateBoolean = false;
+
+  public ClassMissingField(T privateFieldSubClass) {
+    privateMap = new HashMap<>();
+    privateMap.put("z", "x");
+    privateList = new ArrayList<>();
+    privateList.add("c");
+    this.privateFieldSubClass = privateFieldSubClass;
+  }
+}

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

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.fury.serializer.compatible.classes;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+
+@Getter
+@EqualsAndHashCode
+public class SubclassCompleteField {
+  private boolean privateBoolean = true;
+  private int privateInt = 10;
+  private String privateString = "notNull";
+  private Map<String, String> privateMap;
+  private List<String> privateList;
+
+  private boolean privateBoolean2 = true;
+  private int privateInt2 = 10;
+  private String privateString2 = "notNull";
+  private Map<String, String> privateMap2;
+  private List<String> privateList2;
+
+  public SubclassCompleteField() {
+    privateMap = new HashMap<>();
+    privateMap.put("a", "b");
+    privateList = new ArrayList<>();
+    privateList.add("a");
+
+    privateMap2 = new HashMap<>();
+    privateMap2.put("a", "b");
+    privateList2 = new ArrayList<>();
+    privateList2.add("a");
+  }
+}

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

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.fury.serializer.compatible.classes;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+
+@Getter
+@EqualsAndHashCode
+public class SubclassMissingField {
+  private boolean privateBoolean = true;
+  private int privateInt = 10;
+  private String privateString = "notNull";
+  private Map<String, String> privateMap;
+  private List<String> privateList;
+
+  public SubclassMissingField() {
+    privateMap = new HashMap<>();
+    privateMap.put("a", "b");
+    privateList = new ArrayList<>();
+    privateList.add("a");
+  }
+}