[Java][2/N] support type forward/backward compatibility (#195)

* support Compatible Serialization

* add doc

java/fury-core/src/main/java/io/fury/Config.java

@@ -36,6 +36,7 @@
 
 package io.fury;
 
+import io.fury.serializer.CompatibleMode;
 import io.fury.serializer.Serializer;
 import io.fury.serializer.TimeSerializers;
 import io.fury.util.MurmurHash3;
@@ -58,6 +59,7 @@ public class Config implements Serializable {
   private final boolean compressNumber;
   private final boolean compressString;
   private final boolean checkClassVersion;
+  private final CompatibleMode compatibleMode;
   private final Class<? extends Serializer> defaultJDKStreamSerializerType;
   private final boolean secureModeEnabled;
   private final boolean classRegistrationRequired;
@@ -73,6 +75,7 @@ public class Config implements Serializable {
     compressNumber = builder.compressNumber;
     compressString = builder.compressString;
     checkClassVersion = builder.checkClassVersion;
+    compatibleMode = builder.compatibleMode;
     defaultJDKStreamSerializerType = builder.defaultJDKStreamSerializerType;
     secureModeEnabled = builder.secureModeEnabled;
     classRegistrationRequired = builder.requireClassRegistration;
@@ -125,6 +128,10 @@ public boolean checkClassVersion() {
     return checkClassVersion;
   }
 
+  public CompatibleMode getCompatibleMode() {
+    return compatibleMode;
+  }
+
   /**
    * Returns default serializer type for class which implements jdk serialization method such as
    * `writeObject/readObject`.

java/fury-core/src/main/java/io/fury/Fury.java

@@ -32,6 +32,7 @@
 import io.fury.serializer.ArraySerializers;
 import io.fury.serializer.BufferCallback;
 import io.fury.serializer.BufferObject;
+import io.fury.serializer.CompatibleMode;
 import io.fury.serializer.JavaSerializer;
 import io.fury.serializer.OpaqueObjects;
 import io.fury.serializer.Serializer;
@@ -1035,6 +1036,7 @@ public static final class FuryBuilder {
     ClassLoader classLoader;
     boolean compressNumber = false;
     boolean compressString = true;
+    CompatibleMode compatibleMode = CompatibleMode.SCHEMA_CONSISTENT;
     // TODO(chaokunyang) switch to object stream serializer.
     Class<? extends Serializer> defaultJDKStreamSerializerType = JavaSerializer.class;
     boolean secureModeEnabled = true;
@@ -1079,6 +1081,11 @@ public FuryBuilder withClassLoader(ClassLoader classLoader) {
       return this;
     }
 
+    public FuryBuilder withCompatibleMode(CompatibleMode compatibleMode) {
+      this.compatibleMode = compatibleMode;
+      return this;
+    }
+
     public FuryBuilder requireClassRegistration(boolean requireClassRegistration) {
       this.requireClassRegistration = requireClassRegistration;
       return this;

java/fury-core/src/main/java/io/fury/resolver/ClassResolver.java

@@ -35,6 +35,7 @@
 import io.fury.serializer.ArraySerializers;
 import io.fury.serializer.BufferSerializers;
 import io.fury.serializer.CollectionSerializers;
+import io.fury.serializer.CompatibleSerializer;
 import io.fury.serializer.ExternalizableSerializer;
 import io.fury.serializer.JavaSerializer;
 import io.fury.serializer.JdkProxySerializer;
@@ -636,12 +637,24 @@ public Class<? extends Serializer> getSerializerClass(Class<?> cls) {
       if (requireJavaSerialization(cls)) {
         return getJavaSerializer(cls);
       }
-      return ObjectSerializer.class;
+      return getObjectSerializerClass(cls);
     }
   }
 
   public Class<? extends Serializer> getObjectSerializerClass(Class<?> cls) {
-    return ObjectSerializer.class;
+    if (fury.getLanguage() != Language.JAVA) {
+      LOG.warn("Class {} isn't supported for cross-language serialization.", cls);
+    }
+    LOG.debug("Object of type {} can't be serialized by jit", cls);
+    switch (fury.getConfig().getCompatibleMode()) {
+      case SCHEMA_CONSISTENT:
+        return ObjectSerializer.class;
+      case COMPATIBLE:
+        return CompatibleSerializer.class;
+      default:
+        throw new UnsupportedOperationException(
+            String.format("Unsupported mode %s", fury.getConfig().getCompatibleMode()));
+    }
   }
 
   public Class<? extends Serializer> getJavaSerializer(Class<?> clz) {

java/fury-core/src/main/java/io/fury/resolver/FieldResolver.java

@@ -100,6 +100,7 @@
  * info is less than current field, then it will be a field not exists in current class and can be
  * skipped.
  *
+ * @see io.fury.serializer.CompatibleSerializerBase
  * @author chaokunyang
  */
 @SuppressWarnings({"rawtypes", "UnstableApiUsage"})