feat: 2.6.0

1.修复节点重试时导致决策错误问题

easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/AbstractWorkflowExecutor.java

@@ -3,10 +3,7 @@ package com.aizuda.easy.retry.server.job.task.support.executor.workflow;
 import akka.actor.ActorRef;
 import cn.hutool.core.lang.Assert;
 import cn.hutool.core.util.StrUtil;
-import com.aizuda.easy.retry.common.core.enums.JobArgsTypeEnum;
+import com.aizuda.easy.retry.common.core.enums.*;
-import com.aizuda.easy.retry.common.core.enums.JobOperationReasonEnum;
-import com.aizuda.easy.retry.common.core.enums.JobTaskBatchStatusEnum;
-import com.aizuda.easy.retry.common.core.enums.StatusEnum;
 import com.aizuda.easy.retry.server.common.akka.ActorGenerator;
 import com.aizuda.easy.retry.server.common.exception.EasyRetryServerException;
 import com.aizuda.easy.retry.server.job.task.dto.WorkflowNodeTaskExecuteDTO;
@@ -30,12 +27,16 @@ import org.springframework.transaction.support.TransactionCallbackWithoutResult;
 import org.springframework.transaction.support.TransactionSynchronization;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.transaction.support.TransactionTemplate;
+import org.springframework.util.CollectionUtils;
 
 import java.text.MessageFormat;
 import java.time.Duration;
+import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
 
+import static com.aizuda.easy.retry.common.core.enums.JobOperationReasonEnum.WORKFLOW_SUCCESSOR_SKIP_EXECUTION;
+
 /**
  * @author xiaowoniu
  * @date 2023-12-24 08:15:19
@@ -64,10 +65,32 @@ public abstract class AbstractWorkflowExecutor implements WorkflowExecutor, Init
         distributedLockHandler.lockWithDisposableAndRetry(
                 () -> {
 
-                    Long total = jobTaskBatchMapper.selectCount(new LambdaQueryWrapper<JobTaskBatch>()
+                    long total = 0;
+                    // 条件节点存在并发问题，需要特殊处理
+                    if (WorkflowNodeTypeEnum.DECISION.getType() == context.getNodeType()) {
+                        List<JobTaskBatch> jobTaskBatches = jobTaskBatchMapper.selectList(new LambdaQueryWrapper<JobTaskBatch>()
+                                .select(JobTaskBatch::getOperationReason)
                                 .eq(JobTaskBatch::getWorkflowTaskBatchId, context.getWorkflowTaskBatchId())
                                 .eq(JobTaskBatch::getWorkflowNodeId, context.getWorkflowNodeId())
                         );
+
+                        if (!CollectionUtils.isEmpty(jobTaskBatches)) {
+                            total = jobTaskBatches.size();
+                            JobTaskBatch jobTaskBatch = jobTaskBatches.get(0);
+                            if (WORKFLOW_SUCCESSOR_SKIP_EXECUTION.contains(jobTaskBatch.getOperationReason())) {
+                                context.setEvaluationResult(Boolean.FALSE);
+                            } else {
+                                context.setEvaluationResult(Boolean.TRUE);
+                            }
+                        }
+
+                    } else {
+                        total = jobTaskBatchMapper.selectCount(new LambdaQueryWrapper<JobTaskBatch>()
+                                .eq(JobTaskBatch::getWorkflowTaskBatchId, context.getWorkflowTaskBatchId())
+                                .eq(JobTaskBatch::getWorkflowNodeId, context.getWorkflowNodeId())
+                        );
+                    }
+
                     if (total > 0) {
                         log.warn("任务节点[{}]已被执行，请勿重复执行", context.getWorkflowNodeId());
                         return;

easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/WorkflowExecutorContext.java

@@ -98,4 +98,8 @@ public class WorkflowExecutorContext {
      */
     private Integer taskExecutorScene;
 
+    /**
+     * 1、任务节点 2、条件节点 3、回调节点
+     */
+    private Integer nodeType;
 }

easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/handler/WorkflowBatchHandler.java

@@ -37,6 +37,7 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import static com.aizuda.easy.retry.common.core.enums.JobOperationReasonEnum.WORKFLOW_SUCCESSOR_SKIP_EXECUTION;
 import static com.aizuda.easy.retry.common.core.enums.JobTaskBatchStatusEnum.NOT_COMPLETE;
 
 /**
@@ -131,7 +132,7 @@ public class WorkflowBatchHandler {
 
             for (JobTaskBatch jobTaskBatch : jobTaskBatchList) {
                 // 只要是无需处理的说明后面的子节点都不需要处理了，isNeedProcess为false
-                if (JobOperationReasonEnum.WORKFLOW_SUCCESSOR_SKIP_EXECUTION.contains(jobTaskBatch.getOperationReason())) {
+                if (WORKFLOW_SUCCESSOR_SKIP_EXECUTION.contains(jobTaskBatch.getOperationReason())) {
                     isNeedProcess = false;
                     continue;
                 }
@@ -227,8 +228,7 @@ public class WorkflowBatchHandler {
         // 判定条件节点是否已经执行完成
         JobTaskBatch parentJobTaskBatch = jobTaskBatchMap.get(parentId);
         if (Objects.nonNull(parentJobTaskBatch) &&
-                JobOperationReasonEnum.WORKFLOW_NODE_NO_REQUIRED.getReason()
+                WORKFLOW_SUCCESSOR_SKIP_EXECUTION.contains(parentJobTaskBatch.getOperationReason())) {
-                        == parentJobTaskBatch.getOperationReason()) {
             return;
         }
 

easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/service/impl/JobLogServiceImpl.java

@@ -47,7 +47,7 @@ public class JobLogServiceImpl implements JobLogService {
                 .ge(JobLogMessage::getJobId, queryVO.getJobId())
                 .eq(JobLogMessage::getTaskId, queryVO.getTaskId());
 
-        queryWrapper.orderByAsc(JobLogMessage::getRealTime).orderByDesc(JobLogMessage::getId);
+        queryWrapper.orderByAsc(JobLogMessage::getRealTime).orderByAsc(JobLogMessage::getId);
         PageDTO<JobLogMessage> selectPage = jobLogMessageMapper.selectPage(pageDTO, queryWrapper);
         List<JobLogMessage> records = selectPage.getRecords();
         if (CollectionUtils.isEmpty(records)) {
@@ -86,7 +86,13 @@ public class JobLogServiceImpl implements JobLogService {
 
         long nextStartId = 0;
         List<String> messages = Lists.newArrayList();
-        List<JobLogMessage> jobLogMessages = jobLogMessageMapper.selectBatchIds(ids);
+        List<JobLogMessage> jobLogMessages = jobLogMessageMapper.selectList(
+                new LambdaQueryWrapper<JobLogMessage>()
+                        .in(JobLogMessage::getId, ids)
+                        .orderByAsc(JobLogMessage::getRealTime)
+                        .orderByAsc(JobLogMessage::getId)
+        );
+
         for (final JobLogMessage jobLogMessage : jobLogMessages) {
 
             List<String> originalList = JsonUtil.parseObject(jobLogMessage.getMessage(), List.class);