You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by ji...@apache.org on 2026/09/28 03:09:16 UTC
(incubator-seata) branch 2.x updated: bugfix: guard concurrent fastjson2 JSONB deserialization (#8278)
This is an automated email from the ASF dual-hosted git repository.
funky-eyes pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/incubator-seata.git
The following commit(s) were added to refs/heads/2.x by this push:
new f51f27a978 bugfix: guard concurrent fastjson2 JSONB deserialization (#8278)
f51f27a978 is described below
commit f51f27a9787273a4a525a811ae96e226e2bb99f5
Author: jimin <sl...@163.com>
AuthorDate: Mon Sep 28 11:09:11 2026 +0800
bugfix: guard concurrent fastjson2 JSONB deserialization (#8278)
---
changes/en-us/2.x.md | 2 +
changes/zh-cn/2.x.md | 2 +
.../seata/common/json/Fastjson2JsonbParser.java | 41 ++++++++
.../common/json/Fastjson2JsonbParserTest.java | 115 +++++++++++++++++++++
.../undo/parser/Fastjson2UndoLogParser.java | 3 +-
.../Fastjson2ConcurrentRefDeserializationTest.java | 42 +++++---
.../Fastjson2Serializer.java | 3 +-
.../Fastjson2ConcurrentRefDeserializationTest.java | 41 +++++---
8 files changed, 214 insertions(+), 35 deletions(-)
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
index 5433973380..692bc536d3 100644
--- a/changes/en-us/2.x.md
+++ b/changes/en-us/2.x.md
@@ -26,6 +26,8 @@ Add changes here for all PR submitted to the 2.x branch.
### bugfix:
+- [[#8278](https://github.com/apache/incubator-seata/pull/8278)] guard concurrent fastjson2 JSONB decoding to preserve RPC and undo-log references
+
- [[#8138](https://github.com/apache/incubator-seata/pull/8138)] fix TCC fence cleanup deleting in-progress/unexpired sibling branch records
- [[#8191](https://github.com/apache/incubator-seata/pull/8191)] fix NetUtil filtered lookups bypassing ignoredInterfaces and preferredNetworks once the default address is cached
- [[#8145](https://github.com/apache/incubator-seata/pull/8145)] fix global lock batch acquire false-failure on Dameng(DM)
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
index 8b482bc891..a61b772eae 100644
--- a/changes/zh-cn/2.x.md
+++ b/changes/zh-cn/2.x.md
@@ -26,6 +26,8 @@
### bugfix:
+- [[#8278](https://github.com/apache/incubator-seata/pull/8278)] 保护 Fastjson2 JSONB 并发解码，避免 RPC 与 UndoLog 引用字段丢失
+
- [[#8138](https://github.com/apache/incubator-seata/pull/8138)] 修复 TCC fence 清理时误删同一全局事务中仍在进行(TRIED)或未过期的分支记录的问题
- [[#8191](https://github.com/apache/incubator-seata/pull/8191)] 修复默认本地地址缓存后 NetUtil 参数化查询绕过 ignoredInterfaces 和 preferredNetworks 的问题
- [#8145](https://github.com/apache/incubator-seata/pull/8145) 修复达梦(DM)数据库下全局锁批量获取被误判为失败的问题
diff --git a/json-common/json-common-core/src/main/java/org/apache/seata/common/json/Fastjson2JsonbParser.java b/json-common/json-common-core/src/main/java/org/apache/seata/common/json/Fastjson2JsonbParser.java
new file mode 100644
index 0000000000..3ef60812fc
--- /dev/null
+++ b/json-common/json-common-core/src/main/java/org/apache/seata/common/json/Fastjson2JsonbParser.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.seata.common.json;
+
+import com.alibaba.fastjson2.JSONB;
+import com.alibaba.fastjson2.JSONReader;
+import com.alibaba.fastjson2.filter.Filter;
+
+/**
+ * Serializes Seata JSONB decoding while fastjson2 shares mutable reference-resolution state.
+ * Reader warmup alone does not protect the cached reference paths used during parsing.
+ */
+public final class Fastjson2JsonbParser {
+
+ private Fastjson2JsonbParser() {}
+
+ // Both entry points must use the same monitor: RPC and undo-log parsers share fastjson2 caches.
+ // Keep the entire parse guarded, including the deferred reference-resolution phase.
+ public static synchronized <T> T parseObject(byte[] bytes, Class<T> type, JSONReader.Feature... features) {
+ return JSONB.parseObject(bytes, type, features);
+ }
+
+ public static synchronized <T> T parseObject(
+ byte[] bytes, Class<T> type, Filter filter, JSONReader.Feature... features) {
+ return JSONB.parseObject(bytes, type, filter, features);
+ }
+}
diff --git a/json-common/json-common-core/src/test/java/org/apache/seata/common/json/Fastjson2JsonbParserTest.java b/json-common/json-common-core/src/test/java/org/apache/seata/common/json/Fastjson2JsonbParserTest.java
new file mode 100644
index 0000000000..25f4ddd4fe
--- /dev/null
+++ b/json-common/json-common-core/src/test/java/org/apache/seata/common/json/Fastjson2JsonbParserTest.java
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.seata.common.json;
+
+import com.alibaba.fastjson2.JSONB;
+import com.alibaba.fastjson2.JSONReader;
+import com.alibaba.fastjson2.JSONWriter;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class Fastjson2JsonbParserTest {
+
+ @Test
+ public void concurrentEntryPointsPreserveReferences() throws Exception {
+ byte[] bytes = payload();
+ ExecutorService executor = Executors.newFixedThreadPool(16);
+ CountDownLatch start = new CountDownLatch(1);
+ List<Future<?>> futures = new ArrayList<>();
+ try {
+ for (int i = 0; i < 16; i++) {
+ boolean filtered = i % 2 == 0;
+ futures.add(executor.submit(() -> {
+ assertTrue(start.await(30, TimeUnit.SECONDS));
+ for (int j = 0; j < 25; j++) {
+ SharedReferences value = filtered
+ ? (SharedReferences) Fastjson2JsonbParser.parseObject(
+ bytes,
+ Object.class,
+ JSONReader.autoTypeFilter(SharedReferences.class),
+ JSONReader.Feature.FieldBased)
+ : Fastjson2JsonbParser.parseObject(
+ bytes, SharedReferences.class, JSONReader.Feature.FieldBased);
+ assertReferences(value);
+ }
+ return null;
+ }));
+ }
+ start.countDown();
+ for (Future<?> future : futures) {
+ future.get(30, TimeUnit.SECONDS);
+ }
+ } finally {
+ start.countDown();
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
+ }
+ }
+
+ @Test
+ public void malformedInputDoesNotPreventSubsequentDecoding() throws Exception {
+ assertThrows(
+ RuntimeException.class,
+ () -> Fastjson2JsonbParser.parseObject(
+ new byte[0], SharedReferences.class, JSONReader.Feature.FieldBased));
+ // Use a different thread so a leaked lock cannot be hidden by reentrancy.
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ SharedReferences value = executor.submit(() -> Fastjson2JsonbParser.parseObject(
+ payload(), SharedReferences.class, JSONReader.Feature.FieldBased))
+ .get(30, TimeUnit.SECONDS);
+ assertReferences(value);
+ } finally {
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
+ }
+ }
+
+ private static byte[] payload() {
+ SharedReferences value = new SharedReferences();
+ value.first = new ArrayList<>(Arrays.asList("seata"));
+ value.second = value.first;
+ return JSONB.toBytes(
+ value,
+ JSONWriter.Feature.FieldBased,
+ JSONWriter.Feature.WriteClassName,
+ JSONWriter.Feature.ReferenceDetection);
+ }
+
+ private static void assertReferences(SharedReferences value) {
+ assertEquals(Arrays.asList("seata"), value.first);
+ assertSame(value.first, value.second);
+ }
+
+ public static class SharedReferences {
+ private List<String> first;
+ private List<String> second;
+ }
+}
diff --git a/rm-datasource/src/main/java/org/apache/seata/rm/datasource/undo/parser/Fastjson2UndoLogParser.java b/rm-datasource/src/main/java/org/apache/seata/rm/datasource/undo/parser/Fastjson2UndoLogParser.java
index 7849d10ea4..4571edcb72 100644
--- a/rm-datasource/src/main/java/org/apache/seata/rm/datasource/undo/parser/Fastjson2UndoLogParser.java
+++ b/rm-datasource/src/main/java/org/apache/seata/rm/datasource/undo/parser/Fastjson2UndoLogParser.java
@@ -20,6 +20,7 @@ import com.alibaba.fastjson2.JSONB;
import com.alibaba.fastjson2.JSONReader;
import com.alibaba.fastjson2.JSONWriter;
import org.apache.seata.common.executor.Initialize;
+import org.apache.seata.common.json.Fastjson2JsonbParser;
import org.apache.seata.common.json.Fastjson2ObjectReaderWarmup;
import org.apache.seata.common.loader.LoadLevel;
import org.apache.seata.rm.datasource.sql.struct.Field;
@@ -93,6 +94,6 @@ public class Fastjson2UndoLogParser implements UndoLogParser, Initialize {
@Override
public BranchUndoLog decode(byte[] bytes) {
- return JSONB.parseObject(bytes, BranchUndoLog.class, jsonReaderFeature);
+ return Fastjson2JsonbParser.parseObject(bytes, BranchUndoLog.class, jsonReaderFeature);
}
}
diff --git a/rm-datasource/src/test/java/org/apache/seata/rm/datasource/undo/parser/Fastjson2ConcurrentRefDeserializationTest.java b/rm-datasource/src/test/java/org/apache/seata/rm/datasource/undo/parser/Fastjson2ConcurrentRefDeserializationTest.java
index 7bc3768eff..d578156036 100644
--- a/rm-datasource/src/test/java/org/apache/seata/rm/datasource/undo/parser/Fastjson2ConcurrentRefDeserializationTest.java
+++ b/rm-datasource/src/test/java/org/apache/seata/rm/datasource/undo/parser/Fastjson2ConcurrentRefDeserializationTest.java
@@ -52,12 +52,15 @@ public class Fastjson2ConcurrentRefDeserializationTest {
public static void main(String[] args) throws Exception {
Fastjson2UndoLogParser parser =
(Fastjson2UndoLogParser) EnhancedServiceLoader.load(UndoLogParser.class, Fastjson2UndoLogParser.NAME);
+ Fastjson2UndoLogParser otherParser = new Fastjson2UndoLogParser();
+ otherParser.init();
+ Fastjson2UndoLogParser[] parsers = {parser, otherParser};
byte[] bytes = parser.encode(referenceHeavyUndoLog());
- int nullTasks =
- runConcurrentStress(parser, bytes, Integer.getInteger("seata.fastjson2.concurrentRef.threads", 200));
- if (nullTasks > 0) {
- throw new AssertionError("Concurrent deserialization dropped $ref fields: " + nullTasks);
+ int invalidTasks =
+ runConcurrentStress(parsers, bytes, Integer.getInteger("seata.fastjson2.concurrentRef.threads", 200));
+ if (invalidTasks > 0) {
+ throw new AssertionError("Concurrent deserialization lost reference identity: " + invalidTasks);
}
}
@@ -92,41 +95,46 @@ public class Fastjson2ConcurrentRefDeserializationTest {
return tableRecords;
}
- private static int countNullRefFields(BranchUndoLog branchUndoLog) {
+ private static int countInvalidRefFields(BranchUndoLog branchUndoLog) {
if (branchUndoLog == null || branchUndoLog.getSqlUndoLogs() == null) {
return 1;
}
- int nullCount = 0;
+ int invalidCount = 0;
+ TableRecords sharedImage = branchUndoLog.getSqlUndoLogs().get(0).getBeforeImage();
for (SQLUndoLog sqlUndoLog : branchUndoLog.getSqlUndoLogs()) {
if (sqlUndoLog == null) {
- nullCount++;
+ invalidCount++;
continue;
}
if (sqlUndoLog.getBeforeImage() == null
+ || sqlUndoLog.getBeforeImage() != sharedImage
|| sqlUndoLog.getBeforeImage().getRows() == null) {
- nullCount++;
+ invalidCount++;
}
- if (sqlUndoLog.getAfterImage() == null || sqlUndoLog.getAfterImage().getRows() == null) {
- nullCount++;
+ if (sqlUndoLog.getAfterImage() == null
+ || sqlUndoLog.getAfterImage() != sharedImage
+ || sqlUndoLog.getAfterImage().getRows() == null) {
+ invalidCount++;
}
}
- return nullCount;
+ return invalidCount;
}
- private static int runConcurrentStress(Fastjson2UndoLogParser parser, byte[] bytes, int threadCount)
+ private static int runConcurrentStress(Fastjson2UndoLogParser[] parsers, byte[] bytes, int threadCount)
throws Exception {
AtomicReference<Throwable> failure = new AtomicReference<>();
CyclicBarrier barrier = new CyclicBarrier(threadCount);
CountDownLatch endLatch = new CountDownLatch(threadCount);
- AtomicInteger nullTasks = new AtomicInteger();
+ AtomicInteger invalidTasks = new AtomicInteger();
for (int i = 0; i < threadCount; i++) {
+ Fastjson2UndoLogParser parser = parsers[i % parsers.length];
Thread thread = new Thread(
() -> {
try {
barrier.await(CONCURRENT_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- if (countNullRefFields(parser.decode(bytes)) > 0) {
- nullTasks.incrementAndGet();
+ if (countInvalidRefFields(parser.decode(bytes)) > 0) {
+ invalidTasks.incrementAndGet();
}
} catch (Throwable throwable) {
failure.compareAndSet(null, throwable);
@@ -144,12 +152,14 @@ public class Fastjson2ConcurrentRefDeserializationTest {
throw new AssertionError("Concurrent deserialization failed", failure.get());
}
- return nullTasks.get();
+ return invalidTasks.get();
}
private static void assertChildProcessSucceeds() throws Exception {
Process process = new ProcessBuilder(
System.getProperty("java.home") + "/bin/java",
+ "-Dseata.fastjson2.concurrentRef.threads="
+ + Integer.getInteger("seata.fastjson2.concurrentRef.threads", 200),
"-cp",
System.getProperty("java.class.path"),
Fastjson2ConcurrentRefDeserializationTest.class.getName())
diff --git a/serializer/seata-serializer-fastjson2/src/main/java/org.apache.seata.serializer.fastjson2/Fastjson2Serializer.java b/serializer/seata-serializer-fastjson2/src/main/java/org.apache.seata.serializer.fastjson2/Fastjson2Serializer.java
index fcc8cd9c8b..e0179a9ad2 100644
--- a/serializer/seata-serializer-fastjson2/src/main/java/org.apache.seata.serializer.fastjson2/Fastjson2Serializer.java
+++ b/serializer/seata-serializer-fastjson2/src/main/java/org.apache.seata.serializer.fastjson2/Fastjson2Serializer.java
@@ -18,6 +18,7 @@ package org.apache.seata.serializer.fastjson2;
import com.alibaba.fastjson2.JSONB;
import org.apache.seata.common.executor.Initialize;
+import org.apache.seata.common.json.Fastjson2JsonbParser;
import org.apache.seata.common.json.Fastjson2ObjectReaderWarmup;
import org.apache.seata.common.loader.LoadLevel;
import org.apache.seata.core.serializer.Serializer;
@@ -38,7 +39,7 @@ public class Fastjson2Serializer implements Serializer, Initialize {
@Override
public <T> T deserialize(byte[] bytes) {
- return (T) JSONB.parseObject(
+ return (T) Fastjson2JsonbParser.parseObject(
bytes,
Object.class,
Fastjson2SerializerFactory.getInstance().getFilter(),
diff --git a/serializer/seata-serializer-fastjson2/src/test/java/org/apache/seata/serializer/fastjson2/Fastjson2ConcurrentRefDeserializationTest.java b/serializer/seata-serializer-fastjson2/src/test/java/org/apache/seata/serializer/fastjson2/Fastjson2ConcurrentRefDeserializationTest.java
index 2c04ad0896..44f902f3f8 100644
--- a/serializer/seata-serializer-fastjson2/src/test/java/org/apache/seata/serializer/fastjson2/Fastjson2ConcurrentRefDeserializationTest.java
+++ b/serializer/seata-serializer-fastjson2/src/test/java/org/apache/seata/serializer/fastjson2/Fastjson2ConcurrentRefDeserializationTest.java
@@ -47,12 +47,15 @@ public class Fastjson2ConcurrentRefDeserializationTest {
public static void main(String[] args) throws Exception {
Fastjson2Serializer serializer = new Fastjson2Serializer();
((Initialize) serializer).init();
+ Fastjson2Serializer otherSerializer = new Fastjson2Serializer();
+ otherSerializer.init();
+ Fastjson2Serializer[] serializers = {serializer, otherSerializer};
byte[] bytes = serializer.serialize(referenceHeavyMessage());
- int nullTasks = runConcurrentStress(
- serializer, bytes, Integer.getInteger("seata.fastjson2.concurrentRef.threads", 200));
- if (nullTasks > 0) {
- throw new AssertionError("Concurrent deserialization dropped $ref fields: " + nullTasks);
+ int invalidTasks = runConcurrentStress(
+ serializers, bytes, Integer.getInteger("seata.fastjson2.concurrentRef.threads", 200));
+ if (invalidTasks > 0) {
+ throw new AssertionError("Concurrent deserialization lost reference identity: " + invalidTasks);
}
}
@@ -70,41 +73,43 @@ public class Fastjson2ConcurrentRefDeserializationTest {
return message;
}
- private static int countNullRefFields(MergedWarpMessage message) {
+ private static int countInvalidRefFields(MergedWarpMessage message) {
if (message == null || message.msgs == null) {
return 1;
}
- int nullCount = 0;
+ int invalidCount = 0;
+ Object sharedList = ((BatchResultMessage) message.msgs.get(0)).getResultMessages();
for (AbstractMessage child : message.msgs) {
if (!(child instanceof BatchResultMessage)) {
- nullCount++;
+ invalidCount++;
continue;
}
BatchResultMessage batchResult = (BatchResultMessage) child;
- if (batchResult.getResultMessages() == null) {
- nullCount++;
+ if (batchResult.getResultMessages() == null || batchResult.getResultMessages() != sharedList) {
+ invalidCount++;
}
- if (batchResult.getMsgIds() == null) {
- nullCount++;
+ if (batchResult.getMsgIds() == null || batchResult.getMsgIds() != sharedList) {
+ invalidCount++;
}
}
- return nullCount;
+ return invalidCount;
}
- private static int runConcurrentStress(Fastjson2Serializer serializer, byte[] bytes, int threadCount)
+ private static int runConcurrentStress(Fastjson2Serializer[] serializers, byte[] bytes, int threadCount)
throws Exception {
AtomicReference<Throwable> failure = new AtomicReference<>();
CyclicBarrier barrier = new CyclicBarrier(threadCount);
CountDownLatch endLatch = new CountDownLatch(threadCount);
- AtomicInteger nullTasks = new AtomicInteger();
+ AtomicInteger invalidTasks = new AtomicInteger();
for (int i = 0; i < threadCount; i++) {
+ Fastjson2Serializer serializer = serializers[i % serializers.length];
Thread thread = new Thread(
() -> {
try {
barrier.await(CONCURRENT_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- if (countNullRefFields((MergedWarpMessage) serializer.deserialize(bytes)) > 0) {
- nullTasks.incrementAndGet();
+ if (countInvalidRefFields((MergedWarpMessage) serializer.deserialize(bytes)) > 0) {
+ invalidTasks.incrementAndGet();
}
} catch (Throwable throwable) {
failure.compareAndSet(null, throwable);
@@ -122,12 +127,14 @@ public class Fastjson2ConcurrentRefDeserializationTest {
throw new AssertionError("Concurrent deserialization failed", failure.get());
}
- return nullTasks.get();
+ return invalidTasks.get();
}
private static void assertChildProcessSucceeds() throws Exception {
Process process = new ProcessBuilder(
System.getProperty("java.home") + "/bin/java",
+ "-Dseata.fastjson2.concurrentRef.threads="
+ + Integer.getInteger("seata.fastjson2.concurrentRef.threads", 200),
"-cp",
System.getProperty("java.class.path"),
Fastjson2ConcurrentRefDeserializationTest.class.getName())
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org