feat(java): Support copy capabilities for some classes without no-arg…

…ument constructors (#1794)

## What does this PR do?
Some classes with no-argument constructors will report an error when
calling `copy()`.

This pr:
- implement the copy method for the no-argument constructor serializer
- add test case


## Related issues
#1777
#1679

## 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.
-->

---------

Co-authored-by: Shawn Yang <[email protected]>

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

@@ -44,6 +44,14 @@ public void write(MemoryBuffer buffer, ByteBuffer value) {
       fury.writeBufferObject(buffer, new BufferObject.ByteBufferBufferObject(value));
     }
 
+    @Override
+    public ByteBuffer copy(ByteBuffer value) {
+      ByteBuffer dst = ByteBuffer.allocate(value.remaining());
+      dst.put(value.duplicate());
+      dst.rewind();
+      return dst;
+    }
+
     @Override
     public ByteBuffer read(MemoryBuffer buffer) {
       MemoryBuffer newBuffer = fury.readBufferObject(buffer);

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

@@ -30,7 +30,8 @@
 import org.apache.fury.reflect.ReflectionUtils;
 
 /** Serializer for class implements {@link Externalizable}. */
-public class ExternalizableSerializer<T extends Externalizable> extends Serializer<T> {
+public class ExternalizableSerializer<T extends Externalizable>
+    extends AbstractObjectSerializer<T> {
   private final MethodHandle constructor;
   private final MemoryBufferObjectInput objectInput;
   private final MemoryBufferObjectOutput objectOutput;

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

@@ -96,6 +96,10 @@ protected Object deserialize(T serializer, MemoryBuffer buffer) {
       byte[] bytes = buffer.getRemainingBytes();
       return deserialize(serializer, bytes);
     }
+
+    protected Object copy(T serializer, Object obj) {
+      throw new UnsupportedOperationException();
+    }
   }
 
   public static class DefaultFuryProxy extends SerializerProxy<LoaderBinding> {
@@ -174,6 +178,11 @@ protected Object deserialize(LoaderBinding serializer, byte[] bytes) {
     protected Object deserialize(LoaderBinding serializer, MemoryBuffer buffer) {
       return serializer.get().deserialize(buffer);
     }
+
+    @Override
+    protected Object copy(LoaderBinding serializer, Object obj) {
+      return serializer.get().copy(obj);
+    }
   }
 
   private final SerializerProxy proxy;
@@ -264,4 +273,8 @@ public <T> T deserialize(ByteBuffer byteBuffer) {
   public <T> T deserialize(MemoryBuffer buffer) {
     return (T) proxy.deserialize(serializerLocal.get(), buffer);
   }
+
+  public <T> T copy(T obj) {
+    return (T) proxy.copy(serializerLocal.get(), obj);
+  }
 }

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

@@ -48,7 +48,7 @@
  * true, this serializer will be used.
  */
 @SuppressWarnings({"rawtypes", "unchecked"})
-public class JavaSerializer extends Serializer {
+public class JavaSerializer extends AbstractObjectSerializer {
   private static final Logger LOG = LoggerFactory.getLogger(JavaSerializer.class);
   private final MemoryBufferObjectInput objectInput;
   private final MemoryBufferObjectOutput objectOutput;

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

@@ -60,6 +60,20 @@ public void write(MemoryBuffer buffer, Object value) {
     fury.writeRef(buffer, Proxy.getInvocationHandler(value));
   }
 
+  @Override
+  public Object copy(Object value) {
+    Class<?>[] interfaces = value.getClass().getInterfaces();
+    InvocationHandler invocationHandler = Proxy.getInvocationHandler(value);
+    Preconditions.checkNotNull(interfaces);
+    Preconditions.checkNotNull(invocationHandler);
+    Object proxy = Proxy.newProxyInstance(fury.getClassLoader(), interfaces, STUB_HANDLER);
+    if (needToCopyRef) {
+      fury.reference(value, proxy);
+    }
+    Platform.putObject(proxy, PROXY_HANDLER_FIELD_OFFSET, fury.copyObject(invocationHandler));
+    return proxy;
+  }
+
   @Override
   public Object read(MemoryBuffer buffer) {
     final RefResolver resolver = fury.getRefResolver();

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

@@ -80,6 +80,17 @@ public void write(MemoryBuffer buffer, Object value) {
     }
   }
 
+  @Override
+  public Object copy(Object value) {
+    try {
+      Object replacement = writeReplaceMethod.invoke(value);
+      Object newReplacement = getDataSerializer().copy(replacement);
+      return READ_RESOLVE_METHOD.invoke(newReplacement);
+    } catch (Exception e) {
+      throw new RuntimeException("Can't copy lambda " + value, e);
+    }
+  }
+
   @Override
   public Object read(MemoryBuffer buffer) {
     try {

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

@@ -83,7 +83,9 @@ public ObjectSerializer(Fury fury, Class<T> cls, boolean resolveParent) {
     // avoid recursive building serializers.
     // Use `setSerializerIfAbsent` to avoid overwriting existing serializer for class when used
     // as data serializer.
-    classResolver.setSerializerIfAbsent(cls, this);
+    if (resolveParent) {
+      classResolver.setSerializerIfAbsent(cls, this);
+    }
     Collection<Descriptor> descriptors;
     boolean shareMeta = fury.getConfig().isMetaShareEnabled();
     if (shareMeta) {

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

@@ -58,7 +58,6 @@
 import org.apache.fury.memory.Platform;
 import org.apache.fury.reflect.ReflectionUtils;
 import org.apache.fury.resolver.ClassInfo;
-import org.apache.fury.resolver.ClassResolver;
 import org.apache.fury.resolver.FieldResolver;
 import org.apache.fury.resolver.FieldResolver.ClassField;
 import org.apache.fury.util.ExceptionUtils;
@@ -78,11 +77,10 @@
  * <p>`ObjectInputStream#setObjectInputFilter` will be ignored by this serializer.
  */
 @SuppressWarnings({"unchecked", "rawtypes"})
-public class ObjectStreamSerializer extends Serializer {
+public class ObjectStreamSerializer extends AbstractObjectSerializer {
   private static final Logger LOG = LoggerFactory.getLogger(ObjectStreamSerializer.class);
 
   private final Constructor constructor;
-  private final ClassResolver classResolver;
   private final SlotsInfo[] slotsInfos;
 
   public ObjectStreamSerializer(Fury fury, Class<?> type) {
@@ -109,7 +107,6 @@ public ObjectStreamSerializer(Fury fury, Class<?> type) {
       constructor =
           (Constructor) ReflectionUtils.getObjectFieldValue(ObjectStreamClass.lookup(type), "cons");
     }
-    this.classResolver = fury.getClassResolver();
     this.constructor = constructor;
     List<SlotsInfo> slotsInfoList = new ArrayList<>();
     Class<?> end = type;

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

@@ -44,7 +44,7 @@
  * will skip classname writing if object returned by `writeReplace` is different from current class.
  */
 @SuppressWarnings({"unchecked", "rawtypes"})
-public class ReplaceResolveSerializer extends Serializer {
+public class ReplaceResolveSerializer extends AbstractObjectSerializer {
   private static final Logger LOG = LoggerFactory.getLogger(ReplaceResolveSerializer.class);
 
   /**

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

@@ -27,7 +27,7 @@
 
 /** Serializer for {@link URL}. */
 // TODO(chaokunyang) ensure security to avoid dnslog detection.
-public final class URLSerializer extends ImmutableSerializer<URL> {
+public final class URLSerializer extends AbstractObjectSerializer<URL> {
 
   public URLSerializer(Fury fury, Class<URL> type) {
     super(fury, type);

java/fury-core/src/main/java/org/apache/fury/serializer/collection/AbstractCollectionSerializer.java

@@ -542,6 +542,35 @@ public Collection newCollection() {
     }
   }
 
+  public void copyElements(Collection originCollection, Collection newCollection) {
+    ClassResolver classResolver = fury.getClassResolver();
+    for (Object element : originCollection) {
+      if (element != null) {
+        ClassInfo classInfo =
+            classResolver.getClassInfo(element.getClass(), elementClassInfoHolder);
+        if (!classInfo.getSerializer().isImmutable()) {
+          element = fury.copyObject(element, classInfo.getClassId());
+        }
+      }
+      newCollection.add(element);
+    }
+  }
+
+  public void copyElements(Collection originCollection, Object[] elements) {
+    int index = 0;
+    ClassResolver classResolver = fury.getClassResolver();
+    for (Object element : originCollection) {
+      if (element != null) {
+        ClassInfo classInfo =
+            classResolver.getClassInfo(element.getClass(), elementClassInfoHolder);
+        if (!classInfo.getSerializer().isImmutable()) {
+          element = fury.copyObject(element, classInfo.getSerializer());
+        }
+      }
+      elements[index++] = element;
+    }
+  }
+
   private RuntimeException buildException(Throwable e) {
     return new IllegalArgumentException(
         "Please provide public no arguments constructor for class " + type, e);

java/fury-core/src/main/java/org/apache/fury/serializer/collection/AbstractMapSerializer.java

@@ -21,8 +21,10 @@
 
 import static org.apache.fury.type.TypeUtils.MAP_TYPE;
 
+import com.google.common.collect.ImmutableMap.Builder;
 import java.lang.invoke.MethodHandle;
 import java.util.Map;
+import java.util.Map.Entry;
 import org.apache.fury.Fury;
 import org.apache.fury.collection.IdentityMap;
 import org.apache.fury.collection.Tuple2;
@@ -439,6 +441,28 @@ protected <K, V> void copyEntry(Map<K, V> originMap, Map<K, V> newMap) {
     }
   }
 
+  protected <K, V> void copyEntry(Map<K, V> originMap, Builder<K, V> builder) {
+    ClassResolver classResolver = fury.getClassResolver();
+    for (Entry<K, V> entry : originMap.entrySet()) {
+      K key = entry.getKey();
+      if (key != null) {
+        ClassInfo classInfo = classResolver.getClassInfo(key.getClass(), keyClassInfoWriteCache);
+        if (!classInfo.getSerializer().isImmutable()) {
+          key = fury.copyObject(key, classInfo.getClassId());
+        }
+      }
+      V value = entry.getValue();
+      if (value != null) {
+        ClassInfo classInfo =
+            classResolver.getClassInfo(value.getClass(), valueClassInfoWriteCache);
+        if (!classInfo.getSerializer().isImmutable()) {
+          value = fury.copyObject(value, classInfo.getClassId());
+        }
+      }
+      builder.put(key, value);
+    }
+  }
+
   @SuppressWarnings("unchecked")
   protected final void readElements(MemoryBuffer buffer, int size, Map map) {
     Serializer keySerializer = this.keySerializer;

java/fury-core/src/main/java/org/apache/fury/serializer/collection/ChildContainerSerializers.java

@@ -127,7 +127,7 @@ public static class ChildCollectionSerializer<T extends Collection>
             ArrayList.class, LinkedList.class, ArrayDeque.class, Vector.class, HashSet.class
             // PriorityQueue/TreeSet/ConcurrentSkipListSet need comparator as constructor argument
             );
-
+    protected Serializer<T> objectSerializer;
     protected final Serializer[] slotsSerializers;
 
     public ChildCollectionSerializer(Fury fury, Class<T> cls) {
@@ -149,6 +149,14 @@ public Collection newCollection(MemoryBuffer buffer) {
       readAndSetFields(buffer, collection, slotsSerializers);
       return collection;
     }
+
+    @Override
+    public T copy(T originCollection) {
+      if (objectSerializer == null) {
+        objectSerializer = new ObjectSerializer<>(fury, type, false);
+      }
+      return fury.copyObject(originCollection, objectSerializer);
+    }
   }
 
   public static final class ChildArrayListSerializer<T extends ArrayList>
@@ -177,6 +185,7 @@ public static class ChildMapSerializer<T extends Map> extends MapSerializer<T> {
             HashMap.class, LinkedHashMap.class, ConcurrentHashMap.class
             // TreeMap/ConcurrentSkipListMap need comparator as constructor argument
             );
+    private Serializer<T> objectSerializer;
     private final Serializer[] slotsSerializers;
 
     public ChildMapSerializer(Fury fury, Class<T> cls) {
@@ -199,6 +208,14 @@ public Map newMap(MemoryBuffer buffer) {
       readAndSetFields(buffer, map, slotsSerializers);
       return map;
     }
+
+    @Override
+    public T copy(T originMap) {
+      if (objectSerializer == null) {
+        objectSerializer = new ObjectSerializer<>(fury, type, false);
+      }
+      return fury.copyObject(originMap, objectSerializer);
+    }
   }
 
   private static <T> Serializer[] buildSlotsSerializers(

java/fury-core/src/main/java/org/apache/fury/serializer/collection/CollectionSerializer.java

@@ -22,8 +22,6 @@
 import java.util.Collection;
 import org.apache.fury.Fury;
 import org.apache.fury.memory.MemoryBuffer;
-import org.apache.fury.resolver.ClassInfo;
-import org.apache.fury.resolver.ClassResolver;
 
 /** Base serializer for all java collections. */
 @SuppressWarnings({"unchecked", "rawtypes"})
@@ -65,20 +63,6 @@ public T copy(T originCollection) {
     return (T) newCollection;
   }
 
-  public void copyElements(T originCollection, Collection newCollection) {
-    ClassResolver classResolver = fury.getClassResolver();
-    for (Object element : originCollection) {
-      if (element != null) {
-        ClassInfo classInfo =
-            classResolver.getClassInfo(element.getClass(), elementClassInfoHolder);
-        if (!classInfo.getSerializer().isImmutable()) {
-          element = fury.copyObject(element, classInfo.getClassId());
-        }
-      }
-      newCollection.add(element);
-    }
-  }
-
   @Override
   public T read(MemoryBuffer buffer) {
     Collection collection = newCollection(buffer);

java/fury-core/src/main/java/org/apache/fury/serializer/collection/CollectionSerializers.java

@@ -107,23 +107,12 @@ public List<?> copy(List<?> originCollection) {
       Object[] elements = new Object[originCollection.size()];
       List<?> newCollection = Arrays.asList(elements);
       if (needToCopyRef) {
-        List<?> copyObject = (List<?>) fury.getCopyObject(originCollection);
-        if (copyObject != null) {
-          return copyObject;
-        }
         fury.reference(originCollection, newCollection);
       }
       copyElements(originCollection, elements);
       return newCollection;
     }
 
-    private void copyElements(List<?> originCollection, Object[] elements) {
-      int size = originCollection.size();
-      for (int i = 0; i < size; i++) {
-        elements[i] = fury.copyObject(originCollection.get(i));
-      }
-    }
-
     @Override
     public short getXtypeId() {
       return (short) -Type.LIST.getId();