fix(java): fix compatible metadata state regressions

java/fory-core/src/main/java/org/apache/fory/builder/BaseObjectCodecBuilder.java

@@ -1192,27 +1192,37 @@ protected Reference addExactTypeInfoField(Class<?> cls) {
   }
 
   protected Reference addTypeInfoHolderField(Class<?> cls) {
+    return addTypeInfoHolderField(cls, "TypeInfoHolder");
+  }
+
+  private Reference addTypeInfoHolderField(Class<?> cls, String fieldSuffix) {
     // Final type need to write classinfo when meta share enabled.
-    String key;
+    String fieldKey;
     if (ReflectionUtils.isMonomorphic(cls)) {
-      key = "classInfoHolder:" + cls;
+      fieldKey = fieldSuffix + ":" + cls;
     } else {
-      key = "classInfoHolder:" + cls + walkPath;
+      fieldKey = fieldSuffix + ":" + cls + walkPath;
     }
-    Reference reference = (Reference) sharedFieldMap.get(key);
+    Reference reference = (Reference) sharedFieldMap.get(fieldKey);
     if (reference != null) {
       return reference;
     }
     Expression classInfoHolderExpr =
         inlineInvoke(typeResolverRef, "nilTypeInfoHolder", classInfoHolderTypeRef);
-    String name = ctx.newName(cls, "TypeInfoHolder");
+    String name = ctx.newName(cls, fieldSuffix);
     ctx.addField(true, ctx.type(TypeInfoHolder.class), name, classInfoHolderExpr);
     // The class info field read only once, no need to shallow.
     reference = new Reference(name, classInfoHolderTypeRef);
-    sharedFieldMap.put(key, reference);
+    sharedFieldMap.put(fieldKey, reference);
     return reference;
   }
 
+  protected Reference addWriteTypeInfoHolderField(Class<?> cls) {
+    // Compatible reads may retain remote schema metadata. Keep generated collection writes on a
+    // distinct local cache even when both paths have the same class and walk path.
+    return addTypeInfoHolderField(cls, "WriteTypeInfoHolder");
+  }
+
   protected Expression readTypeInfo(Class<?> cls, Expression ignored) {
     return readTypeInfo(cls, ignored, true);
   }
