feat(java): Type annotation hints for serialization(WIP)

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

@@ -697,6 +697,25 @@ public void writeJavaStringRef(MemoryBuffer buffer, String str) {
     }
   }
 
+  public void writeJavaStringRef(MemoryBuffer buffer, String str, boolean refTracking, boolean nullable) {
+      if (refTracking) {
+        if (!refResolver.writeRefOrNull(buffer, str)) {
+          stringSerializer.writeJavaString(buffer, str);
+        }
+      } else {
+          if (nullable) {
+              if (str == null) {
+                  buffer.writeByte(Fury.NULL_FLAG);
+              } else {
+                  buffer.writeByte(Fury.NOT_NULL_VALUE_FLAG);
+                  stringSerializer.write(buffer, str);
+              }
+          } else {
+              stringSerializer.write(buffer, str);
+          }
+      }
+  }
+
   public String readJavaStringRef(MemoryBuffer buffer) {
     RefResolver refResolver = this.refResolver;
     if (stringSerializer.needToWriteRef()) {
@@ -935,15 +954,6 @@ public Object readNullable(MemoryBuffer buffer) {
     }
   }
 
-  public Object readNullable(MemoryBuffer buffer, ClassInfoHolder classInfoHolder) {
-    byte headFlag = buffer.readByte();
-    if (headFlag == Fury.NULL_FLAG) {
-      return null;
-    } else {
-      return readNonRef(buffer, classInfoHolder);
-    }
-  }
-
   /** Class should be read already. */
   public Object readData(MemoryBuffer buffer, ClassInfo classInfo) {
     depth++;

java/fury-core/src/main/java/org/apache/fury/annotation/FieldInfo.java

@@ -0,0 +1,21 @@
+package org.apache.fury.annotation;
+
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.FIELD)
+public @interface FieldInfo {
+
+    /**
+     * Whether to track reference.
+     */
+    boolean trackingRef() default false;
+    /**
+     * Whether field is nullable.
+     */
+    boolean nullable() default true;
+}

java/fury-core/src/main/java/org/apache/fury/serializer/AbstractObjectSerializer.java

@@ -22,6 +22,7 @@
 import static org.apache.fury.type.DescriptorGrouper.createDescriptorGrouper;
 import static org.apache.fury.type.TypeUtils.getRawType;
 
+import java.lang.annotation.Annotation;
 import java.lang.invoke.MethodHandle;
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
@@ -31,6 +32,7 @@
 import java.util.List;
 import java.util.stream.Collectors;
 import org.apache.fury.Fury;
+import org.apache.fury.annotation.FieldInfo;
 import org.apache.fury.collection.Tuple2;
 import org.apache.fury.collection.Tuple3;
 import org.apache.fury.memory.Platform;
@@ -376,7 +378,7 @@ protected T newBean() {
               descriptor.getField() != null
                   ? FieldAccessor.createAccessor(descriptor.getField())
                   : null,
-              fury);
+              fury, descriptor.getField() != null ? descriptor.getField().getAnnotation(FieldInfo.class) : null);
       otherFields[cnt++] = genericTypeField;
     }
     cnt = 0;
