fix(java): fix nested map field value serialization by private map se…

…rializer (#1820)

## What does this PR do?

<!-- Describe the purpose of this PR. -->

## Related issues
Closes #1816

## Does this PR introduce any user-facing change?

<!--
If any user-facing interface changes, please [open an
issue](https://github.com/apache/fury/issues/new/choose) describing the
need to do so and update the document if necessary.
-->

- [ ] Does this PR introduce any public API change?
- [ ] Does this PR introduce any binary protocol compatibility change?

## Benchmark

<!--
When the PR has an impact on performance (if you don't know whether the
PR will have an impact on performance, you can submit the PR first, and
if it will have impact on performance, the code reviewer will explain
it), be sure to attach a benchmark data here.
-->

java/fury-core/src/main/java/org/apache/fury/builder/BaseObjectCodecBuilder.java

@@ -398,10 +398,18 @@ protected boolean useCollectionSerialization(TypeRef<?> typeRef) {
     return visitFury(f -> f.getClassResolver().isCollection(TypeUtils.getRawType(typeRef)));
   }
 
+  protected boolean useCollectionSerialization(Class<?> type) {
+    return visitFury(f -> f.getClassResolver().isCollection(TypeUtils.getRawType(type)));
+  }
+
   protected boolean useMapSerialization(TypeRef<?> typeRef) {
     return visitFury(f -> f.getClassResolver().isMap(TypeUtils.getRawType(typeRef)));
   }
 
+  protected boolean useMapSerialization(Class<?> type) {
+    return visitFury(f -> f.getClassResolver().isMap(TypeUtils.getRawType(type)));
+  }
+
   /**
    * Whether the provided type should be taken as final. Although the <code>clz</code> can be final,
    * the method can still return false. For example, we return false in meta share mode to write
@@ -491,6 +499,13 @@ protected Expression getOrCreateSerializer(Class<?> cls) {
           serializerClass = Serializer.class;
         }
       }
+      if (useCollectionSerialization(cls)
+          && !AbstractCollectionSerializer.class.isAssignableFrom(serializerClass)) {
+        serializerClass = AbstractCollectionSerializer.class;
+      } else if (useMapSerialization(cls)
+          && !AbstractMapSerializer.class.isAssignableFrom(serializerClass)) {
+        serializerClass = AbstractMapSerializer.class;
+      }
       TypeRef<? extends Serializer> serializerTypeRef = TypeRef.of(serializerClass);
       Expression fieldTypeExpr = getClassExpr(cls);
       // Don't invoke `Serializer.newSerializer` here, since it(ex. ObjectSerializer) may set itself

java/fury-core/src/main/java/org/apache/fury/collection/MapEntry.java

@@ -20,6 +20,7 @@
 package org.apache.fury.collection;
 
 import java.util.Map;
+import java.util.Objects;
 
 /** Map Entry implementation of {@link Map.Entry}. */
 public class MapEntry<K, V> implements Map.Entry<K, V> {
@@ -47,4 +48,21 @@ public V setValue(V value) {
     this.value = value;
     return o;
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    MapEntry<?, ?> mapEntry = (MapEntry<?, ?>) o;
+    return Objects.equals(key, mapEntry.key) && Objects.equals(value, mapEntry.value);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(key, value);
+  }
 }

java/fury-core/src/test/java/org/apache/fury/serializer/collection/MapSerializersTest.java

@@ -38,6 +38,7 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
@@ -503,4 +504,105 @@ public void testStringKeyMapSerializer(Fury fury) {
       copyCheck(fury, map);
     }
   }
+
+  // must be final class to test nested map value by private MapSerializer
+  private static final class PrivateMap<K, V> implements Map<K, V> {
+    private Set<Entry<K, V>> set = new HashSet<>();
+
+    @Override
+    public int size() {
+      return set.size();
+    }
+
+    @Override
+    public boolean isEmpty() {
+      return set.isEmpty();
+    }
+
+    @Override
+    public boolean containsKey(Object key) {
+      return set.stream().anyMatch(e -> e.getKey().equals(key));
+    }
+
+    @Override
+    public boolean containsValue(Object value) {
+      return set.stream().anyMatch(e -> e.getValue().equals(value));
+    }
+
+    @Override
+    public V get(Object key) {
+      for (Entry<K, V> kvEntry : set) {
+        if (kvEntry.getKey().equals(key)) {
+          return kvEntry.getValue();
+        }
+      }
+      return null;
+    }
+
+    @Override
+    public V put(K key, V value) {
+      set.add(new MapEntry<>(key, value));
+      return null;
+    }
+
+    @Override
+    public V remove(Object key) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void putAll(Map<? extends K, ? extends V> m) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void clear() {
+      set.clear();
+    }
+
+    @Override
+    public Set<K> keySet() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Collection<V> values() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Set<Entry<K, V>> entrySet() {
+      return set;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      PrivateMap<?, ?> that = (PrivateMap<?, ?>) o;
+      return Objects.equals(set, that.set);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(set);
+    }
+  }
+
+  @Data
+  @AllArgsConstructor
+  public static class LazyMapCollectionFieldStruct {
+    List<PrivateMap<String, Integer>> mapList;
+    PrivateMap<String, Integer> map;
+  }
+
+  @Test
+  public void testNestedValueByPrivateMapSerializer() {
+    Fury fury = builder().withRefTracking(false).build();
+    // test private map serializer
+    fury.registerSerializer(PrivateMap.class, new MapSerializer(fury, PrivateMap.class) {});
+    PrivateMap<String, Integer> map = new PrivateMap<>();
+    map.put("k", 1);
+    serDeCheck(fury, new LazyMapCollectionFieldStruct(ofArrayList(map), map));
+  }
 }