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

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 (shareMeta && 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 = getSerializationContext().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 (shareMeta && 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);
+  }
 }