fix(java): raise exception when registering invalid serializer for Map/List (#2291)

## What does this PR do?
raise an exception with error message when when registering serializer
for Map/List.

## Related issues

- #2251
- #2270

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

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

## Benchmark

Author: OmCheeLin

java/fory-core/src/main/java/org/apache/fory/resolver/ClassResolver.java

@@ -135,6 +135,7 @@
 import org.apache.fory.serializer.OptionalSerializers;
 import org.apache.fory.serializer.PrimitiveSerializers;
 import org.apache.fory.serializer.ReplaceResolveSerializer;
+import org.apache.fory.serializer.SerializationUtils;
 import org.apache.fory.serializer.Serializer;
 import org.apache.fory.serializer.SerializerFactory;
 import org.apache.fory.serializer.Serializers;
@@ -737,6 +738,9 @@ public <T> void registerSerializer(Class<T> type, Class<? extends Serializer> se
    * @param serializer serializer for object of {@code type}
    */
   public void registerSerializer(Class<?> type, Serializer<?> serializer) {
+    if (!serializer.getClass().getPackage().getName().startsWith("org.apache.fory")) {
+      SerializationUtils.validate(type, serializer.getClass());
+    }
     if (!extRegistry.registeredClassIdMap.containsKey(type) && !fory.isCrossLanguage()) {
       register(type);
     }

java/fory-core/src/main/java/org/apache/fory/resolver/XtypeResolver.java

@@ -71,6 +71,7 @@
 import org.apache.fory.serializer.NonexistentClass;
 import org.apache.fory.serializer.NonexistentClassSerializers;
 import org.apache.fory.serializer.ObjectSerializer;
+import org.apache.fory.serializer.SerializationUtils;
 import org.apache.fory.serializer.Serializer;
 import org.apache.fory.serializer.Serializers;
 import org.apache.fory.serializer.collection.AbstractCollectionSerializer;
@@ -272,11 +273,17 @@ private ClassInfo newClassInfo(
 
   public <T> void registerSerializer(Class<T> type, Class<? extends Serializer> serializerClass) {
     ClassInfo classInfo = checkClassRegistration(type);
+    if (!serializerClass.getPackage().getName().startsWith("org.apache.fory")) {
+      SerializationUtils.validate(type, serializerClass);
+    }
     classInfo.serializer = Serializers.newSerializer(fory, type, serializerClass);
   }
 
   public void registerSerializer(Class<?> type, Serializer<?> serializer) {
     ClassInfo classInfo = checkClassRegistration(type);
+    if (!serializer.getClass().getPackage().getName().startsWith("org.apache.fory")) {
+      SerializationUtils.validate(type, serializer.getClass());
+    }
     classInfo.serializer = serializer;
   }
 

java/fory-core/src/main/java/org/apache/fory/serializer/SerializationUtils.java

@@ -19,18 +19,47 @@
 
 package org.apache.fory.serializer;
 
+import java.util.Collection;
+import java.util.Map;
 import org.apache.fory.Fory;
 import org.apache.fory.annotation.Internal;
 import org.apache.fory.resolver.ClassInfo;
 import org.apache.fory.resolver.TypeResolver;
+import org.apache.fory.serializer.collection.AbstractCollectionSerializer;
+import org.apache.fory.serializer.collection.AbstractMapSerializer;
 
 @Internal
-class SerializationUtils {
+public class SerializationUtils {
   public static TypeResolver getTypeResolver(Fory fory) {
     return fory.isCrossLanguage() ? fory.getXtypeResolver() : fory.getClassResolver();
   }
 
   public static ClassInfo getClassInfo(Fory fory, Class<?> cls) {
     return getTypeResolver(fory).getClassInfo(cls);
   }
+
+  public static void validateSerializer(
+      Class<?> type,
+      Class<? extends Serializer> serializerClass,
+      Class<?> parentType,
+      Class<?> requiredSerializerBase) {
+    if (!parentType.isAssignableFrom(type)) {
+      return;
+    }
+    boolean valid = requiredSerializerBase.isAssignableFrom(serializerClass);
+    if (!valid) {
+      throw new IllegalArgumentException(
+          "Serializer for type "
+              + type.getName()
+              + " must extend "
+              + requiredSerializerBase.getSimpleName()
+              + ", but got "
+              + serializerClass.getName());
+    }
+  }
+
+  public static void validate(Class<?> type, Class<? extends Serializer> serializerClass) {
+    validateSerializer(type, serializerClass, Collection.class, AbstractCollectionSerializer.class);
+    validateSerializer(type, serializerClass, Map.class, AbstractMapSerializer.class);
+  }
 }

java/fory-core/src/test/java/javax/fory/test/ResolverValidateSerializerTest.java

@@ -0,0 +1,195 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package javax.fory.test;
+
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+import java.util.AbstractList;
+import java.util.AbstractMap;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import org.apache.fory.Fory;
+import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.serializer.Serializer;
+import org.apache.fory.serializer.collection.AbstractCollectionSerializer;
+import org.apache.fory.serializer.collection.AbstractMapSerializer;
+import org.testng.annotations.Test;
+
+public class ResolverValidateSerializerTest {
+  static final class InvalidList extends AbstractList<Object> {
+    @Override
+    public Object get(int index) {
+      throw new IndexOutOfBoundsException();
+    }
+
+    @Override
+    public int size() {
+      return 0;
+    }
+
+    public static final class InvalidListSerializer extends Serializer<InvalidList> {
+      public InvalidListSerializer(Fory fory) {
+        super(fory, InvalidList.class);
+      }
+
+      @Override
+      public void write(MemoryBuffer buffer, InvalidList value) {
+        // no-op
+      }
+
+      @Override
+      public InvalidList read(MemoryBuffer buffer) {
+        return new InvalidList();
+      }
+    }
+  }
+
+  static final class ValidList extends AbstractList<Object> {
+    @Override
+    public Object get(int index) {
+      throw new IndexOutOfBoundsException();
+    }
+
+    @Override
+    public int size() {
+      return 0;
+    }
+
+    public static final class ValidListSerializer extends AbstractCollectionSerializer<ValidList> {
+      public ValidListSerializer(Fory fory) {
+        super(fory, ValidList.class);
+      }
+
+      @Override
+      public Collection<?> onCollectionWrite(MemoryBuffer buffer, ValidList value) {
+        return Collections.emptyList();
+      }
+
+      @Override
+      public ValidList read(MemoryBuffer buffer) {
+        return onCollectionRead(Collections.emptyList());
+      }
+
+      @Override
+      public ValidList onCollectionRead(Collection collection) {
+        return new ValidList();
+      }
+    }
+  }
+
+  static final class InvalidMap extends AbstractMap<Object, Object> {
+    @Override
+    public Set<Entry<Object, Object>> entrySet() {
+      return Collections.emptySet();
+    }
+
+    public static final class InvalidMapSerializer extends Serializer<InvalidMap> {
+      public InvalidMapSerializer(Fory fory) {
+        super(fory, InvalidMap.class);
+      }
+
+      @Override
+      public void write(MemoryBuffer buffer, InvalidMap value) {
+        // no-op
+      }
+
+      @Override
+      public InvalidMap read(MemoryBuffer buffer) {
+        return new InvalidMap();
+      }
+    }
+  }
+
+  static final class ValidMap extends AbstractMap<Object, Object> {
+    @Override
+    public Set<Entry<Object, Object>> entrySet() {
+      return Collections.emptySet();
+    }
+
+    public static final class ValidMapSerializer extends AbstractMapSerializer<ValidMap> {
+      public ValidMapSerializer(Fory fory) {
+        super(fory, ValidMap.class);
+      }
+
+      @Override
+      public Map<?, ?> onMapWrite(MemoryBuffer buffer, ValidMap value) {
+        return Collections.emptyMap();
+      }
+
+      @Override
+      public ValidMap onMapCopy(Map map) {
+        return new ValidMap();
+      }
+
+      @Override
+      public ValidMap read(MemoryBuffer buffer) {
+        return onMapRead(Collections.emptyMap());
+      }
+
+      @Override
+      public ValidMap onMapRead(Map map) {
+        return new ValidMap();
+      }
+    }
+  }
+
+  @Test
+  public void testListAndMapSerializerRegistration() {
+    Fory fory = Fory.builder().withRefTracking(true).requireClassRegistration(false).build();
+    // List invalid
+    assertThrows(
+        IllegalArgumentException.class,
+        () -> fory.registerSerializer(InvalidList.class, InvalidList.InvalidListSerializer.class));
+    assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            fory.registerSerializer(
+                InvalidList.class, new InvalidList.InvalidListSerializer(fory)));
+    assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            fory.registerSerializer(
+                InvalidList.class, f -> new InvalidList.InvalidListSerializer(f)));
+    // List valid
+    fory.register(ValidList.class);
+    fory.registerSerializer(ValidList.class, new ValidList.ValidListSerializer(fory));
+    Object listResult = fory.deserialize(fory.serialize(new ValidList()));
+    assertTrue(listResult instanceof ValidList);
+    // Map invalid
+    assertThrows(
+        IllegalArgumentException.class,
+        () -> fory.registerSerializer(InvalidMap.class, InvalidMap.InvalidMapSerializer.class));
+    assertThrows(
+        IllegalArgumentException.class,
+        () -> fory.registerSerializer(InvalidMap.class, new InvalidMap.InvalidMapSerializer(fory)));
+    assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            fory.registerSerializer(InvalidMap.class, f -> new InvalidMap.InvalidMapSerializer(f)));
+    // Map valid
+    fory.register(ValidMap.class);
+    fory.registerSerializer(ValidMap.class, new ValidMap.ValidMapSerializer(fory));
+    Object mapResult = fory.deserialize(fory.serialize(new ValidMap()));
+    assertTrue(mapResult instanceof ValidMap);
+  }
+}