[Bug] Concurrent deserialization with $ref references produces null fields in FastJson2Serialization

### Pre-check

- [x] I am sure that all the content I provide is in English.

### Search before asking

- [x] I had searched in the [issues](https://github.com/apache/dubbo/issues?q=is%3Aissue) and found no similar issues.

### Apache Dubbo Component

Java SDK (apache/dubbo)

### Dubbo Version

## Environment

- Dubbo: 3.3.6
- fastjson2: 2.0.61 / 2.0.62
- JDK: 25.0.3 (also verified on JDK 17/21)
- OS: Windows / Linux
- serialization=fastjson2

### Steps to reproduce this issue

## Test Code

```java
package com.example.dubbo.bug;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.beans.factory.ScopeBeanFactory;
import org.apache.dubbo.common.serialize.fastjson2.FastJson2Serialization;
import org.apache.dubbo.common.serialize.ObjectInput;
import org.apache.dubbo.common.serialize.ObjectOutput;
import org.apache.dubbo.common.utils.SerializeSecurityManager;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.jupiter.api.Assertions.assertTrue;

/**
* Reproduces a bug in Dubbo 3.3.6 FastJson2Serialization where concurrent
* deserialization of bytes containing $ref references produces null fields.
*
*

Self-contained test, no business classes required.
*
*

Environment: Dubbo 3.3.6, fastjson2 2.0.61+, JDK 17+.
*/
public class DubboFastjson2ConcurrentRefBugTest {

// ==================== Self-contained POJOs (no business classes) ====================

public static class Outer implements Serializable {
private List items;
public List getItems() { return items; }
public void setItems(List items) { this.items = items; }
}

public static class Inner implements Serializable {
private String name;
private List ids;
public String getName() { return name; }
public void setName(String name) { this.name = name; }
public List getIds() { return ids; }
public void setIds(List ids) { this.ids = ids; }
}

// ==================== Environment setup ====================

/**
* Simulate production: STRICT mode + register allowlist.
* Production Dubbo defaults to STRICT mode; POJO classes must be added to the allowlist
* before they can be deserialized.
*/
@BeforeAll
static void simulateProductionEnvironment() throws Exception {
FrameworkModel frameworkModel = FrameworkModel.defaultModel();
ScopeBeanFactory beanFactory = frameworkModel.getBeanFactory();
SerializeSecurityManager securityManager = beanFactory.getBean(SerializeSecurityManager.class);

// Simulate production: checkStatus=null (allow deserialization, skip STRICT check)
Field checkStatusField = SerializeSecurityManager.class.getDeclaredField("checkStatus");
checkStatusField.setAccessible(true);
checkStatusField.set(securityManager, null);

// Register POJO classes to the allowlist
securityManager.addToAllowed(Outer.class.getName());
securityManager.addToAllowed(Inner.class.getName());
securityManager.addToAllowed("java.util.ImmutableCollections$ListN");
securityManager.addToAllowed("java.util.ImmutableCollections$List12");
securityManager.addToAllowed("java.util.ArrayList");
securityManager.addToAllowed("java.lang.Long");
securityManager.addToAllowed("java.lang.String");
}

// ==================== Helpers ====================

/**
* Build test data: one Outer containing 20 Inner objects, all referencing the same
* List.of() singleton for their ids field.
* During serialization, only items[0].ids is written as the actual value;
* items[1..19].ids are written as $ref references.
*/
private static byte[] buildDubboBytes() throws Exception {
Outer outer = new Outer();
List items = new ArrayList<>();
List sharedIds = List.of(); // shared singleton → produces $ref
for (int i = 0; i < 20; i++) {
Inner inner = new Inner();
inner.setName("item-" + i);
inner.setIds(sharedIds);
items.add(inner);
}
outer.setItems(items);

FastJson2Serialization serialization = new FastJson2Serialization();
URL dubboUrl = URL.valueOf("dubbo://127.0.0.1:20880/DemoService?serialization=fastjson2");
ByteArrayOutputStream baos = new ByteArrayOutputStream();
ObjectOutput output = serialization.serialize(dubboUrl, baos);
output.writeObject(outer);
output.flushBuffer();
return baos.toByteArray();
}

/**
* Clear fastjson2 ObjectReaderProvider caches via reflection to force a cache miss.
*/
@SuppressWarnings("unchecked")
private static void clearObjectReaderCache() throws Exception {
com.alibaba.fastjson2.reader.ObjectReaderProvider provider =
com.alibaba.fastjson2.JSONFactory.getDefaultObjectReaderProvider();

for (String fieldName : new String[]{"cache", "cacheFieldBased"}) {
Field f = com.alibaba.fastjson2.reader.ObjectReaderProvider.class.getDeclaredField(fieldName);
f.setAccessible(true);
Object cache = f.get(provider);
if (cache instanceof Map) {
((Map) cache).clear();
}
}

Field readerCacheField =
com.alibaba.fastjson2.reader.ObjectReaderProvider.class.getDeclaredField("readerCache");
readerCacheField.setAccessible(true);
readerCacheField.set(null, null);
}

/**
* Deserialize using Dubbo FastJson2Serialization and count the number of null fields.
*/
private static int countNullFields(byte[] data) throws Exception {
FastJson2Serialization serialization = new FastJson2Serialization();
URL dubboUrl = URL.valueOf("dubbo://127.0.0.1:20880/DemoService?serialization=fastjson2");
ObjectInput input = serialization.deserialize(dubboUrl, new ByteArrayInputStream(data));
Object result = input.readObject(Outer.class);
if (!(result instanceof Outer outer)) {
return -1;
}
List items = outer.getItems();
if (items == null) {
return -1;
}
int nullCount = 0;
for (Inner inner : items) {
if (inner == null || inner.getIds() == null) {
nullCount++;
}
}
return nullCount;
}

// ==================== Test cases ====================

/**
* Baseline: single-thread deserialization produces no null.
*/
@Test
void baseline_singleThread_noNull() throws Exception {
byte[] data = buildDubboBytes();
int nullCount = countNullFields(data);
System.out.println("[Baseline] single-thread null fields: " + nullCount);
assertTrue(nullCount == 0,
"Single-thread deserialization should not produce null fields, actual: " + nullCount);
}

/**
* Core reproduction test: platform threads + CyclicBarrier + cache cleared each round.
*
*

All platform threads are pre-created and synchronized with a CyclicBarrier so they
* start deserializing at the same time. The ObjectReader cache is cleared each round
* to force a cache miss and trigger concurrent createObjectReader.
*
*

The bug is probabilistic; typically reproduced within 3-5 rounds.
*/
@Test
void reproduce_platformThread_concurrentRefNpe() throws Exception {
byte[] data = buildDubboBytes();

int rounds = 10;
int threadCount = 200;
AtomicInteger totalNullTasks = new AtomicInteger(0);
AtomicInteger totalNullFields = new AtomicInteger(0);
AtomicInteger totalSuccessTasks = new AtomicInteger(0);
List details = Collections.synchronizedList(new ArrayList<>());

System.out.println("==================== Platform Threads + CyclicBarrier + Dubbo FastJson2 Concurrent $ref NPE Reproduction ====================");
System.out.println("Dubbo: " + org.apache.dubbo.common.Version.getVersion());
System.out.println("fastjson2: " + com.alibaba.fastjson2.JSON.VERSION);
System.out.println("JDK: " + System.getProperty("java.version"));
System.out.println("Rounds: " + rounds + ", threads per round: " + threadCount);
System.out.println("ObjectReader cache cleared each round to force cache miss");
System.out.println();

for (int roundIdx = 1; roundIdx <= rounds; roundIdx++) {
final int round = roundIdx;
clearObjectReaderCache();

AtomicInteger roundNullTasks = new AtomicInteger(0);
AtomicInteger roundNullFields = new AtomicInteger(0);
AtomicInteger roundSuccessTasks = new AtomicInteger(0);

CyclicBarrier barrier = new CyclicBarrier(threadCount);
CountDownLatch endLatch = new CountDownLatch(threadCount);

Thread[] threads = new Thread[threadCount];
for (int t = 0; t < threadCount; t++) {
final int taskId = t;
threads[t] = new Thread(() -> {
try {
barrier.await(); // all threads start together
int nullCount = countNullFields(data);
if (nullCount > 0) {
roundNullTasks.incrementAndGet();
roundNullFields.addAndGet(nullCount);
if (details.size() < 30) {
details.add("round" + round + "-task[" + taskId + "] null=" + nullCount);
}
} else {
roundSuccessTasks.incrementAndGet();
}
} catch (Throwable e) {
if (details.size() < 10) {
details.add("round" + round + "-task[" + taskId + "] exception: " + e);
}
} finally {
endLatch.countDown();
}
}, "platform-worker-" + t);
}

for (Thread thread : threads) {
thread.start();
}
endLatch.await();

int nt = roundNullTasks.get();
totalNullTasks.addAndGet(nt);
totalNullFields.addAndGet(roundNullFields.get());
totalSuccessTasks.addAndGet(roundSuccessTasks.get());
System.out.println("Round " + round + ": success=" + roundSuccessTasks.get()
+ " nullTasks=" + nt + " nullFields=" + roundNullFields.get());
}

System.out.println();
System.out.println("========== Summary ==========");
System.out.println("Total tasks: " + (rounds * threadCount));
System.out.println("Success tasks: " + totalSuccessTasks.get());
System.out.println("Null tasks: " + totalNullTasks.get());
System.out.println("Null fields: " + totalNullFields.get());

if (!details.isEmpty()) {
System.out.println();
System.out.println("--- null details (first 30) ---");
details.forEach(System.out::println);
}

System.out.println();
if (totalNullTasks.get() > 0) {
System.out.println("Conclusion: Reproduced! Dubbo FastJson2 concurrent deserialization of $ref references has a bug");
} else {
System.out.println("Conclusion: Not reproduced this run (the bug is probabilistic, please run a few more times)");
}
}

/**
* Control test: concurrent but cache hit (cache not cleared), confirms no null.
* Proves the root cause is concurrent createObjectReader on cache miss.
*/
@Test
void control_concurrentCacheHit_noNull() throws Exception {
byte[] data = buildDubboBytes();

// Warm up the cache: deserialize once in a single thread
countNullFields(data);

int rounds = 5;
int threadCount = 200;
AtomicInteger totalNullTasks = new AtomicInteger(0);

System.out.println("==================== Control: concurrent + cache hit ====================");
System.out.println("After warming up the cache, concurrent deserialization should not produce null");

for (int roundIdx = 1; roundIdx <= rounds; roundIdx++) {
final int round = roundIdx;
AtomicInteger roundNullTasks = new AtomicInteger(0);
AtomicInteger roundSuccessTasks = new AtomicInteger(0);

CyclicBarrier barrier = new CyclicBarrier(threadCount);
CountDownLatch endLatch = new CountDownLatch(threadCount);

Thread[] threads = new Thread[threadCount];
for (int t = 0; t < threadCount; t++) {
threads[t] = new Thread(() -> {
try {
barrier.await();
int nullCount = countNullFields(data);
if (nullCount > 0) {
roundNullTasks.incrementAndGet();
} else {
roundSuccessTasks.incrementAndGet();
}
} catch (Throwable e) {
// ignore
} finally {
endLatch.countDown();
}
});
}

for (Thread thread : threads) {
thread.start();
}
endLatch.await();

totalNullTasks.addAndGet(roundNullTasks.get());
System.out.println("Round " + round + ": success=" + roundSuccessTasks.get()
+ " nullTasks=" + roundNullTasks.get());
}

System.out.println();
System.out.println("Total null tasks: " + totalNullTasks.get());
assertTrue(totalNullTasks.get() == 0,
"Cache hit should not produce null, actual: " + totalNullTasks.get());
}
}
```

### What you expected to happen

## Summary

When using Dubbo `FastJson2Serialization` to deserialize JSONB bytes that contain `$ref` references, some fields of the deserialized result become `null` under concurrent load (platform threads synchronized with `CyclicBarrier`). The issue is **probabilistic** — deserializing the same bytes multiple times succeeds in most cases, but a small number of attempts produce null fields.

## Root Cause

The `FastJson2ObjectInput` constructor calls `fastjson2CreatorManager.setCreator(classLoader)` on every invocation, which sets a fastjson2 **ThreadLocal** via `JSONFactory.setContextReaderCreator()`. Combined with fastjson2's `ObjectReaderProvider.getObjectReaderInternal`, which on a cache miss **first calls `createObjectReader` then `putIfAbsent`** (a non-atomic check-then-act), concurrent cache misses across multiple threads produce corrupted `ObjectReader` instances. When `$ref` references are subsequently resolved via `fieldReader.accept()`, the write fails silently and the field remains `null`.

### Bug Chain

```
1. Multiple objects reference the same shared singleton (e.g. List.of())
→ During serialization, fastjson2 writes the actual value for items[0].ids
and $ref references for items[1..n].ids

2. Dubbo FastJson2ObjectInput constructor calls setCreator
→ JSONFactory.setContextReaderCreator(creator) → ThreadLocal.set(creator)

3. Under concurrent deserialization, an ObjectReader cache miss occurs for some type
→ ObjectReaderProvider.getObjectReaderInternal line 1151: if (objectReader == null)
→ Multiple threads enter createObjectReader simultaneously (non-atomic, not thread-safe)

4. Concurrent createObjectReader produces a corrupted ObjectReader
(FieldReader array is incomplete or incorrect)

5. handleResolveTasks resolves $ref, but fieldReader.accept() fails
→ the field keeps its initial value (null)
```

## Trigger Conditions

1. **Serialization produces `$ref` references**: multiple objects reference the same shared singleton (e.g. `List.of()`, `Collections.emptyList()`, or the same List instance referenced by multiple fields)
2. **Dubbo FastJson2Serialization deserialization**: goes through `FastJson2ObjectInput` → `setCreator`
3. **Concurrency + ObjectReader cache miss**: cold start or first deserialization of a given type
4. **Platform threads synchronized with `CyclicBarrier`** are sufficient to reproduce (virtual threads trigger it more easily due to faster startup, but are not required)

## Reproduce

### Test Output

```
==================== Platform Threads + CyclicBarrier ====================
Dubbo: 3.3.6
fastjson2: 2.0.61
JDK: 25.0.3
Rounds: 10, threads per round: 200
ObjectReader cache is cleared each round to force cache miss

Round 1: success=200 nullTasks=0 nullFields=0
Round 2: success=198 nullTasks=2 nullFields=38
Round 3: success=198 nullTasks=2 nullFields=38
Round 4: success=197 nullTasks=3 nullFields=57
Round 5: success=198 nullTasks=2 nullFields=38
Round 6: success=198 nullTasks=2 nullFields=38
Round 7: success=199 nullTasks=1 nullFields=19
Round 8: success=199 nullTasks=1 nullFields=19
Round 9: success=198 nullTasks=2 nullFields=38
Round 10: success=197 nullTasks=3 nullFields=47

Total tasks: 2000
Success tasks: 1982
Null tasks: 18
Null fields: 332
Conclusion: Reproduced! Dubbo FastJson2 concurrent deserialization of $ref references has a bug
```

**Null pattern**: out of 20 items, `items[0]` is always correct (actual value, no `$ref`), while `items[1..19]` are all null (`$ref` resolution failed).

### Test Cases

| Test | Scenario | Result | Proves |
|------|----------|--------|--------|
| `baseline_singleThread_noNull` | single thread | 0 null | bytes are correct |
| `reproduce_platformThread_concurrentRefNpe` | platform threads + CyclicBarrier + cache miss | **18/2000 null** | reproduces the bug |
| `control_concurrentCacheHit_noNull` | concurrent + cache hit | 0 null | root cause is concurrent cache miss |

## Suggested Fix

Either:
- Dubbo's `FastJson2ObjectInput` constructor should add concurrency protection around the `setCreator` call, or
- fastjson2's `ObjectReaderProvider.getObjectReaderInternal` cache-miss path should be atomized using `ConcurrentHashMap.computeIfAbsent` (to prevent multiple threads from concurrently invoking `createObjectReader`).

## Workaround

At the application layer, eliminate the source of `$ref` references: replace shared singletons such as `List.of()` / `Collections.emptyList()` with `new ArrayList<>()`.

### Anything else

_No response_

### Do you have a (mini) reproduction demo?

- [x] Yes, I have a minimal reproduction demo to help resolve this issue more effectively!

### Are you willing to submit a pull request to fix on your own?

- [x] Yes I am willing to submit a pull request on my own!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)

Start by running the supplied DubboFastjson2ConcurrentRefBugTest reproducer, including its baseline_singleThread_noNull and reproduce_platformThread_concurrentRefNpe cases. Trace FastJson2Serialization deserialization and the fastjson2 ObjectReader cache behavior used by countNullFields. Done means concurrent $ref deserialization produces no null fields while the single-thread baseline still passes.

Written by the indexing model from the issue text.