Support loop element resolution during compensation

changes/en-us/2.x.md

@@ -24,6 +24,20 @@ Add changes here for all PR submitted to the 2.x branch.
 
 ### bugfix:
 
+- [[#7929](https://github.com/apache/incubator-seata/pull/7929)] fix Kingbase undo log insert SQL error
+- [[#7940](https://github.com/apache/incubator-seata/pull/7940)] fix incorrect Jakarta package paths
+- [[#7960](https://github.com/apache/incubator-seata/pull/7960)] fix console API service bean not found
+- [[#7956](https://github.com/apache/incubator-seata/pull/7956)] fix empty JaCoCo report on JDK 17+
+- [[#7965](https://github.com/apache/incubator-seata/pull/7965)] fix primary key index recognition failure in DM and Kingbase
+- [[#7981](https://github.com/apache/incubator-seata/pull/7981)] fix resource leak when closing DataSourceProxy
+- [[#7992](https://github.com/apache/incubator-seata/pull/7992)] fix missing branch type when reporting branch transaction status
+- [[#8035](https://github.com/apache/incubator-seata/pull/8035)] fix IllegalArgumentException when GET request contains body
+- [[#8078](https://github.com/apache/incubator-seata/pull/8078)] fix MySQL undo log serialization exception
+- [[#8106](https://github.com/apache/incubator-seata/pull/8106)] fix NPE during AOT proxy creation
+- [[#8113](https://github.com/apache/incubator-seata/pull/8113)] fix console export JSON consistency and download issues
+- [[#8118](https://github.com/apache/incubator-seata/pull/8118)] Use explicit columns in rollback validation query
+- [[#8124](https://github.com/apache/incubator-seata/pull/8124)] Fix PK extraction for batch inserts with SQL functions
+- [[#8127](https://github.com/apache/incubator-seata/pull/8127)] Support loop element resolution during compensation
 - [[#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

compatible/src/test/java/io/seata/saga/engine/pcext/StateHandlerInterceptorTest.java

@@ -16,9 +16,30 @@
  */
 package io.seata.saga.engine.pcext;
 
+import org.apache.seata.saga.engine.StateMachineConfig;
+import org.apache.seata.saga.engine.pcext.StateInstruction;
+import org.apache.seata.saga.engine.pcext.interceptors.ServiceTaskHandlerInterceptor;
+import org.apache.seata.saga.engine.sequence.SeqGenerator;
+import org.apache.seata.saga.proctrl.HierarchicalProcessContext;
+import org.apache.seata.saga.statelang.domain.DomainConstants;
+import org.apache.seata.saga.statelang.domain.StateInstance;
+import org.apache.seata.saga.statelang.domain.StateMachine;
+import org.apache.seata.saga.statelang.domain.StateMachineInstance;
+import org.apache.seata.saga.statelang.domain.impl.ServiceTaskStateImpl;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 
+import java.util.Date;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 /**
  * Test cases for StateHandlerInterceptor interface compatibility wrapper.
@@ -44,4 +65,48 @@ public void testExtendsApacheStateHandlerInterceptor() {
                         StateHandlerInterceptor.class),
                 "StateHandlerInterceptor should extend org.apache.seata.saga.engine.pcext.StateHandlerInterceptor");
     }
+
+    @Test
+    public void preProcessWhenLoopElementPresent_SetsExtensionParamsTest() {
+        HierarchicalProcessContext context = mock(HierarchicalProcessContext.class);
+        StateInstruction instruction = mock(StateInstruction.class);
+        StateMachineInstance stateMachineInstance = mock(StateMachineInstance.class);
+        StateMachineConfig stateMachineConfig = mock(StateMachineConfig.class);
+        StateMachine stateMachine = mock(StateMachine.class);
+        ServiceTaskStateImpl state = mock(ServiceTaskStateImpl.class);
+        SeqGenerator seqGenerator = mock(SeqGenerator.class);
+
+        when(context.getInstruction(StateInstruction.class)).thenReturn(instruction);
+        when(instruction.getState(context)).thenReturn(state);
+        when(context.getVariable(DomainConstants.VAR_NAME_STATEMACHINE_INST)).thenReturn(stateMachineInstance);
+        when(context.getVariable(DomainConstants.VAR_NAME_STATEMACHINE_CONFIG)).thenReturn(stateMachineConfig);
+
+        when(stateMachineInstance.getGmtUpdated()).thenReturn(new Date());
+        when(stateMachineConfig.getTransOperationTimeout()).thenReturn(Math.toIntExact(100000L));
+        when(stateMachineInstance.getStateMachine()).thenReturn(stateMachine);
+        when(stateMachine.isPersist()).thenReturn(false);
+
+        when(state.getName()).thenReturn("LoopTask");
+        when(state.isForCompensation()).thenReturn(false);
+
+        Object mockLoopElement = "testLoopElementValue";
+        when(context.getVariable(DomainConstants.VAR_NAME_LOOP_ELEMENT)).thenReturn(mockLoopElement);
+
+        when(stateMachineConfig.getSeqGenerator()).thenReturn(seqGenerator);
+        when(seqGenerator.generate(anyString())).thenReturn("SEQ_1001");
+
+        ServiceTaskHandlerInterceptor interceptor = new ServiceTaskHandlerInterceptor();
+        assertDoesNotThrow(() -> interceptor.preProcess(context));
+
+        ArgumentCaptor<StateInstance> captor = ArgumentCaptor.forClass(StateInstance.class);
+        verify(context).setVariableLocally(eq(DomainConstants.VAR_NAME_STATE_INST), captor.capture());
+
+        StateInstance capturedInstance = captor.getValue();
+        Assertions.assertNotNull(capturedInstance);
+        Assertions.assertNotNull(capturedInstance.getExtensionParams());
+
+        @SuppressWarnings("unchecked")
+        Map<String, Object> extensionParams = (Map<String, Object>) capturedInstance.getExtensionParams();
+        Assertions.assertEquals(mockLoopElement, extensionParams.get(DomainConstants.VAR_NAME_LOOP_ELEMENT));
+    }
 }

compatible/src/test/resources/saga/sql/h2_init.sql

@@ -71,5 +71,6 @@ create table if not exists seata_state_inst
     excep                    blob comment 'exception',
     gmt_updated              timestamp(3) comment 'update time',
     gmt_end                  timestamp(3) comment 'end time',
+    ext_params               LONGTEXT     DEFAULT NULL,
     primary key (id, machine_inst_id)
-);
+);

saga/seata-saga-engine-store/src/main/java/org/apache/seata/saga/engine/store/db/StateLogStoreSqls.java

@@ -63,13 +63,13 @@ public class StateLogStoreSqls {
     private static final String STATE_INSTANCE_FIELDS =
             "id, machine_inst_id, name, type, business_key, gmt_started, service_name, service_method, service_type, "
                     + "is_for_update, status, input_params, output_params, excep, gmt_end, state_id_compensated_for, "
-                    + "state_id_retried_for";
+                    + "state_id_retried_for, ext_params";
 
     private static final String RECORD_STATE_STARTED_SQL =
             "INSERT INTO ${TABLE_PREFIX}state_inst (id, machine_inst_id, name, type,"
                     + " gmt_started, service_name, service_method, service_type, is_for_update, input_params, status, "
-                    + "business_key, state_id_compensated_for, state_id_retried_for, gmt_updated)\n"
-                    + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+                    + "business_key, state_id_compensated_for, state_id_retried_for, gmt_updated, ext_params)\n"
+                    + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
 
     private static final String RECORD_STATE_FINISHED_SQL =
             "UPDATE ${TABLE_PREFIX}state_inst SET gmt_end = ?, excep = ?, status = ?, output_params = ?, gmt_updated = ? "

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

@@ -91,6 +91,10 @@ public void process(ProcessContext context) throws EngineExecutionException {
         }
 
         startParams.put(DomainConstants.VAR_NAME_PARENT_ID, EngineUtils.generateParentId(stateInstance));
+        Object loopElement = context.getVariable(DomainConstants.VAR_NAME_LOOP_ELEMENT);
+        if (loopElement != null) {
+            startParams.put(DomainConstants.VAR_NAME_LOOP_ELEMENT, loopElement);
+        }
         try {
             if (LOGGER.isDebugEnabled()) {
                 LOGGER.debug(

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

@@ -17,7 +17,12 @@
 package org.apache.seata.saga.engine.pcext.interceptors;
 
 import org.apache.seata.common.loader.LoadLevel;
+import org.apache.seata.common.util.StringUtils;
+import org.apache.seata.saga.engine.StateMachineConfig;
 import org.apache.seata.saga.engine.exception.EngineExecutionException;
+import org.apache.seata.saga.engine.expression.Expression;
+import org.apache.seata.saga.engine.expression.ExpressionFactory;
+import org.apache.seata.saga.engine.expression.ExpressionFactoryManager;
 import org.apache.seata.saga.engine.pcext.InterceptableStateHandler;
 import org.apache.seata.saga.engine.pcext.StateHandlerInterceptor;
 import org.apache.seata.saga.engine.pcext.StateInstruction;
@@ -34,10 +39,13 @@
 import org.apache.seata.saga.statelang.domain.StateInstance;
 import org.apache.seata.saga.statelang.domain.TaskState.Loop;
 import org.apache.seata.saga.statelang.domain.impl.AbstractTaskState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Semaphore;
 
@@ -48,6 +56,8 @@
 @LoadLevel(name = "LoopTask", order = 90)
 public class LoopTaskHandlerInterceptor implements StateHandlerInterceptor {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(LoopTaskHandlerInterceptor.class);
+
     @Override
     public boolean match(Class<? extends InterceptableStateHandler> clazz) {
         return clazz != null
@@ -58,37 +68,86 @@ public boolean match(Class<? extends InterceptableStateHandler> clazz) {
 
     @Override
     public void preProcess(ProcessContext context) throws EngineExecutionException {
+        if (!context.hasVariable(DomainConstants.VAR_NAME_IS_LOOP_STATE)) {
+            return;
+        }
 
-        if (context.hasVariable(DomainConstants.VAR_NAME_IS_LOOP_STATE)) {
-            StateInstruction instruction = context.getInstruction(StateInstruction.class);
-            AbstractTaskState currentState = (AbstractTaskState) instruction.getState(context);
-
-            int loopCounter;
-            Loop loop;
-
-            // get loop config
-            if (context.hasVariable(DomainConstants.VAR_NAME_CURRENT_COMPEN_TRIGGER_STATE)) {
-                // compensate condition should get stateToBeCompensated 's config
-                CompensationHolder compensationHolder = CompensationHolder.getCurrent(context, true);
-                StateInstance stateToBeCompensated =
-                        compensationHolder.getStatesNeedCompensation().get(currentState.getName());
-                AbstractTaskState compensateState = (AbstractTaskState) stateToBeCompensated
-                        .getStateMachineInstance()
-                        .getStateMachine()
-                        .getState(EngineUtils.getOriginStateName(stateToBeCompensated));
-                loop = compensateState.getLoop();
-                loopCounter = LoopTaskUtils.reloadLoopCounter(stateToBeCompensated.getName());
-            } else {
-                loop = currentState.getLoop();
-                loopCounter = (int) context.getVariable(DomainConstants.LOOP_COUNTER);
+        StateInstruction instruction = context.getInstruction(StateInstruction.class);
+        AbstractTaskState currentState = (AbstractTaskState) instruction.getState(context);
+
+        @SuppressWarnings("unchecked")
+        Map<String, Object> contextVariables =
+                (Map<String, Object>) context.getVariable(DomainConstants.VAR_NAME_STATEMACHINE_CONTEXT);
+
+        int loopCounter;
+        Loop loop;
+        Collection<?> collection = null;
+        Object element = null;
+        boolean isCompensation = context.hasVariable(DomainConstants.VAR_NAME_CURRENT_COMPEN_TRIGGER_STATE);
+
+        if (isCompensation) {
+            CompensationHolder compensationHolder = CompensationHolder.getCurrent(context, true);
+            StateInstance stateToBeCompensated =
+                    compensationHolder.getStatesNeedCompensation().get(currentState.getName());
+            AbstractTaskState compensateState = (AbstractTaskState) stateToBeCompensated
+                    .getStateMachineInstance()
+                    .getStateMachine()
+                    .getState(EngineUtils.getOriginStateName(stateToBeCompensated));
+
+            loop = compensateState.getLoop();
+            loopCounter = LoopTaskUtils.reloadLoopCounter(stateToBeCompensated.getName());
+
+            Object extensionParamsObj = stateToBeCompensated.getExtensionParams();
+            if (extensionParamsObj instanceof Map) {
+                Map<?, ?> extensionMap = (Map<?, ?>) extensionParamsObj;
+                if (extensionMap.containsKey(DomainConstants.VAR_NAME_LOOP_ELEMENT)) {
+                    element = extensionMap.get(DomainConstants.VAR_NAME_LOOP_ELEMENT);
+                }
             }
 
-            Collection collection = LoopContextHolder.getCurrent(context, true).getCollection();
-            Map<String, Object> contextVariables =
-                    (Map<String, Object>) context.getVariable(DomainConstants.VAR_NAME_STATEMACHINE_CONTEXT);
-            Map<String, Object> copyContextVariables = new ConcurrentHashMap<>(contextVariables);
+            if (element == null) {
+                StateMachineConfig stateMachineConfig =
+                        (StateMachineConfig) context.getVariable(DomainConstants.VAR_NAME_STATEMACHINE_CONFIG);
+                ExpressionFactoryManager expressionFactoryManager =
+                        stateMachineConfig != null ? stateMachineConfig.getExpressionFactoryManager() : null;
+
+                if (expressionFactoryManager != null && StringUtils.isNotBlank(loop.getCollection())) {
+                    ExpressionFactory expressionFactory = expressionFactoryManager.getExpressionFactory(
+                            ExpressionFactoryManager.DEFAULT_EXPRESSION_TYPE);
+                    Expression expression = expressionFactory.createExpression(loop.getCollection());
+                    Object evaluatedResult = expression.getValue(contextVariables);
+
+                    if (evaluatedResult instanceof Collection) {
+                        collection = (Collection<?>) evaluatedResult;
+                        element = iterator(collection, loopCounter, true);
+                        LOGGER.warn(
+                                "Loop element not found in StateInstance for state [{}]. Re-evaluating expression [{}] during compensation.",
+                                compensateState.getName(),
+                                loop.getCollection());
+                    }
+                }
+            }
+        } else {
+            loop = currentState.getLoop();
+            loopCounter = (int) context.getVariable(DomainConstants.LOOP_COUNTER);
+            collection = LoopContextHolder.getCurrent(context, true).getCollection();
+            element = iterator(collection, loopCounter, false);
+        }
+
+        if (!isCompensation || element != null || collection != null) {
+            Map<String, Object> copyContextVariables =
+                    new ConcurrentHashMap<>(Objects.requireNonNull(contextVariables));
             copyContextVariables.put(loop.getElementIndexName(), loopCounter);
-            copyContextVariables.put(loop.getElementVariableName(), iterator(collection, loopCounter));
+
+            if (element != null) {
+                copyContextVariables.put(loop.getElementVariableName(), element);
+
+                if (!isCompensation) {
+                    ((HierarchicalProcessContext) context)
+                            .setVariableLocally(DomainConstants.VAR_NAME_LOOP_ELEMENT, element);
+                }
+            }
+
             ((HierarchicalProcessContext) context)
                     .setVariableLocally(DomainConstants.VAR_NAME_STATEMACHINE_CONTEXT, copyContextVariables);
         }
@@ -131,14 +190,26 @@ public void postProcess(ProcessContext context, Exception e) throws EngineExecut
         }
     }
 
-    private Object iterator(Collection collection, int loopCounter) {
-        Iterator iterator = collection.iterator();
+    private Object iterator(Collection<?> collection, int loopCounter, boolean isCompensation) {
+        if (collection == null) {
+            return null;
+        }
+
+        if (isCompensation && loopCounter >= collection.size()) {
+            LOGGER.warn(
+                    "Collection size ({}) is smaller than loopCounter ({}). The collection likely mutated between forward execution and compensation. Skipping loop element injection.",
+                    collection.size(),
+                    loopCounter);
+            return null;
+        }
+
+        Iterator<?> iterator = collection.iterator();
         int index = 0;
         Object value = null;
-        while (index <= loopCounter) {
+        while (index <= loopCounter && iterator.hasNext()) {
             value = iterator.next();
-            index += 1;
+            index++;
         }
-        return value;
+        return index == loopCounter + 1 ? value : null;
     }
 }

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

@@ -126,6 +126,7 @@ public void postProcess(ProcessContext context, Exception exp) throws EngineExec
 
         context.removeVariable(DomainConstants.VAR_NAME_OUTPUT_PARAMS);
         context.removeVariable(DomainConstants.VAR_NAME_INPUT_PARAMS);
+        context.removeVariable(DomainConstants.VAR_NAME_LOOP_ELEMENT);
 
         if (exp != null
                 && context.getVariable(DomainConstants.VAR_NAME_IS_EXCEPTION_NOT_CATCH) != null

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

@@ -49,6 +49,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.Date;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -62,7 +63,7 @@ public class ServiceTaskHandlerInterceptor implements StateHandlerInterceptor {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(ServiceTaskHandlerInterceptor.class);
 
-    private final ResourceLock STATUS_LOCK = new ResourceLock();
+    private final ResourceLock statusLock = new ResourceLock();
 
     @Override
     public boolean match(Class<? extends InterceptableStateHandler> clazz) {
@@ -193,6 +194,12 @@ public void preProcess(ProcessContext context) throws EngineExecutionException {
         }
 
         stateInstance.setInputParams(serviceInputParams);
+        Object loopElement = context.getVariable(DomainConstants.VAR_NAME_LOOP_ELEMENT);
+        if (loopElement != null) {
+            Map<String, Object> extensionParams = new HashMap<>();
+            extensionParams.put(DomainConstants.VAR_NAME_LOOP_ELEMENT, loopElement);
+            stateInstance.setExtensionParams(extensionParams);
+        }
 
         if (stateMachineInstance.getStateMachine().isPersist()
                 && state.isPersist()
@@ -288,6 +295,7 @@ public void postProcess(ProcessContext context, Exception exp) throws EngineExec
 
         context.removeVariable(DomainConstants.VAR_NAME_OUTPUT_PARAMS);
         context.removeVariable(DomainConstants.VAR_NAME_INPUT_PARAMS);
+        context.removeVariable(DomainConstants.VAR_NAME_LOOP_ELEMENT);
 
         stateInstance.setGmtEnd(new Date());
 
@@ -328,7 +336,7 @@ private void decideExecutionStatus(
 
                 Map<Object, String> statusEvaluators = state.getStatusEvaluators();
                 if (statusEvaluators == null) {
-                    try (ResourceLock ignored = STATUS_LOCK.obtain()) {
+                    try (ResourceLock ignored = statusLock.obtain()) {
                         statusEvaluators = state.getStatusEvaluators();
                         if (statusEvaluators == null) {
                             statusEvaluators = new LinkedHashMap<>(statusMatchList.size());