Replace synchronized with ReentrantLock in saga module

changes/en-us/2.x.md

@@ -44,6 +44,7 @@ Add changes here for all PR submitted to the 2.x branch.
 - [[#7955](https://github.com/apache/incubator-seata/pull/7955)] change the getProperty call to resolvePlaceholders
 - [[#7971](https://github.com/apache/incubator-seata/pull/7971)] upgrade some dependencies
 - [[#7970](https://github.com/apache/incubator-seata/pull/7970)] Remove unnecessary refreshLeader from ClusterController cluster endpoint
+- [[#8008](https://github.com/apache/incubator-seata/pull/8008)] replace synchronized with ReentrantLock in saga module for Loom virtual thread compatibility
 
 
 ### security:
@@ -77,6 +78,7 @@ Thanks to these contributors for their code commits. Please report an unintended
 - [xingfudeshi](https://github.com/xingfudeshi)
 - [Sumit6307](https://github.com/Sumit6307)
 - [xiaoxiangyeyu0](https://github.com/xiaoxiangyeyu0)
+- [somiljain](https://github.com/somiljain)
 
 
 Also, we receive many valuable issues, questions and advices from our community. Thanks for you all.

saga/seata-saga-engine/src/main/java/org/apache/seata/saga/engine/pcext/handlers/ChoiceStateHandler.java

@@ -17,6 +17,7 @@
 package org.apache.seata.saga.engine.pcext.handlers;
 
 import org.apache.seata.common.exception.FrameworkErrorCode;
+import org.apache.seata.common.lock.ResourceLock;
 import org.apache.seata.common.util.StringUtils;
 import org.apache.seata.saga.engine.StateMachineConfig;
 import org.apache.seata.saga.engine.exception.EngineExecutionException;
@@ -42,6 +43,8 @@
  */
 public class ChoiceStateHandler implements StateHandler {
 
+    private final ResourceLock choiceLock = new ResourceLock();
+
     @Override
     public void process(ProcessContext context) throws EngineExecutionException {
 
@@ -50,7 +53,7 @@ public void process(ProcessContext context) throws EngineExecutionException {
 
         Map<Object, String> choiceEvaluators = choiceState.getChoiceEvaluators();
         if (choiceEvaluators == null) {
-            synchronized (choiceState) {
+            try (ResourceLock ignored = choiceLock.obtain()) {
                 choiceEvaluators = choiceState.getChoiceEvaluators();
                 if (choiceEvaluators == null) {
 

saga/seata-saga-engine/src/main/java/org/apache/seata/saga/engine/pcext/interceptors/ServiceTaskHandlerInterceptor.java

@@ -18,6 +18,7 @@
 
 import org.apache.seata.common.exception.FrameworkErrorCode;
 import org.apache.seata.common.loader.LoadLevel;
+import org.apache.seata.common.lock.ResourceLock;
 import org.apache.seata.common.util.CollectionUtils;
 import org.apache.seata.common.util.StringUtils;
 import org.apache.seata.saga.engine.StateMachineConfig;
@@ -61,6 +62,8 @@ public class ServiceTaskHandlerInterceptor implements StateHandlerInterceptor {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(ServiceTaskHandlerInterceptor.class);
 
+    private final ResourceLock statusLock = new ResourceLock();
+
     @Override
     public boolean match(Class<? extends InterceptableStateHandler> clazz) {
         return clazz != null
@@ -325,7 +328,7 @@ private void decideExecutionStatus(
 
                 Map<Object, String> statusEvaluators = state.getStatusEvaluators();
                 if (statusEvaluators == null) {
-                    synchronized (state) {
+                    try (ResourceLock ignored = statusLock.obtain()) {
                         statusEvaluators = state.getStatusEvaluators();
                         if (statusEvaluators == null) {
                             statusEvaluators = new LinkedHashMap<>(statusMatchList.size());

saga/seata-saga-engine/src/main/java/org/apache/seata/saga/engine/pcext/utils/CompensationHolder.java

@@ -17,6 +17,7 @@
 package org.apache.seata.saga.engine.pcext.utils;
 
 import org.apache.seata.common.exception.FrameworkErrorCode;
+import org.apache.seata.common.lock.ResourceLock;
 import org.apache.seata.common.util.CollectionUtils;
 import org.apache.seata.common.util.StringUtils;
 import org.apache.seata.saga.engine.exception.EngineExecutionException;
@@ -60,12 +61,14 @@ public class CompensationHolder {
      */
     private Stack<StateInstance> stateStackNeedCompensation = new Stack<>();
 
+    private static final ResourceLock holderLock = new ResourceLock();
+
     public static CompensationHolder getCurrent(ProcessContext context, boolean forceCreate) {
 
         CompensationHolder compensationholder =
                 (CompensationHolder) context.getVariable(DomainConstants.VAR_NAME_CURRENT_COMPENSATION_HOLDER);
         if (compensationholder == null && forceCreate) {
-            synchronized (context) {
+            try (ResourceLock ignored = holderLock.obtain()) {
                 compensationholder =
                         (CompensationHolder) context.getVariable(DomainConstants.VAR_NAME_CURRENT_COMPENSATION_HOLDER);
                 if (compensationholder == null) {

saga/seata-saga-engine/src/main/java/org/apache/seata/saga/engine/pcext/utils/EngineUtils.java

@@ -16,6 +16,7 @@
  */
 package org.apache.seata.saga.engine.pcext.utils;
 
+import org.apache.seata.common.lock.ResourceLock;
 import org.apache.seata.common.util.CollectionUtils;
 import org.apache.seata.common.util.StringUtils;
 import org.apache.seata.saga.engine.AsyncCallback;
@@ -45,6 +46,8 @@ public class EngineUtils {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(EngineUtils.class);
 
+    private static final ResourceLock exceptionLock = new ResourceLock();
+
     /**
      * generate parent id
      *
@@ -198,7 +201,7 @@ public static void handleException(ProcessContext context, AbstractTaskState sta
                 List<Class<? extends Exception>> exceptionClasses = exceptionMatch.getExceptionClasses();
                 if (CollectionUtils.isNotEmpty(exceptions)) {
                     if (exceptionClasses == null) {
-                        synchronized (exceptionMatch) {
+                        try (ResourceLock ignored = exceptionLock.obtain()) {
                             exceptionClasses = exceptionMatch.getExceptionClasses();
                             if (exceptionClasses == null) {
 

saga/seata-saga-engine/src/main/java/org/apache/seata/saga/engine/pcext/utils/LoopContextHolder.java

@@ -16,6 +16,7 @@
  */
 package org.apache.seata.saga.engine.pcext.utils;
 
+import org.apache.seata.common.lock.ResourceLock;
 import org.apache.seata.saga.proctrl.ProcessContext;
 import org.apache.seata.saga.statelang.domain.DomainConstants;
 
@@ -38,12 +39,14 @@ public class LoopContextHolder {
     private final Stack<Integer> forwardCounterStack = new Stack<>();
     private Collection collection;
 
+    private static final ResourceLock contextLock = new ResourceLock();
+
     public static LoopContextHolder getCurrent(ProcessContext context, boolean forceCreate) {
         LoopContextHolder loopContextHolder =
                 (LoopContextHolder) context.getVariable(DomainConstants.VAR_NAME_CURRENT_LOOP_CONTEXT_HOLDER);
 
         if (null == loopContextHolder && forceCreate) {
-            synchronized (context) {
+            try (ResourceLock ignored = contextLock.obtain()) {
                 loopContextHolder =
                         (LoopContextHolder) context.getVariable(DomainConstants.VAR_NAME_CURRENT_LOOP_CONTEXT_HOLDER);
                 if (null == loopContextHolder) {

saga/seata-saga-engine/src/main/java/org/apache/seata/saga/engine/pcext/utils/LoopTaskUtils.java

@@ -17,6 +17,7 @@
 package org.apache.seata.saga.engine.pcext.utils;
 
 import org.apache.seata.common.exception.FrameworkErrorCode;
+import org.apache.seata.common.lock.ResourceLock;
 import org.apache.seata.common.util.CollectionUtils;
 import org.apache.seata.common.util.NumberUtils;
 import org.apache.seata.common.util.StringUtils;
@@ -56,6 +57,8 @@ public class LoopTaskUtils {
 
     public static final String LOOP_STATE_NAME_PATTERN = "-loop-";
 
+    private static final ResourceLock loopLock = new ResourceLock();
+
     /**
      * get Loop Config from State
      *
@@ -227,7 +230,7 @@ public static boolean isCompletionConditionSatisfied(ProcessContext context) {
                 currentLoopContext.getNrOfCompletedInstances().get();
 
         if (!currentLoopContext.isCompletionConditionSatisfied()) {
-            synchronized (currentLoopContext) {
+            try (ResourceLock ignored = loopLock.obtain()) {
                 if (!currentLoopContext.isCompletionConditionSatisfied()) {
                     Map<String, Object> stateMachineContext =
                             (Map<String, Object>) context.getVariable(DomainConstants.VAR_NAME_STATEMACHINE_CONTEXT);

saga/seata-saga-engine/src/main/java/org/apache/seata/saga/engine/pcext/utils/ParameterUtils.java

@@ -16,6 +16,7 @@
  */
 package org.apache.seata.saga.engine.pcext.utils;
 
+import org.apache.seata.common.lock.ResourceLock;
 import org.apache.seata.common.util.CollectionUtils;
 import org.apache.seata.common.util.StringUtils;
 import org.apache.seata.saga.engine.expression.Expression;
@@ -36,6 +37,8 @@
  */
 public class ParameterUtils {
 
+    private static final ResourceLock inputLock = new ResourceLock();
+
     public static List<Object> createInputParams(
             ExpressionResolver expressionResolver,
             StateInstanceImpl stateInstance,
@@ -48,7 +51,7 @@ public static List<Object> createInputParams(
 
         List<Object> inputExpressions = serviceTaskState.getInputExpressions();
         if (inputExpressions == null) {
-            synchronized (serviceTaskState) {
+            try (ResourceLock ignored = inputLock.obtain()) {
                 inputExpressions = serviceTaskState.getInputExpressions();
                 if (inputExpressions == null) {
                     inputExpressions = new ArrayList<>(inputAssignments.size());
@@ -77,7 +80,7 @@ public static Map<String, Object> createOutputParams(
 
         Map<String, Object> outputExpressions = serviceTaskState.getOutputExpressions();
         if (outputExpressions == null) {
-            synchronized (serviceTaskState) {
+            try (ResourceLock ignored = inputLock.obtain()) {
                 outputExpressions = serviceTaskState.getOutputExpressions();
                 if (outputExpressions == null) {
                     outputExpressions = new LinkedHashMap<>(outputAssignments.size());

saga/seata-saga-engine/src/main/java/org/apache/seata/saga/engine/repo/impl/StateMachineRepositoryImpl.java

@@ -16,6 +16,7 @@
  */
 package org.apache.seata.saga.engine.repo.impl;
 
+import org.apache.seata.common.lock.ResourceLock;
 import org.apache.seata.common.util.CollectionUtils;
 import org.apache.seata.common.util.StringUtils;
 import org.apache.seata.saga.engine.repo.StateMachineRepository;
@@ -44,6 +45,7 @@
 public class StateMachineRepositoryImpl implements StateMachineRepository {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(StateMachineRepositoryImpl.class);
+    private final ResourceLock itemLock = new ResourceLock();
     private Map<
                     String
                     /** Name_Tenant **/
@@ -66,7 +68,7 @@ public class StateMachineRepositoryImpl implements StateMachineRepository {
     public StateMachine getStateMachineById(String stateMachineId) {
         Item item = CollectionUtils.computeIfAbsent(stateMachineMapById, stateMachineId, key -> new Item());
         if (item.getValue() == null && stateLangStore != null) {
-            synchronized (item) {
+            try (ResourceLock ignored = itemLock.obtain()) {
                 if (item.getValue() == null) {
                     StateMachine stateMachine = stateLangStore.getStateMachineById(stateMachineId);
                     if (stateMachine != null) {
@@ -92,7 +94,7 @@ public StateMachine getStateMachine(String stateMachineName, String tenantId) {
         Item item = CollectionUtils.computeIfAbsent(
                 stateMachineMapByNameAndTenant, stateMachineName + "_" + tenantId, key -> new Item());
         if (item.getValue() == null && stateLangStore != null) {
-            synchronized (item) {
+            try (ResourceLock ignored = itemLock.obtain()) {
                 if (item.getValue() == null) {
                     StateMachine stateMachine = stateLangStore.getLastVersionStateMachine(stateMachineName, tenantId);
                     if (stateMachine != null) {

saga/seata-saga-processctrl/src/main/java/org/apache/seata/saga/proctrl/eventing/impl/DirectEventBus.java

@@ -17,6 +17,7 @@
 package org.apache.seata.saga.proctrl.eventing.impl;
 
 import org.apache.seata.common.exception.FrameworkException;
+import org.apache.seata.common.lock.ResourceLock;
 import org.apache.seata.common.util.CollectionUtils;
 import org.apache.seata.saga.proctrl.ProcessContext;
 import org.apache.seata.saga.proctrl.eventing.EventConsumer;
@@ -36,6 +37,8 @@ public class DirectEventBus extends AbstractEventBus<ProcessContext> {
 
     private static final String VAR_NAME_SYNC_EXE_STACK = "_sync_execution_stack_";
 
+    private final ResourceLock contextLock = new ResourceLock();
+
     @Override
     public boolean offer(ProcessContext context) throws FrameworkException {
         List<EventConsumer> eventHandlers = getEventConsumers(context.getClass());
@@ -49,7 +52,7 @@ public boolean offer(ProcessContext context) throws FrameworkException {
         boolean isFirstEvent = false;
         Stack<ProcessContext> currentStack = (Stack<ProcessContext>) context.getVariable(VAR_NAME_SYNC_EXE_STACK);
         if (currentStack == null) {
-            synchronized (context) {
+            try (ResourceLock ignored = contextLock.obtain()) {
                 currentStack = (Stack<ProcessContext>) context.getVariable(VAR_NAME_SYNC_EXE_STACK);
                 if (currentStack == null) {
                     currentStack = new Stack<>();

saga/seata-saga-spring/src/main/java/org/apache/seata/saga/engine/invoker/impl/SpringBeanServiceInvoker.java

@@ -19,6 +19,7 @@
 import org.apache.seata.common.exception.FrameworkErrorCode;
 import org.apache.seata.common.json.JsonSerializer;
 import org.apache.seata.common.json.JsonSerializerFactory;
+import org.apache.seata.common.lock.ResourceLock;
 import org.apache.seata.common.util.CollectionUtils;
 import org.apache.seata.saga.engine.exception.EngineExecutionException;
 import org.apache.seata.saga.engine.invoker.ServiceInvoker;
@@ -53,6 +54,9 @@ public class SpringBeanServiceInvoker implements ServiceInvoker, ApplicationCont
 
     private static final Logger LOGGER = LoggerFactory.getLogger(SpringBeanServiceInvoker.class);
 
+    private final ResourceLock methodLock = new ResourceLock();
+    private final ResourceLock retryLock = new ResourceLock();
+
     private ApplicationContext applicationContext;
     private ThreadPoolExecutor threadPoolExecutor;
     private String sagaJsonParser;
@@ -105,7 +109,7 @@ protected Object doInvoke(ServiceTaskStateImpl state, Object[] input) throws Thr
 
         Method method = state.getMethod();
         if (method == null) {
-            synchronized (state) {
+            try (ResourceLock ignored = methodLock.obtain()) {
                 method = state.getMethod();
                 if (method == null) {
                     method = findMethod(bean.getClass(), state.getServiceMethod(), state.getParameterTypes());
@@ -196,7 +200,7 @@ private Retry matchRetryConfig(List<Retry> retryList, Throwable e) {
                 } else {
                     List<Class<? extends Exception>> exceptionClasses = retryConfig.getExceptionClasses();
                     if (exceptionClasses == null) {
-                        synchronized (retryConfig) {
+                        try (ResourceLock ignored = retryLock.obtain()) {
                             exceptionClasses = retryConfig.getExceptionClasses();
                             if (exceptionClasses == null) {