perf(java): optimize scoped meta share mode perf (#1734)

## What does this PR do?

This PR optimizes scoped meta share mode writing perf by about 30%:
- Replace ArrayList by ObjectArray, which can save `clear` cost
- Speed up copy performance when writing classdefs 

## Related issues

#1733 

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

Author: chaokunyang

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

@@ -25,7 +25,8 @@
  * An auto-growing array which avoid checks in {@code ArrayList} and faster for {@code clear}
  * method.
  */
-public final class ObjectArray {
+@SuppressWarnings("unchecked")
+public final class ObjectArray<T> {
   private static final int COPY_THRESHOLD = 128;
   private static final int NIL_ARRAY_SIZE = 1024;
   private static final Object[] NIL_ARRAY = new Object[NIL_ARRAY_SIZE];
@@ -41,7 +42,7 @@ public ObjectArray(int initialCapacity) {
     objects = new Object[initialCapacity];
   }
 
-  public void add(Object element) {
+  public void add(T element) {
     Object[] objects = this.objects;
     int size = this.size++;
     if (objects.length <= size) {
@@ -53,22 +54,22 @@ public void add(Object element) {
     objects[size] = element;
   }
 
-  public void set(int index, Object element) {
+  public void set(int index, T element) {
     objects[index] = element;
   }
 
-  public Object get(int index) {
-    return objects[index];
+  public T get(int index) {
+    return (T) objects[index];
   }
 
   /** Returns tail item or null if no available item in the array. */
-  public Object popOrNull() {
+  public T popOrNull() {
     int size = this.size;
     if (size == 0) {
       return null;
     }
     this.size = --size;
-    return objects[size];
+    return (T) objects[size];
   }
 
   public void clear() {

java/fury-core/src/main/java/org/apache/fury/memory/MemoryBuffer.java

@@ -233,6 +233,14 @@ public boolean isOffHeap() {
     return heapMemory == null;
   }
 
+  /**
+   * Returns <tt>true</tt>, if the memory buffer is backed by heap memory and memory buffer can
+   * write to the whole memory region of underlying byte array.
+   */
+  public boolean isHeapFullyWriteable() {
+    return heapMemory != null && heapOffset == 0;
+  }
+
   /**
    * Get the heap byte array object.
    *

java/fury-core/src/main/java/org/apache/fury/meta/ClassDef.java

@@ -197,7 +197,7 @@ public String toString() {
 
   /** Write class definition to buffer. */
   public void writeClassDef(MemoryBuffer buffer) {
-    buffer.writeBytes(encoded);
+    buffer.writeBytes(encoded, 0, encoded.length);
   }
 
   /** Read class definition from buffer. */

java/fury-core/src/main/java/org/apache/fury/resolver/ClassInfo.java

@@ -48,7 +48,7 @@ public class ClassInfo {
   // class id must be less than Integer.MAX_VALUE/2 since we use bit 0 as class id flag.
   short classId;
   ClassDef classDef;
-  public boolean needToWriteClassDef;
+  boolean needToWriteClassDef;
 
   ClassInfo(
       Class<?> cls,

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

@@ -88,6 +88,7 @@
 import org.apache.fury.collection.IdentityMap;
 import org.apache.fury.collection.IdentityObjectIntMap;
 import org.apache.fury.collection.LongMap;
+import org.apache.fury.collection.ObjectArray;
 import org.apache.fury.collection.ObjectMap;
 import org.apache.fury.collection.Tuple2;
 import org.apache.fury.config.CompatibleMode;
@@ -204,7 +205,9 @@ public class ClassResolver {
   private static final float loadFactor = 0.25f;
   private static final float furyMapLoadFactor = 0.25f;
   private static final int estimatedNumRegistered = 150;
-  private static final String META_SHARE_FIELDS_INFO_KEY = "shareFieldsInfo";
+  private static final String SET_META__CONTEXT_MSG =
+      "Meta context must be set before serialization, "
+          + "please set meta context by SerializationContext.setMetaContext";
   private static final ClassInfo NIL_CLASS_INFO =
       new ClassInfo(null, null, null, null, false, null, null, ClassResolver.NO_CLASS_ID);
 
@@ -1285,10 +1288,7 @@ public void writeClassWithMetaShare(MemoryBuffer buffer, ClassInfo classInfo) {
       return;
     }
     MetaContext metaContext = fury.getSerializationContext().getMetaContext();
-    Preconditions.checkNotNull(
-        metaContext,
-        "Meta context must be set before serialization, "
-            + "please set meta context by SerializationContext.setMetaContext");
+    assert metaContext != null : SET_META__CONTEXT_MSG;
     IdentityObjectIntMap<Class<?>> classMap = metaContext.classMap;
     int newId = classMap.size;
     int id = classMap.putOrGet(classInfo.cls, newId);
@@ -1332,19 +1332,16 @@ boolean needToWriteClassDef(Serializer serializer) {
   }
 
   private ClassInfo readClassInfoWithMetaShare(MemoryBuffer buffer, MetaContext metaContext) {
-    Preconditions.checkNotNull(
-        metaContext,
-        "Meta context must be set before serialization,"
-            + " please set meta context by SerializationContext.setMetaContext");
+    assert metaContext != null : SET_META__CONTEXT_MSG;
     int header = buffer.readVarUint32Small14();
     int id = header >>> 1;
     if ((header & 0b1) == 0) {
       return getOrUpdateClassInfo((short) id);
     }
-    List<ClassInfo> readClassInfos = metaContext.readClassInfos;
+    ObjectArray<ClassInfo> readClassInfos = metaContext.readClassInfos;
     ClassInfo classInfo = readClassInfos.get(id);
     if (classInfo == null) {
-      List<ClassDef> readClassDefs = metaContext.readClassDefs;
+      ObjectArray<ClassDef> readClassDefs = metaContext.readClassDefs;
       ClassDef classDef = readClassDefs.get(id);
       Tuple2<ClassDef, ClassInfo> classDefTuple = extRegistry.classIdToDef.get(classDef.getId());
       if (classDefTuple == null || classDefTuple.f1 == null) {
@@ -1433,11 +1430,31 @@ private ClassInfo getMetaSharedClassInfo(ClassDef classDef, Class<?> clz) {
    */
   public void writeClassDefs(MemoryBuffer buffer) {
     MetaContext metaContext = fury.getSerializationContext().getMetaContext();
-    buffer.writeVarUint32Small7(metaContext.writingClassDefs.size());
-    for (ClassDef classDef : metaContext.writingClassDefs) {
-      classDef.writeClassDef(buffer);
+    ObjectArray<ClassDef> writingClassDefs = metaContext.writingClassDefs;
+    final int size = writingClassDefs.size;
+    buffer.writeVarUint32Small7(size);
+    if (buffer.isHeapFullyWriteable()) {
+      writeClassDefs(buffer, writingClassDefs, size);
+    } else {
+      for (int i = 0; i < size; i++) {
+        writingClassDefs.get(i).writeClassDef(buffer);
+      }
+    }
+    metaContext.writingClassDefs.size = 0;
+  }
+
+  private void writeClassDefs(
+      MemoryBuffer buffer, ObjectArray<ClassDef> writingClassDefs, int size) {
+    int writerIndex = buffer.writerIndex();
+    for (int i = 0; i < size; i++) {
+      byte[] encoded = writingClassDefs.get(i).getEncoded();
+      int bytesLen = encoded.length;
+      buffer.ensure(writerIndex + bytesLen);
+      final byte[] targetArray = buffer.getHeapMemory();
+      System.arraycopy(encoded, 0, targetArray, writerIndex, bytesLen);
+      writerIndex += bytesLen;
     }
-    metaContext.writingClassDefs.clear();
+    buffer.writerIndex(writerIndex);
   }
 
   /**

java/fury-core/src/main/java/org/apache/fury/resolver/MetaContext.java

@@ -19,9 +19,8 @@
 
 package org.apache.fury.resolver;
 
-import java.util.ArrayList;
-import java.util.List;
 import org.apache.fury.collection.IdentityObjectIntMap;
+import org.apache.fury.collection.ObjectArray;
 import org.apache.fury.memory.MemoryBuffer;
 import org.apache.fury.meta.ClassDef;
 
@@ -34,15 +33,15 @@ public class MetaContext {
   public final IdentityObjectIntMap<Class<?>> classMap = new IdentityObjectIntMap<>(8, 0.4f);
 
   /** Class definitions read from peer. */
-  public final List<ClassDef> readClassDefs = new ArrayList<>();
+  public final ObjectArray<ClassDef> readClassDefs = new ObjectArray<>();
 
-  public final List<ClassInfo> readClassInfos = new ArrayList<>();
+  public final ObjectArray<ClassInfo> readClassInfos = new ObjectArray<>();
 
   /**
    * New class definition which needs sending to peer. This will be filled up when there are new
    * class definition need sending, and will be cleared after writing to buffer.
    *
    * @see ClassResolver#writeClassDefs(MemoryBuffer)
    */
-  public final List<ClassDef> writingClassDefs = new ArrayList<>();
+  public final ObjectArray<ClassDef> writingClassDefs = new ObjectArray<>();
 }

java/fury-core/src/main/java/org/apache/fury/resolver/SerializationContext.java

@@ -74,7 +74,7 @@ public void resetWrite() {
     }
     if (scopedMetaShareEnabled) {
       metaContext.classMap.clear();
-      metaContext.writingClassDefs.clear();
+      metaContext.writingClassDefs.size = 0;
     } else {
       metaContext = null;
     }
@@ -85,8 +85,8 @@ public void resetRead() {
       objects.clear();
     }
     if (scopedMetaShareEnabled) {
-      metaContext.readClassInfos.clear();
-      metaContext.readClassDefs.clear();
+      metaContext.readClassInfos.size = 0;
+      metaContext.readClassDefs.size = 0;
     } else {
       metaContext = null;
     }
@@ -98,9 +98,9 @@ public void reset() {
     }
     if (scopedMetaShareEnabled) {
       metaContext.classMap.clear();
-      metaContext.writingClassDefs.clear();
-      metaContext.readClassInfos.clear();
-      metaContext.readClassDefs.clear();
+      metaContext.writingClassDefs.size = 0;
+      metaContext.readClassInfos.size = 0;
+      metaContext.readClassDefs.size = 0;
     } else {
       metaContext = null;
     }

java/fury-core/src/test/java/org/apache/fury/resolver/MetaContextTest.java

@@ -79,7 +79,7 @@ private void checkMetaShared(Fury fury, Object o) {
     Assert.assertEquals(fury.deserialize(bytes1), o);
     fury.getSerializationContext().setMetaContext(new MetaContext());
     Assert.assertEquals(fury.serialize(o), bytes);
-    Assert.assertThrows(NullPointerException.class, () -> fury.serialize(o));
+    Assert.assertThrows(AssertionError.class, () -> fury.serialize(o));
   }
 
   // final InnerPojo will be taken as non-final for writing class def.