@@ -398,27 +400,29 @@ private static FinalTypeField buildFinalTypeField(Fury fury, Descriptor d) {
         d.getDeclaringClass() + "." + d.getName(),
         // `d.getField()` will be null when peer class doesn't have this field.
         d.getField() != null ? FieldAccessor.createAccessor(d.getField()) : null,
-        fury);
+        fury, d.getField() != null ? d.getField().getAnnotation(FieldInfo.class) : null);
   }
 
   private static GenericTypeField buildContainerField(Fury fury, Descriptor d) {
     return new GenericTypeField(
         d.getTypeRef(),
         d.getDeclaringClass() + "." + d.getName(),
         d.getField() != null ? FieldAccessor.createAccessor(d.getField()) : null,
-        fury);
+        fury, d.getField() != null ? d.getField().getAnnotation(FieldInfo.class) : null);
   }
 
   public static class InternalFieldInfo {
     protected final short classId;
     protected final String qualifiedFieldName;
     protected final FieldAccessor fieldAccessor;
+    protected FieldInfo fieldInfo;
 
     private InternalFieldInfo(
-        short classId, String qualifiedFieldName, FieldAccessor fieldAccessor) {
+        short classId, String qualifiedFieldName, FieldAccessor fieldAccessor, FieldInfo fieldInfo) {
       this.classId = classId;
       this.qualifiedFieldName = qualifiedFieldName;
       this.fieldAccessor = fieldAccessor;
+      this.fieldInfo = fieldInfo;
     }
 
     @Override
@@ -430,15 +434,17 @@ public String toString() {
           + qualifiedFieldName
           + ", field="
           + (fieldAccessor != null ? fieldAccessor.getField() : null)
+          + ", fieldInfo="
+          + fieldInfo
           + '}';
     }
   }
 
   static final class FinalTypeField extends InternalFieldInfo {
     final ClassInfo classInfo;
 
-    private FinalTypeField(Class<?> type, String fieldName, FieldAccessor accessor, Fury fury) {
-      super(getRegisteredClassId(fury, type), fieldName, accessor);
+    private FinalTypeField(Class<?> type, String fieldName, FieldAccessor accessor, Fury fury, FieldInfo fieldInfo) {
+      super(getRegisteredClassId(fury, type), fieldName, accessor, fieldInfo);
       // invoke `copy` to avoid ObjectSerializer construct clear serializer by `clearSerializer`.
       if (type == FinalObjectTypeStub.class) {
         // `FinalObjectTypeStub` has no fields, using its `classInfo`
@@ -456,17 +462,17 @@ static final class GenericTypeField extends InternalFieldInfo {
     final boolean trackingRef;
 
     private GenericTypeField(
-        Class<?> cls, String qualifiedFieldName, FieldAccessor accessor, Fury fury) {
-      super(getRegisteredClassId(fury, cls), qualifiedFieldName, accessor);
+        Class<?> cls, String qualifiedFieldName, FieldAccessor accessor, Fury fury, FieldInfo fieldInfo) {
+      super(getRegisteredClassId(fury, cls), qualifiedFieldName, accessor, fieldInfo);
       // TODO support generics <T> in Pojo<T>, see ComplexObjectSerializer.getGenericTypes
       genericType = fury.getClassResolver().buildGenericType(cls);
       classInfoHolder = fury.getClassResolver().nilClassInfoHolder();
       trackingRef = fury.getClassResolver().needToWriteRef(cls);
     }
 
     private GenericTypeField(
-        TypeRef<?> typeRef, String qualifiedFieldName, FieldAccessor accessor, Fury fury) {
-      super(getRegisteredClassId(fury, getRawType(typeRef)), qualifiedFieldName, accessor);
+        TypeRef<?> typeRef, String qualifiedFieldName, FieldAccessor accessor, Fury fury, FieldInfo fieldInfo) {
+      super(getRegisteredClassId(fury, getRawType(typeRef)), qualifiedFieldName, accessor, fieldInfo);
       // TODO support generics <T> in Pojo<T>, see ComplexObjectSerializer.getGenericTypes
       genericType = fury.getClassResolver().buildGenericType(typeRef);
       classInfoHolder = fury.getClassResolver().nilClassInfoHolder();
@@ -484,6 +490,8 @@ public String toString() {
           + qualifiedFieldName
           + ", field="
           + (fieldAccessor != null ? fieldAccessor.getField() : null)
+          + ", fieldInfo="
+          + fieldInfo
           + '}';
     }
   }

java/fury-core/src/main/java/org/apache/fury/serializer/CompatibleSerializer.java

@@ -134,7 +134,7 @@ private void readAndWriteFieldValue(
         fury, buffer, targetObject, fieldAccessor, classId)) {
       Object fieldValue;
       fieldValue = fieldAccessor.getObject(targetObject);
-      if (ObjectSerializer.writeBasicObjectFieldValueFailed(fury, buffer, fieldValue, classId)) {
+      if (ObjectSerializer.writeBasicObjectFieldValueFailed(fury, buffer, fieldValue, classId, false, true)) {
         if (classId == ClassResolver.NO_CLASS_ID) { // SEPARATE_TYPES_HASH
           writeSeparateFieldValue(fieldInfo, buffer, fieldValue);
         } else {