feat(java): Chunk by chunk predictive map serialization protocol

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

@@ -884,6 +884,15 @@ 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++;
@@ -1380,6 +1389,10 @@ public Class<? extends Serializer> getDefaultJDKStreamSerializerType() {
     return config.getDefaultJDKStreamSerializerType();
   }
 
+  public boolean isChunkSerializeMapEnabled() {
+    return config.isChunkSerializeMapEnabled();
+  }
+
   public boolean compressString() {
     return config.compressString();
   }

java/fury-core/src/main/java/org/apache/fury/config/Config.java

@@ -56,6 +56,7 @@ public class Config implements Serializable {
   private final boolean asyncCompilationEnabled;
   private final boolean deserializeNonexistentClass;
   private final boolean scalaOptimizationEnabled;
+  private final boolean chunkSerializeMapEnabled;
   private transient int configHash;
   private final boolean deserializeNonexistentEnumValueAsNull;
 
@@ -89,6 +90,7 @@ public Config(FuryBuilder builder) {
     asyncCompilationEnabled = builder.asyncCompilationEnabled;
     scalaOptimizationEnabled = builder.scalaOptimizationEnabled;
     deserializeNonexistentEnumValueAsNull = builder.deserializeNonexistentEnumValueAsNull;
+    chunkSerializeMapEnabled = builder.chunkSerializeMapEnabled;
   }
 
   public Language getLanguage() {
@@ -229,6 +231,10 @@ public boolean isAsyncCompilationEnabled() {
     return asyncCompilationEnabled;
   }
 
+  public boolean isChunkSerializeMapEnabled() {
+    return chunkSerializeMapEnabled;
+  }
+
   /** Whether enable scala-specific serialization optimization. */
   public boolean isScalaOptimizationEnabled() {
     return scalaOptimizationEnabled;

java/fury-core/src/main/java/org/apache/fury/config/FuryBuilder.java

@@ -79,6 +79,7 @@ public final class FuryBuilder {
   boolean scalaOptimizationEnabled = false;
   boolean suppressClassRegistrationWarnings = true;
   boolean deserializeNonexistentEnumValueAsNull = false;
+  boolean chunkSerializeMapEnabled = false;
   MetaCompressor metaCompressor = new DeflaterMetaCompressor();
 
   public FuryBuilder() {}
@@ -304,6 +305,17 @@ public FuryBuilder withScalaOptimizationEnabled(boolean enableScalaOptimization)
     return this;
   }
 
+  /**
+   * use chunk by chunk method to serialize map, TODO: generate code for chunk by chunk method
+   *
+   * @param chunkSerializeMapEnabled
+   * @return
+   */
+  public FuryBuilder withChunkSerializeMapEnable(boolean chunkSerializeMapEnabled) {
+    this.chunkSerializeMapEnabled = chunkSerializeMapEnabled;
+    return this;
+  }
+
   private void finish() {
     if (classLoader == null) {
       classLoader = Thread.currentThread().getContextClassLoader();