chore(java): Disallow writing meta classdef when obj is null (#1686)

## What does this PR do?

<!-- Describe the purpose of this PR. -->
When obj is null, if shareMeta is enabled, no need to write meta
classdef

## Related issues

<!--
Is there any related issue? Please attach here.

- #xxxx0
- #xxxx1
- #xxxx2
-->


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

---------

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

java/fury-core/src/main/java/org/apache/fury/Fury.java

@@ -48,6 +48,7 @@
 import org.apache.fury.resolver.ClassInfoHolder;
 import org.apache.fury.resolver.ClassResolver;
 import org.apache.fury.resolver.MapRefResolver;
+import org.apache.fury.resolver.MetaContext;
 import org.apache.fury.resolver.MetaStringResolver;
 import org.apache.fury.resolver.NoRefResolver;
 import org.apache.fury.resolver.RefResolver;
@@ -155,6 +156,7 @@ public Fury(FuryBuilder builder, ClassLoader classLoader) {
     arrayListSerializer = new ArrayListSerializer(this);
     hashMapSerializer = new HashMapSerializer(this);
     originToCopyMap = new IdentityMap<>();
+    classDefEndOffset = -1;
     LOG.info("Created new fury {}", this);
   }
 
@@ -344,7 +346,8 @@ private void write(MemoryBuffer buffer, Object obj) {
       classResolver.writeClass(buffer, classInfo);
       writeData(buffer, classInfo, obj);
     }
-    if (shareMeta) {
+    MetaContext metaContext = serializationContext.getMetaContext();
+    if (shareMeta && metaContext != null && !metaContext.writingClassDefs.isEmpty()) {
       buffer.putInt32(startOffset, buffer.writerIndex() - startOffset - 4);
       classResolver.writeClassDefs(buffer);
     }
@@ -792,7 +795,7 @@ public Object deserialize(MemoryBuffer buffer, Iterable<MemoryBuffer> outOfBandB
     } catch (Throwable t) {
       throw ExceptionUtils.handleReadFailed(this, t);
     } finally {
-      if (shareMeta) {
+      if (classDefEndOffset != -1) {
         buffer.readerIndex(classDefEndOffset);
       }
       resetRead();
@@ -1063,9 +1066,12 @@ public void serializeJavaObject(MemoryBuffer buffer, Object obj) {
         if (!refResolver.writeRefOrNull(buffer, obj)) {
           ClassInfo classInfo = classResolver.getOrUpdateClassInfo(obj.getClass());
           writeData(buffer, classInfo, obj);
+          MetaContext metaContext = serializationContext.getMetaContext();
+          if (metaContext != null && !metaContext.writingClassDefs.isEmpty()) {
+            buffer.putInt32(startOffset, buffer.writerIndex() - startOffset - 4);
+            classResolver.writeClassDefs(buffer);
+          }
         }
-        buffer.putInt32(startOffset, buffer.writerIndex() - startOffset - 4);
-        classResolver.writeClassDefs(buffer);
       } else {
         if (!refResolver.writeRefOrNull(buffer, obj)) {
           ClassInfo classInfo = classResolver.getOrUpdateClassInfo(obj.getClass());
@@ -1116,7 +1122,7 @@ public <T> T deserializeJavaObject(MemoryBuffer buffer, Class<T> cls) {
     } catch (Throwable t) {
       throw ExceptionUtils.handleReadFailed(this, t);
     } finally {
-      if (shareMeta) {
+      if (classDefEndOffset != -1) {
         buffer.readerIndex(classDefEndOffset);
       }
       resetRead();
@@ -1226,7 +1232,7 @@ public Object deserializeJavaObjectAndClass(MemoryBuffer buffer) {
     } catch (Throwable t) {
       throw ExceptionUtils.handleReadFailed(this, t);
     } finally {
-      if (shareMeta) {
+      if (classDefEndOffset != -1) {
         buffer.readerIndex(classDefEndOffset);
       }
       resetRead();
@@ -1437,6 +1443,9 @@ private void serializeToStream(OutputStream outputStream, Consumer<MemoryBuffer>
 
   private void readClassDefs(MemoryBuffer buffer) {
     int relativeClassDefOffset = buffer.readInt32();
+    if (relativeClassDefOffset == -1) {
+      return;
+    }
     int readerIndex = buffer.readerIndex();
     buffer.readerIndex(readerIndex + relativeClassDefOffset);
     classResolver.readClassDefs(buffer);
@@ -1474,6 +1483,7 @@ public void resetRead() {
     nativeObjects.clear();
     peerOutOfBandEnabled = false;
     depth = 0;
+    classDefEndOffset = -1;
   }
 
   public void resetCopy() {

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

@@ -90,6 +90,10 @@ public int size() {
     return size;
   }
 
+  public boolean isEmpty() {
+    return size == 0;
+  }
+
   /**
    * Set all object array elements to null. This method is faster than {@link Arrays#fill} for large
    * arrays (> 128).

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

@@ -1470,6 +1470,7 @@ private void writeClassDefs(
    */
   public void readClassDefs(MemoryBuffer buffer) {
     MetaContext metaContext = fury.getSerializationContext().getMetaContext();
+    assert metaContext != null : SET_META__CONTEXT_MSG;
     int numClassDefs = buffer.readVarUint32Small14();
     for (int i = 0; i < numClassDefs; i++) {
       long id = buffer.readInt64();

java/fury-core/src/test/java/org/apache/fury/FuryTest.java

@@ -62,6 +62,7 @@
 import org.apache.fury.memory.MemoryBuffer;
 import org.apache.fury.memory.MemoryUtils;
 import org.apache.fury.memory.Platform;
+import org.apache.fury.resolver.MetaContext;
 import org.apache.fury.serializer.ArraySerializersTest;
 import org.apache.fury.serializer.EnumSerializerTest;
 import org.apache.fury.serializer.ObjectSerializer;
@@ -598,4 +599,20 @@ public void testPrintReadObjectsWhenFailed() {
       Assert.assertTrue(e.getMessage().contains("[a, b]"));
     }
   }
+
+  @Test
+  public void testNullObjSerAndDe() {
+    Fury fury =
+        Fury.builder()
+            .withRefTracking(true)
+            .requireClassRegistration(false)
+            .withMetaShare(true)
+            .build();
+    MetaContext metaContext = new MetaContext();
+    fury.getSerializationContext().setMetaContext(metaContext);
+    byte[] bytes = fury.serializeJavaObjectAndClass(null);
+    fury.getSerializationContext().setMetaContext(metaContext);
+    Object obj = fury.deserializeJavaObjectAndClass(bytes);
+    assertNull(obj);
+  }
 }