[Java] use varint for class id encoding to reduce space cost

java/fury-core/src/main/java/io/fury/Fury.java

@@ -743,7 +743,7 @@ public Object readRef(MemoryBuffer buffer) {
     int nextReadRefId = refResolver.tryPreserveRefId(buffer);
     if (nextReadRefId >= NOT_NULL_VALUE_FLAG) {
       // ref value or not-null value
-      Object o = readDataInternal(buffer, classResolver.readAndUpdateClassInfoHolder(buffer));
+      Object o = readDataInternal(buffer, classResolver.readClassInfo(buffer));
       refResolver.setReadObject(nextReadRefId, o);
       return o;
     } else {
@@ -788,7 +788,7 @@ public <T> T readRef(MemoryBuffer buffer, Serializer<T> serializer) {
 
   /** Deserialize not-null and non-reference object from <code>buffer</code>. */
   public Object readNonRef(MemoryBuffer buffer) {
-    return readDataInternal(buffer, classResolver.readAndUpdateClassInfoHolder(buffer));
+    return readDataInternal(buffer, classResolver.readClassInfo(buffer));
   }
 
   public Object readNonRef(MemoryBuffer buffer, ClassInfoHolder classInfoHolder) {

java/fury-core/src/main/java/io/fury/resolver/ClassInfo.java

@@ -37,6 +37,7 @@ public class ClassInfo {
   final EnumStringBytes typeTagBytes;
   Serializer<?> serializer;
   // use primitive to avoid boxing
+  // class id must be less than Integer.MAX_VALUE/2 since we use bit 0 as class id flag.
   short classId;
 
   ClassInfo(

java/fury-core/src/main/java/io/fury/serializer/CollectionSerializers.java

@@ -1107,7 +1107,7 @@ public void write(MemoryBuffer buffer, EnumSet object) {
 
     @Override
     public EnumSet read(MemoryBuffer buffer) {
-      Class elemClass = fury.getClassResolver().readClassAndUpdateCache(buffer);
+      Class elemClass = fury.getClassResolver().readClassInfo(buffer).getCls();
       EnumSet object = EnumSet.noneOf(elemClass);
       Serializer elemSerializer = fury.getClassResolver().getSerializer(elemClass);
       int length = buffer.readPositiveAlignedVarInt();

java/fury-core/src/main/java/io/fury/serializer/MapSerializers.java

@@ -1040,7 +1040,7 @@ public void writeHeader(MemoryBuffer buffer, EnumMap value) {
 
     @Override
     public EnumMap newMap(MemoryBuffer buffer, int numElements) {
-      Class<?> keyType = fury.getClassResolver().readClassAndUpdateCache(buffer);
+      Class<?> keyType = fury.getClassResolver().readClassInfo(buffer).getCls();
       return new EnumMap(keyType);
     }
   }

java/fury-core/src/main/java/io/fury/serializer/ReplaceResolveSerializer.java

@@ -268,7 +268,7 @@ public Object read(MemoryBuffer buffer) {
       int nextReadRefId = refResolver.tryPreserveRefId(buffer);
       if (nextReadRefId >= Fury.NOT_NULL_VALUE_FLAG) {
         // ref value or not-null value
-        Object o = fury.readData(buffer, classResolver.readAndUpdateClassInfoHolder(buffer));
+        Object o = fury.readData(buffer, classResolver.readClassInfo(buffer));
         refResolver.setReadObject(nextReadRefId, o);
         refResolver.setReadObject(outerRefId, o);
         return o;

java/fury-core/src/main/java/io/fury/serializer/UnexistedClassSerializers.java

@@ -104,8 +104,8 @@ private void writeClassDef(MemoryBuffer buffer, UnexistedMetaSharedClass value)
       // Register NotFoundClass ahead to skip write meta shared info,
       // then revert written class id to write class info here,
       // since it's the only place to hold class def for not found class.
-      buffer.increaseWriterIndex(-3);
-      buffer.writeByte(ClassResolver.USE_CLASS_VALUE);
+      buffer.increaseWriterIndex(-2);
+      buffer.writeByte(ClassResolver.USE_CLASS_VALUE_FLAG);
       MetaContext metaContext = fury.getSerializationContext().getMetaContext();
       IdentityObjectIntMap classMap = metaContext.classMap;
       int newId = classMap.size;

java/fury-core/src/test/java/io/fury/resolver/ClassResolverTest.java

@@ -221,8 +221,8 @@ public void testWriteClassName() {
       MemoryBuffer buffer = MemoryUtils.buffer(32);
       classResolver.writeClassAndUpdateCache(buffer, getClass());
       classResolver.writeClassAndUpdateCache(buffer, getClass());
-      Assert.assertSame(classResolver.readClassAndUpdateCache(buffer), getClass());
-      Assert.assertSame(classResolver.readClassAndUpdateCache(buffer), getClass());
+      Assert.assertSame(classResolver.readClassInfo(buffer).getCls(), getClass());
+      Assert.assertSame(classResolver.readClassInfo(buffer).getCls(), getClass());
       classResolver.reset();
       buffer.writerIndex(0);
       buffer.readerIndex(0);