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 5ba6a62e4e bugfix : fastjson2 JSONB concurrent ref deserialization
(#8160)
5ba6a62e4e is described below
commit 5ba6a62e4ec00b4f509d987604e87365514d7c11
Author: legendpei <[email protected]>
AuthorDate: Wed Jul 29 10:41:12 2026 +0800
bugfix : fastjson2 JSONB concurrent ref deserialization (#8160)
---
changes/en-us/2.x.md | 1 +
changes/zh-cn/2.x.md | 2 +
.../common/json/Fastjson2ObjectReaderWarmup.java | 50 ++++++
.../json/Fastjson2ObjectReaderWarmupTest.java | 31 ++--
.../undo/parser/Fastjson2UndoLogParser.java | 18 +++
.../Fastjson2ConcurrentRefDeserializationTest.java | 176 +++++++++++++++++++++
serializer/seata-serializer-fastjson2/pom.xml | 7 +-
.../Fastjson2Serializer.java | 10 +-
.../Fastjson2ConcurrentRefDeserializationTest.java | 154 ++++++++++++++++++
.../fastjson2/Fastjson2SerializerTest.java | 1 +
10 files changed, 432 insertions(+), 18 deletions(-)
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
index 93f46c532b..19c39363b2 100644
--- a/changes/en-us/2.x.md
+++ b/changes/en-us/2.x.md
@@ -27,6 +27,7 @@ Add changes here for all PR submitted to the 2.x branch.
- [[#8138](https://github.com/apache/incubator-seata/pull/8138)] fix TCC fence
cleanup deleting in-progress/unexpired sibling branch records
- [[#8145](https://github.com/apache/incubator-seata/pull/8145)] fix global
lock batch acquire false-failure on Dameng(DM)
- [[#8157](https://github.com/apache/incubator-seata/pull/8157)] fix Saga
auto-configuration being skipped on Spring Boot 4.x because of a hard
`DataSourceAutoConfiguration` class reference
+- [[#8160](https://github.com/apache/incubator-seata/pull/8160)] fix
fastjson2 JSONB concurrent ref deserialization
- [[#8179](https://github.com/apache/incubator-seata/pull/8179)] set
java.version to 17 for console module to fix compilation error
- [[#8170](https://github.com/apache/incubator-seata/pull/8170)] fix
ConfigTools encryption corrupting non-ASCII content on non-UTF-8 platforms
- [[#8182](https://github.com/apache/incubator-seata/issues/8182)] fix
registry preferred networks and ignored interfaces from Spring Boot
application.yml
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
index c37688bf1d..6192449bb6 100644
--- a/changes/zh-cn/2.x.md
+++ b/changes/zh-cn/2.x.md
@@ -27,6 +27,8 @@
- [[#8138](https://github.com/apache/incubator-seata/pull/8138)] 修复 TCC fence
清理时误删同一全局事务中仍在进行(TRIED)或未过期的分支记录的问题
- [#8145](https://github.com/apache/incubator-seata/pull/8145)
修复达梦(DM)数据库下全局锁批量获取被误判为失败的问题
- [[#8157](https://github.com/apache/incubator-seata/pull/8157)] 修复 Spring
Boot 4.x 下由于硬编码 `DataSourceAutoConfiguration` 类引用导致 Saga 自动配置被跳过的问题
+- [[#8160](https://github.com/apache/incubator-seata/pull/8160)] 修复 fastjson2
JSONB 并发反序列化问题
+
- [[#8179](https://github.com/apache/incubator-seata/pull/8179)] 将 console 模块的
java.version 设置为 17 以修复编译错误
- [[#8170](https://github.com/apache/incubator-seata/pull/8170)] 修复
ConfigTools 加密在非 UTF-8 默认字符集平台损坏非 ASCII 内容的问题
- [[#8182](https://github.com/apache/incubator-seata/issues/8182)] 修复 Spring
Boot application.yml 中 registry preferred networks 和 ignored interfaces 配置不生效的问题
diff --git
a/json-common/json-common-core/src/main/java/org/apache/seata/common/json/Fastjson2ObjectReaderWarmup.java
b/json-common/json-common-core/src/main/java/org/apache/seata/common/json/Fastjson2ObjectReaderWarmup.java
new file mode 100644
index 0000000000..1e9e2a2410
--- /dev/null
+++
b/json-common/json-common-core/src/main/java/org/apache/seata/common/json/Fastjson2ObjectReaderWarmup.java
@@ -0,0 +1,50 @@
+/*
+ * 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.JSONFactory;
+import com.alibaba.fastjson2.reader.ObjectReaderProvider;
+
+import java.util.Arrays;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Initializes fastjson2 object readers before concurrent deserialization
begins.
+ */
+public final class Fastjson2ObjectReaderWarmup {
+
+ private static final Lock WARMUP_LOCK = new ReentrantLock();
+
+ private Fastjson2ObjectReaderWarmup() {}
+
+ public static void warmup(Iterable<Class<?>> types) {
+ WARMUP_LOCK.lock();
+ try {
+ ObjectReaderProvider provider =
JSONFactory.getDefaultObjectReaderProvider();
+ for (Class<?> type : types) {
+ provider.getObjectReader(type, true);
+ }
+ } finally {
+ WARMUP_LOCK.unlock();
+ }
+ }
+
+ public static void warmup(Class<?>... types) {
+ warmup(Arrays.asList(types));
+ }
+}
diff --git
a/serializer/seata-serializer-fastjson2/src/main/java/org.apache.seata.serializer.fastjson2/Fastjson2Serializer.java
b/json-common/json-common-core/src/test/java/org/apache/seata/common/json/Fastjson2ObjectReaderWarmupTest.java
similarity index 52%
copy from
serializer/seata-serializer-fastjson2/src/main/java/org.apache.seata.serializer.fastjson2/Fastjson2Serializer.java
copy to
json-common/json-common-core/src/test/java/org/apache/seata/common/json/Fastjson2ObjectReaderWarmupTest.java
index f52ea6ae38..a8d24942fa 100644
---
a/serializer/seata-serializer-fastjson2/src/main/java/org.apache.seata.serializer.fastjson2/Fastjson2Serializer.java
+++
b/json-common/json-common-core/src/test/java/org/apache/seata/common/json/Fastjson2ObjectReaderWarmupTest.java
@@ -14,26 +14,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.seata.serializer.fastjson2;
+package org.apache.seata.common.json;
-import com.alibaba.fastjson2.JSONB;
-import org.apache.seata.common.loader.LoadLevel;
-import org.apache.seata.core.serializer.Serializer;
+import org.junit.jupiter.api.Test;
-@LoadLevel(name = "FASTJSON2")
-public class Fastjson2Serializer implements Serializer {
+import java.util.Arrays;
- @Override
- public <T> byte[] serialize(T t) {
- return JSONB.toBytes(t,
Fastjson2SerializerFactory.getInstance().getJsonWriterFeatureList());
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+public class Fastjson2ObjectReaderWarmupTest {
+
+ @Test
+ public void warmupAcceptsIterableTypes() {
+ assertThatCode(() ->
Fastjson2ObjectReaderWarmup.warmup(Arrays.<Class<?>>asList(String.class,
Integer.class)))
+ .doesNotThrowAnyException();
}
- @Override
- public <T> T deserialize(byte[] bytes) {
- return (T) JSONB.parseObject(
- bytes,
- Object.class,
- Fastjson2SerializerFactory.getInstance().getFilter(),
-
Fastjson2SerializerFactory.getInstance().getJsonReaderFeatureList());
+ @Test
+ public void warmupAcceptsVarargsTypes() {
+ assertThatCode(() -> Fastjson2ObjectReaderWarmup.warmup(String.class,
Integer.class))
+ .doesNotThrowAnyException();
}
}
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 d56ba16bc2..7849d10ea4 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,9 +20,17 @@ 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.Fastjson2ObjectReaderWarmup;
import org.apache.seata.common.loader.LoadLevel;
+import org.apache.seata.rm.datasource.sql.struct.Field;
+import org.apache.seata.rm.datasource.sql.struct.Row;
+import org.apache.seata.rm.datasource.sql.struct.TableRecords;
import org.apache.seata.rm.datasource.undo.BranchUndoLog;
+import org.apache.seata.rm.datasource.undo.SQLUndoLog;
import org.apache.seata.rm.datasource.undo.UndoLogParser;
+import org.apache.seata.sqlparser.SQLType;
+
+import java.util.ArrayList;
@LoadLevel(name = Fastjson2UndoLogParser.NAME)
public class Fastjson2UndoLogParser implements UndoLogParser, Initialize {
@@ -54,6 +62,16 @@ public class Fastjson2UndoLogParser implements
UndoLogParser, Initialize {
JSONWriter.Feature.WriteNameAsSymbol
};
+ Fastjson2ObjectReaderWarmup.warmup(
+ Object.class,
+ ArrayList.class,
+ BranchUndoLog.class,
+ SQLUndoLog.class,
+ SQLType.class,
+ TableRecords.class,
+ Row.class,
+ Field.class);
+
// SerialArray support: Fastjson2 with FieldBased and SupportAutoType
features
// can handle SerialArray serialization automatically through field
access
}
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
new file mode 100644
index 0000000000..7bc3768eff
--- /dev/null
+++
b/rm-datasource/src/test/java/org/apache/seata/rm/datasource/undo/parser/Fastjson2ConcurrentRefDeserializationTest.java
@@ -0,0 +1,176 @@
+/*
+ * 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.rm.datasource.undo.parser;
+
+import org.apache.seata.common.loader.EnhancedServiceLoader;
+import org.apache.seata.rm.datasource.sql.struct.Field;
+import org.apache.seata.rm.datasource.sql.struct.Row;
+import org.apache.seata.rm.datasource.sql.struct.TableRecords;
+import org.apache.seata.rm.datasource.undo.BranchUndoLog;
+import org.apache.seata.rm.datasource.undo.SQLUndoLog;
+import org.apache.seata.rm.datasource.undo.UndoLogParser;
+import org.apache.seata.sqlparser.SQLType;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+import java.sql.Types;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class Fastjson2ConcurrentRefDeserializationTest {
+
+ private static final long CONCURRENT_TEST_TIMEOUT_SECONDS = 30;
+
+ @Test
+ public void
concurrentDeserializeReferenceHeavyUndoLogDoesNotDropRefFields() throws
Exception {
+ int rounds =
Integer.getInteger("seata.fastjson2.concurrentRef.rounds", 3);
+ for (int round = 0; round < rounds; round++) {
+ assertChildProcessSucceeds();
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ Fastjson2UndoLogParser parser =
+ (Fastjson2UndoLogParser)
EnhancedServiceLoader.load(UndoLogParser.class, Fastjson2UndoLogParser.NAME);
+ 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);
+ }
+ }
+
+ private static BranchUndoLog referenceHeavyUndoLog() {
+ BranchUndoLog branchUndoLog = new BranchUndoLog();
+ branchUndoLog.setXid("127.0.0.1:8091:123456");
+ branchUndoLog.setBranchId(123456L);
+
+ TableRecords sharedImage = tableRecords();
+ List<SQLUndoLog> sqlUndoLogs = new ArrayList<>();
+ for (int i = 0; i < 20; i++) {
+ SQLUndoLog sqlUndoLog = new SQLUndoLog();
+ sqlUndoLog.setSqlType(SQLType.UPDATE);
+ sqlUndoLog.setTableName("ref_test");
+ sqlUndoLog.setBeforeImage(sharedImage);
+ sqlUndoLog.setAfterImage(sharedImage);
+ sqlUndoLogs.add(sqlUndoLog);
+ }
+ branchUndoLog.setSqlUndoLogs(sqlUndoLogs);
+ return branchUndoLog;
+ }
+
+ private static TableRecords tableRecords() {
+ TableRecords tableRecords = new TableRecords();
+ tableRecords.setTableName("ref_test");
+ List<Row> rows = new ArrayList<>();
+ Row row = new Row();
+ row.add(new Field("id", Types.INTEGER, 1));
+ row.add(new Field("name", Types.VARCHAR, "seata"));
+ rows.add(row);
+ tableRecords.setRows(rows);
+ return tableRecords;
+ }
+
+ private static int countNullRefFields(BranchUndoLog branchUndoLog) {
+ if (branchUndoLog == null || branchUndoLog.getSqlUndoLogs() == null) {
+ return 1;
+ }
+ int nullCount = 0;
+ for (SQLUndoLog sqlUndoLog : branchUndoLog.getSqlUndoLogs()) {
+ if (sqlUndoLog == null) {
+ nullCount++;
+ continue;
+ }
+ if (sqlUndoLog.getBeforeImage() == null
+ || sqlUndoLog.getBeforeImage().getRows() == null) {
+ nullCount++;
+ }
+ if (sqlUndoLog.getAfterImage() == null ||
sqlUndoLog.getAfterImage().getRows() == null) {
+ nullCount++;
+ }
+ }
+ return nullCount;
+ }
+
+ private static int runConcurrentStress(Fastjson2UndoLogParser parser,
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();
+ for (int i = 0; i < threadCount; i++) {
+ Thread thread = new Thread(
+ () -> {
+ try {
+ barrier.await(CONCURRENT_TEST_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ if (countNullRefFields(parser.decode(bytes)) > 0) {
+ nullTasks.incrementAndGet();
+ }
+ } catch (Throwable throwable) {
+ failure.compareAndSet(null, throwable);
+ } finally {
+ endLatch.countDown();
+ }
+ },
+ "fastjson2-undolog-ref-" + i);
+ thread.start();
+ }
+ if (!endLatch.await(CONCURRENT_TEST_TIMEOUT_SECONDS,
TimeUnit.SECONDS)) {
+ throw new AssertionError("Timed out waiting for concurrent
deserialization");
+ }
+ if (failure.get() != null) {
+ throw new AssertionError("Concurrent deserialization failed",
failure.get());
+ }
+
+ return nullTasks.get();
+ }
+
+ private static void assertChildProcessSucceeds() throws Exception {
+ Process process = new ProcessBuilder(
+ System.getProperty("java.home") + "/bin/java",
+ "-cp",
+ System.getProperty("java.class.path"),
+
Fastjson2ConcurrentRefDeserializationTest.class.getName())
+ .redirectErrorStream(true)
+ .start();
+ if (!process.waitFor(CONCURRENT_TEST_TIMEOUT_SECONDS,
TimeUnit.SECONDS)) {
+ process.destroyForcibly();
+ throw new AssertionError("Timed out waiting for child process");
+ }
+ if (process.exitValue() != 0) {
+ throw new AssertionError("Child process failed: " +
readOutput(process.getInputStream()));
+ }
+ }
+
+ private static String readOutput(InputStream inputStream) throws Exception
{
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ byte[] buffer = new byte[1024];
+ int length;
+ while ((length = inputStream.read(buffer)) != -1) {
+ output.write(buffer, 0, length);
+ }
+ return output.toString("UTF-8");
+ }
+}
diff --git a/serializer/seata-serializer-fastjson2/pom.xml
b/serializer/seata-serializer-fastjson2/pom.xml
index b586ecac99..7a36329a5f 100644
--- a/serializer/seata-serializer-fastjson2/pom.xml
+++ b/serializer/seata-serializer-fastjson2/pom.xml
@@ -42,10 +42,15 @@
<artifactId>seata-core</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>json-common-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>com.alibaba.fastjson2</groupId>
<artifactId>fastjson2</artifactId>
</dependency>
</dependencies>
-</project>
\ No newline at end of file
+</project>
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 f52ea6ae38..fcc8cd9c8b 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
@@ -17,11 +17,19 @@
package org.apache.seata.serializer.fastjson2;
import com.alibaba.fastjson2.JSONB;
+import org.apache.seata.common.executor.Initialize;
+import org.apache.seata.common.json.Fastjson2ObjectReaderWarmup;
import org.apache.seata.common.loader.LoadLevel;
import org.apache.seata.core.serializer.Serializer;
+import org.apache.seata.core.serializer.SerializerSecurityRegistry;
@LoadLevel(name = "FASTJSON2")
-public class Fastjson2Serializer implements Serializer {
+public class Fastjson2Serializer implements Serializer, Initialize {
+
+ @Override
+ public void init() {
+
Fastjson2ObjectReaderWarmup.warmup(SerializerSecurityRegistry.getAllowClassType());
+ }
@Override
public <T> byte[] serialize(T t) {
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
new file mode 100644
index 0000000000..2c04ad0896
--- /dev/null
+++
b/serializer/seata-serializer-fastjson2/src/test/java/org/apache/seata/serializer/fastjson2/Fastjson2ConcurrentRefDeserializationTest.java
@@ -0,0 +1,154 @@
+/*
+ * 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.serializer.fastjson2;
+
+import org.apache.seata.common.executor.Initialize;
+import org.apache.seata.core.protocol.AbstractMessage;
+import org.apache.seata.core.protocol.BatchResultMessage;
+import org.apache.seata.core.protocol.MergedWarpMessage;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class Fastjson2ConcurrentRefDeserializationTest {
+
+ private static final long CONCURRENT_TEST_TIMEOUT_SECONDS = 30;
+
+ @Test
+ public void
concurrentDeserializeReferenceHeavyProtocolMessageDoesNotDropRefFields() throws
Exception {
+ int rounds =
Integer.getInteger("seata.fastjson2.concurrentRef.rounds", 3);
+ for (int round = 0; round < rounds; round++) {
+ assertChildProcessSucceeds();
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ Fastjson2Serializer serializer = new Fastjson2Serializer();
+ ((Initialize) serializer).init();
+ 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);
+ }
+ }
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static MergedWarpMessage referenceHeavyMessage() {
+ MergedWarpMessage message = new MergedWarpMessage();
+ List sharedList = new ArrayList();
+ for (int i = 0; i < 20; i++) {
+ BatchResultMessage resultMessage = new BatchResultMessage();
+ resultMessage.setResultMessages(sharedList);
+ resultMessage.setMsgIds(sharedList);
+ message.msgs.add(resultMessage);
+ message.msgIds.add(i);
+ }
+ return message;
+ }
+
+ private static int countNullRefFields(MergedWarpMessage message) {
+ if (message == null || message.msgs == null) {
+ return 1;
+ }
+ int nullCount = 0;
+ for (AbstractMessage child : message.msgs) {
+ if (!(child instanceof BatchResultMessage)) {
+ nullCount++;
+ continue;
+ }
+ BatchResultMessage batchResult = (BatchResultMessage) child;
+ if (batchResult.getResultMessages() == null) {
+ nullCount++;
+ }
+ if (batchResult.getMsgIds() == null) {
+ nullCount++;
+ }
+ }
+ return nullCount;
+ }
+
+ private static int runConcurrentStress(Fastjson2Serializer serializer,
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();
+ for (int i = 0; i < threadCount; i++) {
+ Thread thread = new Thread(
+ () -> {
+ try {
+ barrier.await(CONCURRENT_TEST_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ if (countNullRefFields((MergedWarpMessage)
serializer.deserialize(bytes)) > 0) {
+ nullTasks.incrementAndGet();
+ }
+ } catch (Throwable throwable) {
+ failure.compareAndSet(null, throwable);
+ } finally {
+ endLatch.countDown();
+ }
+ },
+ "fastjson2-rpc-ref-" + i);
+ thread.start();
+ }
+ if (!endLatch.await(CONCURRENT_TEST_TIMEOUT_SECONDS,
TimeUnit.SECONDS)) {
+ throw new AssertionError("Timed out waiting for concurrent
deserialization");
+ }
+ if (failure.get() != null) {
+ throw new AssertionError("Concurrent deserialization failed",
failure.get());
+ }
+
+ return nullTasks.get();
+ }
+
+ private static void assertChildProcessSucceeds() throws Exception {
+ Process process = new ProcessBuilder(
+ System.getProperty("java.home") + "/bin/java",
+ "-cp",
+ System.getProperty("java.class.path"),
+
Fastjson2ConcurrentRefDeserializationTest.class.getName())
+ .redirectErrorStream(true)
+ .start();
+ if (!process.waitFor(CONCURRENT_TEST_TIMEOUT_SECONDS,
TimeUnit.SECONDS)) {
+ process.destroyForcibly();
+ throw new AssertionError("Timed out waiting for child process");
+ }
+ if (process.exitValue() != 0) {
+ throw new AssertionError("Child process failed: " +
readOutput(process.getInputStream()));
+ }
+ }
+
+ private static String readOutput(InputStream inputStream) throws Exception
{
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ byte[] buffer = new byte[1024];
+ int length;
+ while ((length = inputStream.read(buffer)) != -1) {
+ output.write(buffer, 0, length);
+ }
+ return output.toString("UTF-8");
+ }
+}
diff --git
a/serializer/seata-serializer-fastjson2/src/test/java/org/apache/seata/serializer/fastjson2/Fastjson2SerializerTest.java
b/serializer/seata-serializer-fastjson2/src/test/java/org/apache/seata/serializer/fastjson2/Fastjson2SerializerTest.java
index 4e3cb26489..714d75a362 100644
---
a/serializer/seata-serializer-fastjson2/src/test/java/org/apache/seata/serializer/fastjson2/Fastjson2SerializerTest.java
+++
b/serializer/seata-serializer-fastjson2/src/test/java/org/apache/seata/serializer/fastjson2/Fastjson2SerializerTest.java
@@ -34,6 +34,7 @@ public class Fastjson2SerializerTest {
@BeforeAll
public static void before() {
fastjson2Serializer = new Fastjson2Serializer();
+ fastjson2Serializer.init();
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
