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,9 +56,11 @@ 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;
 
+
   public Config(FuryBuilder builder) {
     language = builder.language;
     trackingRef = builder.trackingRef;
@@ -89,6 +91,7 @@ public Config(FuryBuilder builder) {
     asyncCompilationEnabled = builder.asyncCompilationEnabled;
     scalaOptimizationEnabled = builder.scalaOptimizationEnabled;
     deserializeNonexistentEnumValueAsNull = builder.deserializeNonexistentEnumValueAsNull;
+    chunkSerializeMapEnabled = builder.chunkSerializeMapEnabled;
   }
 
   public Language getLanguage() {
@@ -229,6 +232,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,16 @@ 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();