@@ -1544,7 +1554,7 @@ private Tuple2<Expression, Invoke> writeElementsHeader(
       return Tuple2.of(bitmap, null);
     } else {
       Expression elementTypeExpr = getClassExpr(elementType);
-      Expression classInfoHolder = addTypeInfoHolderField(elementType);
+      Expression classInfoHolder = addWriteTypeInfoHolderField(elementType);
       Expression bitmap;
       if (trackingRef) {
         if (elementType == Object.class) {

java/fory-core/src/main/java/org/apache/fory/builder/CompatibleCodecBuilder.java

@@ -52,6 +52,7 @@
 import org.apache.fory.resolver.TypeResolver;
 import org.apache.fory.serializer.CodegenSerializer;
 import org.apache.fory.serializer.CompatibleSerializer;
+import org.apache.fory.serializer.FieldGroups;
 import org.apache.fory.serializer.FieldGroups.SerializationFieldInfo;
 import org.apache.fory.serializer.ObjectSerializer;
 import org.apache.fory.serializer.Serializer;
@@ -561,13 +562,10 @@ private void ensureCompatibleFieldInfos() {
 
   public static SerializationFieldInfo[] buildRemoteFieldInfos(
       TypeResolver typeResolver, Class<?> cls, TypeDef typeDef) {
-    List<Descriptor> descriptors =
-        typeResolver.createDescriptorGrouper(typeDef, cls).getSortedDescriptors();
-    SerializationFieldInfo[] fieldInfos = new SerializationFieldInfo[descriptors.size()];
-    for (int i = 0; i < descriptors.size(); i++) {
-      fieldInfos[i] = new SerializationFieldInfo(typeResolver, descriptors.get(i));
-    }
-    return fieldInfos;
+    DescriptorGrouper grouper = typeResolver.createDescriptorGrouper(typeDef, cls);
+    // Generated skip helpers dispatch from SerializationFieldInfo.codecCategory. Preserve the
+    // grouper's category instead of rebuilding every remote field as OTHER.
+    return FieldGroups.buildFieldInfos(typeResolver, grouper).allFields;
   }
 
   public static SerializationFieldInfo[] buildLocalFieldInfosByRemoteOrder(

java/fory-core/src/main/java/org/apache/fory/meta/NativeTypeDefDecoder.java

@@ -179,7 +179,9 @@ static TypeDef decodeTypeDef(
           int typeId = i == numClasses - 1 ? rootTypeId : resolver.getTypeIdForTypeDef(cls);
           classSpec = new ClassSpec(cls, typeId, resolver.getUserTypeIdForTypeDef(cls));
           currentClass = cls;
-        } else {
+        } else if (i == numClasses - 1) {
+          // Only the root layer represents a dynamic object type. Non-root layers only label field
+          // ownership, so compatible matching uses their wire names without loading those classes.
           // `loadClassForMeta` keeps name-level checks before Class.forName; do not replace this
           // metadata path with direct class loading from the remote TypeDef name.
           Class<?> cls =

java/fory-core/src/main/java/org/apache/fory/meta/TypeDef.java

@@ -445,6 +445,8 @@ private List<Descriptor> buildDescriptors(
       } else {
         String fieldName = fieldInfo.getFieldName();
         String definedClass = fieldInfo.getDefinedClass();
+        // Match the wire declaring-class name before any compatibility fallback attempts to
+        // resolve that name as a class.
         descriptor = descriptorsMap.get(definedClass + "." + fieldName);
       }
       boolean remoteOnly = false;

java/fory-core/src/main/java/org/apache/fory/serializer/AbstractObjectSerializer.java

@@ -542,7 +542,7 @@ private static Object readContainerFieldValueNoRef(
     if (readContext.getConfig().isXlang()) {
       return readContext.readNonRef(fieldInfo.containerTypeInfo);
     }
-    return readContext.readNonRef(fieldInfo.classInfoHolder);
+    return readContext.readNonRef(fieldInfo.classInfoReadHolder);
   }
 
   private static Object readContainerFieldValueRef(

java/fory-core/src/main/java/org/apache/fory/serializer/ArraySerializers.java

@@ -109,7 +109,10 @@ public static Serializer<?> newObjectArraySerializer(TypeResolver typeResolver,
   @SuppressWarnings({"unchecked", "rawtypes"})
   public static final class ObjectArraySerializer extends Serializer<Object[]> {
     private final TypeResolver typeResolver;
+    // Compatible reads may cache remote schema metadata; local writes and copies retain the
+    // original local holder and must not reuse read state.
     private final TypeInfoHolder elementTypeInfoHolder;
+    private final TypeInfoHolder elementTypeInfoReadHolder;
 
     public ObjectArraySerializer(TypeResolver typeResolver, Class<?> cls) {
       super(typeResolver.getConfig(), (Class) cls);
@@ -119,6 +122,7 @@ public ObjectArraySerializer(TypeResolver typeResolver, Class<?> cls) {
       }
       Preconditions.checkArgument(cls.isArray() && !cls.getComponentType().isPrimitive());
       elementTypeInfoHolder = typeResolver.nilTypeInfoHolder();
+      elementTypeInfoReadHolder = typeResolver.nilTypeInfoHolder();
     }
 
     @Override
@@ -155,7 +159,7 @@ public Object[] read(ReadContext readContext) {
         readArrayElements(
             readContext,
             typeResolver,
-            elementTypeInfoHolder,
+            elementTypeInfoReadHolder,
             type.getComponentType(),
             value,
             numElements);

java/fory-core/src/main/java/org/apache/fory/serializer/FieldGroups.java

@@ -183,7 +183,10 @@ public static final class SerializationFieldInfo {
     public final TypeInfo typeInfo;
     public final Serializer serializer;
     public final GenericType genericType;
+    // Native compatible container reads may cache remote schema metadata; writes retain the
+    // original local holder and must not reuse read state.
     public final TypeInfoHolder classInfoHolder;
+    public final TypeInfoHolder classInfoReadHolder;
     public final TypeInfo containerTypeInfo;
     public final Serializer<?> containerSerializerOverride;
     public final FieldCodecCategory codecCategory;
@@ -289,6 +292,9 @@ public SerializationFieldInfo(TypeResolver resolver, Descriptor d) {
       } else {
         containerSerializerOverride = null;
       }
+      // Container fields retain distinct remote-read and local-write type metadata caches.
+      classInfoReadHolder =
+          codecCategory == FieldCodecCategory.CONTAINER ? resolver.nilTypeInfoHolder() : null;
       if (!resolver.isCrossLanguage()) {
         containerTypeInfo = null;
       } else {

java/fory-core/src/main/java/org/apache/fory/serializer/collection/CollectionLikeSerializer.java

@@ -55,7 +55,10 @@ public abstract class CollectionLikeSerializer<T> extends Serializer<T> {
   private final int collectionOwnerBytes;
   protected final Config config;
   protected final boolean supportCodegenHook;
+  // Compatible reads may cache remote schema metadata; local writes and copies retain the
+  // original local holder and must not reuse read state.
   protected final TypeInfoHolder elementTypeInfoHolder;
+  protected final TypeInfoHolder elementTypeInfoReadHolder;
   protected final TypeResolver typeResolver;
 
   // For subclass whose element type are instantiated already, such as
@@ -97,6 +100,7 @@ protected CollectionLikeSerializer(
     this.collectionOwnerBytes = collectionOwnerBytes;
     this.supportCodegenHook = supportCodegenHook;
     elementTypeInfoHolder = typeResolver.nilTypeInfoHolder();
+    elementTypeInfoReadHolder = typeResolver.nilTypeInfoHolder();
     this.typeResolver = typeResolver;
   }
 
@@ -657,7 +661,8 @@ private void generalJavaRead(
       Serializer serializer;
       TypeResolver typeResolver = this.typeResolver;
       if ((flags & CollectionFlags.IS_DECL_ELEMENT_TYPE) != CollectionFlags.IS_DECL_ELEMENT_TYPE) {
-        serializer = typeResolver.readTypeInfo(readContext, elementTypeInfoHolder).getSerializer();
+        serializer =
+            typeResolver.readTypeInfo(readContext, elementTypeInfoReadHolder).getSerializer();
       } else {
         serializer = elemGenericType.getSerializer(typeResolver);
       }

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

@@ -717,14 +717,20 @@ public Collection onCollectionWrite(WriteContext writeContext, Set<?> value) {
 
   public static final class ConcurrentHashMapKeySetViewSerializer
       extends CollectionSerializer<ConcurrentHashMap.KeySetView> {
+    // Compatible reads may cache remote schema metadata; local writes retain the original local
+    // holders and must not reuse read state.
     private final TypeInfoHolder mapTypeInfoHolder;
+    private final TypeInfoHolder mapTypeInfoReadHolder;
     private final TypeInfoHolder valueTypeInfoHolder;
+    private final TypeInfoHolder valueTypeInfoReadHolder;
 
     public ConcurrentHashMapKeySetViewSerializer(
         TypeResolver typeResolver, Class<ConcurrentHashMap.KeySetView> type) {
       super(typeResolver, type, false);
       mapTypeInfoHolder = typeResolver.nilTypeInfoHolder();
+      mapTypeInfoReadHolder = typeResolver.nilTypeInfoHolder();
       valueTypeInfoHolder = typeResolver.nilTypeInfoHolder();
+      valueTypeInfoReadHolder = typeResolver.nilTypeInfoHolder();
     }
 
     @Override
@@ -735,8 +741,8 @@ public void write(WriteContext writeContext, ConcurrentHashMap.KeySetView value)
 
     @Override
     public ConcurrentHashMap.KeySetView read(ReadContext readContext) {
-      ConcurrentHashMap map = (ConcurrentHashMap) readContext.readRef(mapTypeInfoHolder);
-      Object value = readContext.readRef(valueTypeInfoHolder);
+      ConcurrentHashMap map = (ConcurrentHashMap) readContext.readRef(mapTypeInfoReadHolder);
+      Object value = readContext.readRef(valueTypeInfoReadHolder);
       readContext.reserveGraphMemory(KEY_SET_VIEW_OWNER_BYTES);
       return map.keySet(value);
     }

java/fory-core/src/test/java/org/apache/fory/meta/NativeTypeDefEncoderTest.java

@@ -681,6 +681,47 @@ public void testAbstractParentClass() {
     }
   }
 
+  @Test
+  public void testUnregisteredParentLayer() {
+    Fory writer = registeredChildFory();
+    Fory reader = registeredChildFory();
+    writer.register(ChildClass.class, 100);
+    reader.register(ChildClass.class, 100);
+    ClassResolver readerResolver = (ClassResolver) reader.getTypeResolver();
+    Assert.assertFalse(readerResolver.isRegisteredByName(BaseAbstractClass.class.getName()));
+
+    ChildClass child = new ChildClass();
+    child.setId("123");
+    child.setName("test");
+    ChildClass decoded = (ChildClass) reader.deserialize(writer.serialize(child));
+    Assert.assertEquals(decoded.getId(), "123");
+    Assert.assertEquals(decoded.getName(), "test");
+
+    TypeDef typeDef = TypeDef.buildTypeDef(writer.getTypeResolver(), ChildClass.class);
+    TypeDef decodedTypeDef =
+        TypeDef.readTypeDef(readerResolver, MemoryBuffer.fromByteArray(typeDef.getEncoded()));
+    FieldInfo parentField =
+        decodedTypeDef.getFieldsInfo().stream()
+            .filter(field -> field.getFieldName().equals("id"))
+            .findFirst()
+            .orElseThrow(AssertionError::new);
+    Assert.assertEquals(parentField.getDefinedClass(), BaseAbstractClass.class.getName());
+    Assert.assertEquals(decodedTypeDef.getClassSpec().type, ChildClass.class);
+  }
+
+  private static Fory registeredChildFory() {
+    return Fory.builder()
+        .withXlang(false)
+        .withRefTracking(false)
+        .withMetaShare(true)
+        .withScopedMetaShare(true)
+        .withCompatible(true)
+        .withDeserializeUnknownClass(false)
+        .withAsyncCompilation(false)
+        .requireClassRegistration(true)
+        .build();
+  }
+
   @Data
   public abstract static class BaseAbstractClass {
     private String id;