nop_batch_task保存额外的flowId和flowStepId，用于关联逻辑流

entropy-cloud · Nov 30, 2024 · 911184a · 911184a
1 parent 3d8e4be
commit 911184a
Show file tree

Hide file tree

Showing 11 changed files with 30 additions and 28 deletions.
diff --git a/nop-batch/model/nop-batch.orm.xlsx b/nop-batch/model/nop-batch.orm.xlsx
diff --git a/nop-batch/nop-batch-dao/src/main/resources/_vfs/nop/batch/orm/_app.orm.xml b/nop-batch/nop-batch-dao/src/main/resources/_vfs/nop/batch/orm/_app.orm.xml
@@ -71,9 +71,9 @@
                 <column code="INPUT_FILE_ID" displayName="输入文件" name="inputFileId" precision="32" propId="10"
                         stdDataType="string" stdSqlType="VARCHAR" i18n-en:displayName="Input File"/>
                 <column code="FLOW_STEP_ID" displayName="关联流程步骤ID" name="flowStepId" precision="50" propId="11"
-                        stdDataType="string" stdSqlType="VARCHAR" i18n-en:displayName="Flow Step ID"/>
+                        stdDataType="string" stdSqlType="VARCHAR" tagSet="var" i18n-en:displayName="Flow Step ID"/>
                 <column code="FLOW_ID" displayName="关联流程ID" name="flowId" precision="50" propId="12"
-                        stdDataType="string" stdSqlType="VARCHAR" i18n-en:displayName="Flow ID"/>
+                        stdDataType="string" stdSqlType="VARCHAR" tagSet="var" i18n-en:displayName="Flow ID"/>
                 <column code="RESTART_TIME" displayName="重启时间" name="restartTime" propId="13" stdDataType="timestamp"
                         stdSqlType="TIMESTAMP" tagSet="clock" i18n-en:displayName="Restart Time"/>
                 <column code="RESULT_STATUS" comment="0表示成功" displayName="返回状态码" name="resultStatus" propId="14"

diff --git a/...dsl/cases/io/nop/batch/dsl/TestBatchTaskDsl/testBatchTask/input/tables/nop_batch_task.csv b/...dsl/cases/io/nop/batch/dsl/TestBatchTaskDsl/testBatchTask/input/tables/nop_batch_task.csv
@@ -1 +1 @@
-SID,TASK_NAME,TASK_KEY,TASK_STATUS,START_TIME,END_TIME,TASK_PARAMS,EXEC_COUNT,WORKER_ID,INPUT_FILE_ID,FLOW_STEP_ID,RESTART_TIME,RESULT_STATUS,RESULT_CODE,RESULT_MSG,ERROR_STACK,COMPLETED_INDEX,COMPLETE_ITEM_COUNT,LOAD_RETRY_COUNT,LOAD_SKIP_COUNT,RETRY_ITEM_COUNT,PROCESS_ITEM_COUNT,SKIP_ITEM_COUNT,WRITE_ITEM_COUNT,VERSION,CREATED_BY,CREATE_TIME,UPDATED_BY,UPDATE_TIME,REMARK
+SID,TASK_NAME,TASK_KEY,TASK_STATUS,START_TIME,END_TIME,TASK_PARAMS,EXEC_COUNT,WORKER_ID,INPUT_FILE_ID,FLOW_STEP_ID,FLOW_ID,RESTART_TIME,RESULT_STATUS,RESULT_CODE,RESULT_MSG,ERROR_STACK,COMPLETED_INDEX,COMPLETE_ITEM_COUNT,LOAD_RETRY_COUNT,LOAD_SKIP_COUNT,RETRY_ITEM_COUNT,PROCESS_ITEM_COUNT,SKIP_ITEM_COUNT,WRITE_ITEM_COUNT,VERSION,CREATED_BY,CREATE_TIME,UPDATED_BY,UPDATE_TIME,REMARK
diff --git a/...sl/cases/io/nop/batch/dsl/TestBatchTaskDsl/testBatchTask/output/tables/nop_batch_task.csv b/...sl/cases/io/nop/batch/dsl/TestBatchTaskDsl/testBatchTask/output/tables/nop_batch_task.csv
@@ -1,2 +1,2 @@
-_chgType,SID,TASK_NAME,TASK_KEY,TASK_STATUS,START_TIME,END_TIME,TASK_PARAMS,EXEC_COUNT,WORKER_ID,INPUT_FILE_ID,FLOW_STEP_ID,RESTART_TIME,RESULT_STATUS,RESULT_CODE,RESULT_MSG,ERROR_STACK,COMPLETED_INDEX,COMPLETE_ITEM_COUNT,LOAD_RETRY_COUNT,LOAD_SKIP_COUNT,RETRY_ITEM_COUNT,PROCESS_ITEM_COUNT,SKIP_ITEM_COUNT,WRITE_ITEM_COUNT,VERSION,CREATED_BY,CREATE_TIME,UPDATED_BY,UPDATE_TIME,REMARK
-A,@var:NopBatchTask@sid,test.loadData,2024-01-01,30,@var:NopBatchTask@startTime,@var:NopBatchTask@endTime,,1,@var:NopBatchTask@workerId,,,,,,,,0,997,0,0,0,997,0,0,11,autotest-ref,*,autotest-ref,*,
+_chgType,SID,TASK_NAME,TASK_KEY,TASK_STATUS,START_TIME,END_TIME,TASK_PARAMS,EXEC_COUNT,WORKER_ID,INPUT_FILE_ID,FLOW_STEP_ID,FLOW_ID,RESTART_TIME,RESULT_STATUS,RESULT_CODE,RESULT_MSG,ERROR_STACK,COMPLETED_INDEX,COMPLETE_ITEM_COUNT,LOAD_RETRY_COUNT,LOAD_SKIP_COUNT,RETRY_ITEM_COUNT,PROCESS_ITEM_COUNT,SKIP_ITEM_COUNT,WRITE_ITEM_COUNT,VERSION,CREATED_BY,CREATE_TIME,UPDATED_BY,UPDATE_TIME,REMARK
+A,@var:NopBatchTask@sid,test.loadData,2024-01-01,30,@var:NopBatchTask@startTime,@var:NopBatchTask@endTime,,1,@var:NopBatchTask@workerId,,@var:NopBatchTask@flowStepId,@var:NopBatchTask@flowId,,,,,,0,997,0,0,0,997,0,0,11,autotest-ref,*,autotest-ref,*,
diff --git a/nop-batch/nop-batch-dsl/src/main/resources/_vfs/nop/batch/xlib/batch.xlib b/nop-batch/nop-batch-dsl/src/main/resources/_vfs/nop/batch/xlib/batch.xlib
@@ -31,9 +31,9 @@
                             <c:script>
                                const batchTaskManager = inject('nopBatchTaskManager');
                                const batchTaskContext = batchTaskManager.newBatchTaskContext(svcCtx,$scope);
-                               const stepRt = get('stepRt');
-                               batchTaskContext.setFlowId(stepRt?.taskId);
-                               batchTaskContext.setFlowStepId(stepRt?.stepId);
+                               const taskStepRt = get('taskStepRt');
+                               batchTaskContext.setFlowId(taskStepRt?.taskInstanceId);
+                               batchTaskContext.setFlowStepId(taskStepRt?.stepInstanceId);
                                return batchTaskBuilder.buildTask(batchTaskContext).executeAsync(batchTaskContext);
                             </c:script>
                          </c:ast>

diff --git a/nop-task/nop-task-core/src/main/java/io/nop/task/ITaskStepRuntime.java b/nop-task/nop-task-core/src/main/java/io/nop/task/ITaskStepRuntime.java
@@ -13,12 +13,12 @@ default IContext getContext() {
         return getTaskRuntime().getContext();
     }
 
-    default String getTaskId() {
-        return getState().getTaskId();
+    default String getTaskInstanceId() {
+        return getState().getTaskInstanceId();
     }
 
-    default String getStepId() {
-        return getState().getStepId();
+    default String getStepInstanceId() {
+        return getState().getStepInstanceId();
     }
 
     ITaskRuntime getTaskRuntime();

diff --git a/nop-task/nop-task-core/src/main/java/io/nop/task/ITaskStepState.java b/nop-task/nop-task-core/src/main/java/io/nop/task/ITaskStepState.java
@@ -16,13 +16,13 @@
  * 持久化状态包含返回结果（正常返回值以及抛出的异常），因此重复执行时发现结果已存在，则直接返回结果，跳过具体执行过程。
  */
 public interface ITaskStepState extends ITaskStateCommon {
-    String getTaskId();
+    String getTaskInstanceId();
 
-    String getStepId();
+    String getStepInstanceId();
 
-    void setTaskId(String taskId);
+    void setTaskInstanceId(String taskInstanceId);
 
-    void setStepId(String stepId);
+    void setStepInstanceId(String stepInstanceId);
 
     /**
      * stepName为静态定义的步骤name. runId为动态执行路径所确定的id. 两者结合在一起唯一确定一个TaskStepState

diff --git a/nop-task/nop-task-core/src/main/java/io/nop/task/TaskConstants.java b/nop-task/nop-task-core/src/main/java/io/nop/task/TaskConstants.java
@@ -44,7 +44,7 @@ public interface TaskConstants {
     String STEP_NAME_SUSPEND = "@suspend";
 
     String VAR_TASK_RT = "taskRt";
-    String VAR_STEP_RT = "stepRt";
+    String VAR_TASK_STEP_RT = "taskStepRt";
     String VAR_RESULT = "RESULT";
 
     String VAR_STEP_RESULTS = "STEP_RESULTS";

diff --git a/nop-task/nop-task-core/src/main/java/io/nop/task/impl/TaskStepRuntimeImpl.java b/nop-task/nop-task-core/src/main/java/io/nop/task/impl/TaskStepRuntimeImpl.java
@@ -37,7 +37,7 @@ public TaskStepRuntimeImpl(ITaskRuntime taskRt, ITaskStateStore stateStore, IEva
         this.taskRt = taskRt;
         this.stateStore = stateStore;
         this.scope = scope;
-        this.scope.setLocalValue(TaskConstants.VAR_STEP_RT, this);
+        this.scope.setLocalValue(TaskConstants.VAR_TASK_STEP_RT, this);
     }
 
     @Override

diff --git a/nop-task/nop-task-core/src/main/java/io/nop/task/state/DefaultTaskStateStore.java b/nop-task/nop-task-core/src/main/java/io/nop/task/state/DefaultTaskStateStore.java
@@ -33,6 +33,8 @@ public ITaskStepState newMainStepState(ITaskState taskState) {
     @Override
     public ITaskStepState newStepState(ITaskStepState parentState, String stepName, String stepType, ITaskRuntime taskRt) {
         TaskStepStateBean state = new TaskStepStateBean();
+        state.setStepInstanceId(StringHelper.generateUUID());
+        state.setTaskInstanceId(taskRt.getTaskInstanceId());
         String parentPath = parentState == null ? null : parentState.getStepPath();
         state.setStepPath(TaskStepHelper.buildStepPath(parentPath, stepName));
         state.setRunId(taskRt.newRunId());

diff --git a/nop-task/nop-task-core/src/main/java/io/nop/task/state/TaskStepStateBean.java b/nop-task/nop-task-core/src/main/java/io/nop/task/state/TaskStepStateBean.java
@@ -12,8 +12,8 @@
 import io.nop.task.TaskStepReturn;
 
 public class TaskStepStateBean extends AbstractTaskStateCommon implements ITaskStepState {
-    private String taskId;
-    private String stepId;
+    private String taskInstanceId;
+    private String stepInstanceId;
     private String stepPath;
     private int runId;
 
@@ -77,23 +77,23 @@ public void exception(Throwable exp) {
     }
 
     @Override
-    public String getTaskId() {
-        return taskId;
+    public String getTaskInstanceId() {
+        return taskInstanceId;
     }
 
     @Override
-    public void setTaskId(String taskId) {
-        this.taskId = taskId;
+    public void setTaskInstanceId(String taskInstanceId) {
+        this.taskInstanceId = taskInstanceId;
     }
 
     @Override
-    public String getStepId() {
-        return stepId;
+    public String getStepInstanceId() {
+        return stepInstanceId;
     }
 
     @Override
-    public void setStepId(String stepId) {
-        this.stepId = stepId;
+    public void setStepInstanceId(String stepInstanceId) {
+        this.stepInstanceId = stepInstanceId;
     }
 
     @Override
Original file line number	Diff line number	Diff line change
		@@ -1 +1 @@
		SID,TASK_NAME,TASK_KEY,TASK_STATUS,START_TIME,END_TIME,TASK_PARAMS,EXEC_COUNT,WORKER_ID,INPUT_FILE_ID,FLOW_STEP_ID,RESTART_TIME,RESULT_STATUS,RESULT_CODE,RESULT_MSG,ERROR_STACK,COMPLETED_INDEX,COMPLETE_ITEM_COUNT,LOAD_RETRY_COUNT,LOAD_SKIP_COUNT,RETRY_ITEM_COUNT,PROCESS_ITEM_COUNT,SKIP_ITEM_COUNT,WRITE_ITEM_COUNT,VERSION,CREATED_BY,CREATE_TIME,UPDATED_BY,UPDATE_TIME,REMARK
		SID,TASK_NAME,TASK_KEY,TASK_STATUS,START_TIME,END_TIME,TASK_PARAMS,EXEC_COUNT,WORKER_ID,INPUT_FILE_ID,FLOW_STEP_ID,FLOW_ID,RESTART_TIME,RESULT_STATUS,RESULT_CODE,RESULT_MSG,ERROR_STACK,COMPLETED_INDEX,COMPLETE_ITEM_COUNT,LOAD_RETRY_COUNT,LOAD_SKIP_COUNT,RETRY_ITEM_COUNT,PROCESS_ITEM_COUNT,SKIP_ITEM_COUNT,WRITE_ITEM_COUNT,VERSION,CREATED_BY,CREATE_TIME,UPDATED_BY,UPDATE_TIME,REMARK