You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by ca...@apache.org on 2024/08/28 11:27:45 UTC
(dolphinscheduler) branch dev updated: [DSIP-67] Use command to trigger workflow instance rather generate workflow instance (#16523)
This is an automated email from the ASF dual-hosted git repository.
caishunfeng pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 20a0e50cce [DSIP-67] Use command to trigger workflow instance rather generate workflow instance (#16523)
20a0e50cce is described below
commit 20a0e50cce84d024c7cadad09359ee736095c582
Author: Wenjun Ruan <we...@apache.org>
AuthorDate: Wed Aug 28 19:27:40 2024 +0800
[DSIP-67] Use command to trigger workflow instance rather generate workflow instance (#16523)
---
.../api/test/cases/ExecutorAPITest.java | 4 +-
.../api/test/cases/ProcessInstanceAPITest.java | 19 +--
.../api/controller/ExecutorController.java | 95 ++++++++-------
.../workflow/BackfillWorkflowExecutorDelegate.java | 108 ++++++++++-------
.../PauseWorkflowInstanceExecutorDelegate.java | 12 +-
.../StopWorkflowInstanceExecutorDelegate.java | 12 +-
.../workflow/TriggerWorkflowExecutorDelegate.java | 73 ++++++-----
.../api/service/ExecutorService.java | 9 +-
.../impl/AlertPluginInstanceServiceImpl.java | 9 +-
.../api/service/impl/ExecutorServiceImpl.java | 17 +--
.../api/service/impl/LoggerServiceImpl.java | 33 ++---
.../service/impl/MetricsCleanUpServiceImpl.java | 9 +-
.../api/service/impl/TaskInstanceServiceImpl.java | 27 ++---
.../api/validator/workflow/TriggerWorkflowDTO.java | 2 -
.../TriggerWorkflowRequestTransformer.java | 2 -
...kflowInstanceExecuteFunctionControllerTest.java | 10 +-
.../dolphinscheduler/common/enums/CommandType.java | 1 +
.../common/enums/WorkflowExecutionStatus.java | 2 +
.../dolphinscheduler/dao/entity/Command.java | 35 ++++--
.../dao/entity/TriggerRelation.java | 63 ----------
.../dao/mapper/TriggerRelationMapper.java | 72 -----------
.../dolphinscheduler/dao/repository/UserDao.java | 14 +--
.../dao/repository/impl/UserDaoImpl.java | 23 ++--
.../dao/mapper/TriggerRelationMapper.xml | 61 ----------
.../src/main/resources/sql/dolphinscheduler_h2.sql | 16 ---
.../main/resources/sql/dolphinscheduler_mysql.sql | 16 ---
.../resources/sql/dolphinscheduler_postgresql.sql | 15 ---
.../3.3.0_schema/mysql/dolphinscheduler_ddl.sql | 2 +
.../postgresql/dolphinscheduler_ddl.sql | 2 +
.../dao/mapper/TriggerRelationMapperTest.java | 134 ---------------------
...amicRpcClientProxyFactory.java => Clients.java} | 22 ++--
...lientProxyFactoryTest.java => ClientsTest.java} | 17 +--
.../extract/master/IWorkflowControlClient.java | 53 ++++++++
.../master/IWorkflowInstanceController.java | 39 ------
.../workflow/WorkflowBackfillTriggerRequest.java | 41 ++++---
.../WorkflowBackfillTriggerResponse.java} | 26 ++--
.../WorkflowInstancePauseRequest.java | 2 +-
.../WorkflowInstancePauseResponse.java | 2 +-
.../WorkflowInstanceStopRequest.java | 2 +-
.../WorkflowInstanceStopResponse.java | 2 +-
.../workflow/WorkflowManualTriggerRequest.java | 43 ++++---
.../WorkflowManualTriggerResponse.java} | 25 ++--
.../workflow/WorkflowScheduleTriggerRequest.java | 45 ++++---
.../WorkflowScheduleTriggerResponse.java} | 26 ++--
.../server/master/engine/TaskGroupCoordinator.java | 4 +-
.../master/engine/command/CommandEngine.java | 7 ++
.../command/handler/AbstractCommandHandler.java | 8 +-
.../handler/BackfillWorkflowCommandHandler.java | 21 ----
.../handler/ReRunWorkflowCommandHandler.java | 5 +
.../command/handler/RunWorkflowCommandHandler.java | 53 ++------
.../handler/ScheduleWorkflowCommandHandler.java | 17 ---
.../client/LogicTaskExecutorClientDelegator.java | 6 +-
.../PhysicalTaskExecutorClientDelegator.java | 6 +-
.../task/statemachine/AbstractTaskStateAction.java | 4 +-
.../listener/WorkflowSuccessLifecycleListener.java | 37 +++---
.../runnable/IWorkflowExecutionRunnable.java | 59 +++++++--
.../runnable/WorkflowExecutionRunnable.java | 50 ++------
.../runnable/WorkflowExecutionRunnableFactory.java | 8 +-
.../statemachine/IWorkflowStateAction.java | 1 +
.../statemachine/WorkflowStateActionFactory.java | 4 +-
.../statemachine/WorkflowSubmittedStateAction.java | 121 +++++++++++++++++++
.../workflow/trigger/AbstractWorkflowTrigger.java | 86 +++++++++++++
.../engine/workflow/trigger/IWorkflowTrigger.java | 19 ++-
.../workflow/trigger/WorkflowBackfillTrigger.java | 114 ++++++++++++++++++
.../workflow/trigger/WorkflowManualTrigger.java | 111 +++++++++++++++++
.../workflow/trigger/WorkflowScheduleTrigger.java | 106 ++++++++++++++++
.../master/rpc/TaskExecutionEventListenerImpl.java | 14 +--
.../server/master/rpc/WorkflowControlClient.java | 126 +++++++++++++++++++
.../master/rpc/WorkflowInstanceControllerImpl.java | 81 -------------
.../master/runner/WorkflowExecuteContext.java | 5 -
.../runner/dispatcher/WorkerTaskDispatcher.java | 4 +-
...ogicTaskInstanceExecuteDispatchEventSender.java | 4 +-
...LogicTaskInstanceExecuteRunningEventSender.java | 4 +-
...ogicTaskInstanceExecutionFailedEventSender.java | 4 +-
...ogicTaskInstanceExecutionKilledEventSender.java | 4 +-
...ogicTaskInstanceExecutionPausedEventSender.java | 4 +-
...gicTaskInstanceExecutionSuccessEventSender.java | 4 +-
.../runner/task/dynamic/DynamicLogicTask.java | 12 +-
.../task/subworkflow/SubWorkflowLogicTask.java | 20 +--
.../server/master/it/Repository.java | 14 +++
.../server/master/it/WorkflowOperator.java | 81 ++++++-------
.../master/it/cases/WorkflowInstancePauseIT.java | 91 +++++++++-----
.../master/it/cases/WorkflowInstanceStopIT.java | 70 ++++++-----
.../server/master/it/cases/WorkflowStartIT.java | 22 ++--
.../microbench/rpc/RpcBenchMarkTest.java | 7 +-
.../registry/api/RegistryClient.java | 12 ++
.../dolphinscheduler-scheduler-quartz/pom.xml | 5 +
.../scheduler/quartz/ProcessScheduleTask.java | 49 ++++----
.../service/process/ProcessServiceImpl.java | 12 +-
.../service/process/TriggerRelationService.java | 40 ------
.../process/TriggerRelationServiceImpl.java | 94 ---------------
.../service/process/ProcessServiceTest.java | 3 -
.../process/TriggerRelationServiceTest.java | 97 ---------------
.../workflow/components/dag/dag-startup-param.tsx | 5 +-
.../message/TaskExecutionDispatchEventSender.java | 4 +-
.../message/TaskExecutionFailedEventSender.java | 4 +-
.../message/TaskExecutionKilledEventSender.java | 4 +-
.../message/TaskExecutionPausedEventSender.java | 4 +-
.../message/TaskExecutionSuccessEventSender.java | 4 +-
.../TaskInstanceExecutionRunningEventSender.java | 12 +-
.../server/worker/runner/WorkerTaskExecutor.java | 9 +-
101 files changed, 1544 insertions(+), 1474 deletions(-)
diff --git a/dolphinscheduler-api-test/dolphinscheduler-api-test-case/src/test/java/org/apache/dolphinscheduler/api/test/cases/ExecutorAPITest.java b/dolphinscheduler-api-test/dolphinscheduler-api-test-case/src/test/java/org/apache/dolphinscheduler/api/test/cases/ExecutorAPITest.java
index 9d4ee34b5b..55639bd20f 100644
--- a/dolphinscheduler-api-test/dolphinscheduler-api-test-case/src/test/java/org/apache/dolphinscheduler/api/test/cases/ExecutorAPITest.java
+++ b/dolphinscheduler-api-test/dolphinscheduler-api-test-case/src/test/java/org/apache/dolphinscheduler/api/test/cases/ExecutorAPITest.java
@@ -73,7 +73,7 @@ public class ExecutorAPITest {
private static long processDefinitionCode;
- private static long triggerCode;
+ private static List<Integer> workflowInstanceIds;
@BeforeAll
public static void setup() {
@@ -138,7 +138,7 @@ public class ExecutorAPITest {
processDefinitionCode, scheduleTime, FailureStrategy.END, WarningType.NONE);
Assertions.assertTrue(startProcessInstanceResponse.getBody().getSuccess());
- triggerCode = (long) startProcessInstanceResponse.getBody().getData();
+ workflowInstanceIds = (List<Integer>) startProcessInstanceResponse.getBody().getData();
} catch (Exception e) {
log.error("failed", e);
Assertions.fail();
diff --git a/dolphinscheduler-api-test/dolphinscheduler-api-test-case/src/test/java/org/apache/dolphinscheduler/api/test/cases/ProcessInstanceAPITest.java b/dolphinscheduler-api-test/dolphinscheduler-api-test-case/src/test/java/org/apache/dolphinscheduler/api/test/cases/ProcessInstanceAPITest.java
index 35708f4955..1c95046c1c 100644
--- a/dolphinscheduler-api-test/dolphinscheduler-api-test-case/src/test/java/org/apache/dolphinscheduler/api/test/cases/ProcessInstanceAPITest.java
+++ b/dolphinscheduler-api-test/dolphinscheduler-api-test-case/src/test/java/org/apache/dolphinscheduler/api/test/cases/ProcessInstanceAPITest.java
@@ -19,6 +19,7 @@
package org.apache.dolphinscheduler.api.test.cases;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.dolphinscheduler.api.test.core.DolphinScheduler;
@@ -44,6 +45,7 @@ import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
@@ -51,7 +53,6 @@ import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.testcontainers.shaded.org.awaitility.Awaitility;
@@ -105,7 +106,7 @@ public class ProcessInstanceAPITest {
@Test
@Order(1)
- public void testQueryProcessInstancesByTriggerCode() {
+ public void testQueryProcessInstancesByWorkflowInstanceId() {
try {
// create test project
HttpResponse createProjectResponse = projectPage.createProject(loginUser, "project-test");
@@ -145,6 +146,10 @@ public class ProcessInstanceAPITest {
HttpResponse startProcessInstanceResponse = executorPage.startProcessInstance(loginUser, projectCode,
processDefinitionCode, scheduleTime, FailureStrategy.END, WarningType.NONE);
assertTrue(startProcessInstanceResponse.getBody().getSuccess());
+ final List<Integer> workflowInstanceIds = (List<Integer>) startProcessInstanceResponse.getBody().getData();
+
+ assertEquals(1, workflowInstanceIds.size());
+ processInstanceId = workflowInstanceIds.get(0);
// make sure process instance has completed and successfully persisted into db
Awaitility.await()
@@ -152,10 +157,11 @@ public class ProcessInstanceAPITest {
.untilAsserted(() -> {
// query workflow instance by trigger code
HttpResponse queryProcessInstanceListResponse =
- processInstancePage.queryProcessInstanceList(loginUser, projectCode, 1, 10);
+ processInstancePage.queryProcessInstanceById(loginUser, projectCode, processInstanceId);
assertTrue(queryProcessInstanceListResponse.getBody().getSuccess());
- assertTrue(queryProcessInstanceListResponse.getBody().getData().toString()
- .contains("test_import"));
+ final Map<String, Object> workflowInstance =
+ (Map<String, Object>) queryProcessInstanceListResponse.getBody().getData();
+ assertEquals("SUCCESS", workflowInstance.get("state"));
});
} catch (Exception e) {
log.error("failed", e);
@@ -174,7 +180,6 @@ public class ProcessInstanceAPITest {
@Test
@Order(3)
- @Disabled
public void testQueryTaskListByProcessId() {
HttpResponse queryTaskListByProcessIdResponse =
processInstancePage.queryTaskListByProcessId(loginUser, projectCode, processInstanceId);
@@ -184,7 +189,6 @@ public class ProcessInstanceAPITest {
@Test
@Order(4)
- @Disabled
public void testQueryProcessInstanceById() {
HttpResponse queryProcessInstanceByIdResponse =
processInstancePage.queryProcessInstanceById(loginUser, projectCode, processInstanceId);
@@ -194,7 +198,6 @@ public class ProcessInstanceAPITest {
@Test
@Order(5)
- @Disabled
public void testDeleteProcessInstanceById() {
HttpResponse deleteProcessInstanceByIdResponse =
processInstancePage.deleteProcessInstanceById(loginUser, projectCode, processInstanceId);
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java
index 5169894c68..218bb6263d 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java
@@ -64,6 +64,8 @@ import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;
+import com.google.common.collect.Lists;
+
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.Parameters;
@@ -129,27 +131,27 @@ public class ExecutorController extends BaseController {
@ResponseStatus(HttpStatus.OK)
@ApiException(START_PROCESS_INSTANCE_ERROR)
@OperatorLog(auditType = AuditType.PROCESS_START)
- public Result<Long> triggerWorkflowDefinition(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
- @RequestParam(value = "processDefinitionCode") long processDefinitionCode,
- @RequestParam(value = "scheduleTime") String scheduleTime,
- @RequestParam(value = "failureStrategy") FailureStrategy failureStrategy,
- @RequestParam(value = "startNodeList", required = false) String startNodeList,
- @RequestParam(value = "taskDependType", required = false, defaultValue = "TASK_POST") TaskDependType taskDependType,
- @RequestParam(value = "execType", required = false, defaultValue = "START_PROCESS") CommandType execType,
- @RequestParam(value = "warningType") WarningType warningType,
- @RequestParam(value = "warningGroupId", required = false) Integer warningGroupId,
- @RequestParam(value = "runMode", required = false) RunMode runMode,
- @RequestParam(value = "processInstancePriority", required = false) Priority processInstancePriority,
- @RequestParam(value = "workerGroup", required = false, defaultValue = "default") String workerGroup,
- @RequestParam(value = "tenantCode", required = false, defaultValue = "default") String tenantCode,
- @RequestParam(value = "environmentCode", required = false, defaultValue = "-1") Long environmentCode,
- @RequestParam(value = "startParams", required = false) String startParams,
- @RequestParam(value = "expectedParallelismNumber", required = false) Integer expectedParallelismNumber,
- @RequestParam(value = "dryRun", defaultValue = "0", required = false) int dryRun,
- @RequestParam(value = "testFlag", defaultValue = "0") int testFlag,
- @RequestParam(value = "complementDependentMode", required = false) ComplementDependentMode complementDependentMode,
- @RequestParam(value = "allLevelDependent", required = false, defaultValue = "false") boolean allLevelDependent,
- @RequestParam(value = "executionOrder", required = false) ExecutionOrder executionOrder) {
+ public Result<List<Integer>> triggerWorkflowDefinition(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
+ @RequestParam(value = "processDefinitionCode") long processDefinitionCode,
+ @RequestParam(value = "scheduleTime") String scheduleTime,
+ @RequestParam(value = "failureStrategy") FailureStrategy failureStrategy,
+ @RequestParam(value = "startNodeList", required = false) String startNodeList,
+ @RequestParam(value = "taskDependType", required = false, defaultValue = "TASK_POST") TaskDependType taskDependType,
+ @RequestParam(value = "execType", required = false, defaultValue = "START_PROCESS") CommandType execType,
+ @RequestParam(value = "warningType") WarningType warningType,
+ @RequestParam(value = "warningGroupId", required = false) Integer warningGroupId,
+ @RequestParam(value = "runMode", required = false) RunMode runMode,
+ @RequestParam(value = "processInstancePriority", required = false) Priority processInstancePriority,
+ @RequestParam(value = "workerGroup", required = false, defaultValue = "default") String workerGroup,
+ @RequestParam(value = "tenantCode", required = false, defaultValue = "default") String tenantCode,
+ @RequestParam(value = "environmentCode", required = false, defaultValue = "-1") Long environmentCode,
+ @RequestParam(value = "startParams", required = false) String startParams,
+ @RequestParam(value = "expectedParallelismNumber", required = false) Integer expectedParallelismNumber,
+ @RequestParam(value = "dryRun", defaultValue = "0", required = false) int dryRun,
+ @RequestParam(value = "testFlag", defaultValue = "0") int testFlag,
+ @RequestParam(value = "complementDependentMode", required = false) ComplementDependentMode complementDependentMode,
+ @RequestParam(value = "allLevelDependent", required = false, defaultValue = "false") boolean allLevelDependent,
+ @RequestParam(value = "executionOrder", required = false) ExecutionOrder executionOrder) {
switch (execType) {
case START_PROCESS:
@@ -170,7 +172,8 @@ public class ExecutorController extends BaseController {
.dryRun(Flag.of(dryRun))
.testFlag(Flag.of(testFlag))
.build();
- return Result.success(execService.triggerWorkflowDefinition(workflowTriggerRequest));
+ return Result
+ .success(Lists.newArrayList(execService.triggerWorkflowDefinition(workflowTriggerRequest)));
case COMPLEMENT_DATA:
final WorkflowBackFillRequest workflowBackFillRequest = WorkflowBackFillRequest.builder()
.loginUser(loginUser)
@@ -250,35 +253,35 @@ public class ExecutorController extends BaseController {
@ResponseStatus(HttpStatus.OK)
@ApiException(BATCH_START_PROCESS_INSTANCE_ERROR)
@OperatorLog(auditType = AuditType.PROCESS_BATCH_START)
- public Result<List<Long>> batchTriggerWorkflowDefinitions(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
- @RequestParam(value = "processDefinitionCodes") String processDefinitionCodes,
- @RequestParam(value = "scheduleTime") String scheduleTime,
- @RequestParam(value = "failureStrategy") FailureStrategy failureStrategy,
- @RequestParam(value = "startNodeList", required = false) String startNodeList,
- @RequestParam(value = "taskDependType", required = false) TaskDependType taskDependType,
- @RequestParam(value = "execType", required = false) CommandType execType,
- @RequestParam(value = "warningType") WarningType warningType,
- @RequestParam(value = "warningGroupId", required = false) Integer warningGroupId,
- @RequestParam(value = "runMode", required = false) RunMode runMode,
- @RequestParam(value = "processInstancePriority", required = false) Priority processInstancePriority,
- @RequestParam(value = "workerGroup", required = false, defaultValue = "default") String workerGroup,
- @RequestParam(value = "tenantCode", required = false, defaultValue = "default") String tenantCode,
- @RequestParam(value = "environmentCode", required = false, defaultValue = "-1") Long environmentCode,
- @RequestParam(value = "startParams", required = false) String startParams,
- @RequestParam(value = "expectedParallelismNumber", required = false) Integer expectedParallelismNumber,
- @RequestParam(value = "dryRun", defaultValue = "0", required = false) int dryRun,
- @RequestParam(value = "testFlag", defaultValue = "0") int testFlag,
- @RequestParam(value = "complementDependentMode", required = false) ComplementDependentMode complementDependentMode,
- @RequestParam(value = "allLevelDependent", required = false, defaultValue = "false") boolean allLevelDependent,
- @RequestParam(value = "executionOrder", required = false) ExecutionOrder executionOrder) {
+ public Result<List<Integer>> batchTriggerWorkflowDefinitions(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
+ @RequestParam(value = "processDefinitionCodes") String processDefinitionCodes,
+ @RequestParam(value = "scheduleTime") String scheduleTime,
+ @RequestParam(value = "failureStrategy") FailureStrategy failureStrategy,
+ @RequestParam(value = "startNodeList", required = false) String startNodeList,
+ @RequestParam(value = "taskDependType", required = false) TaskDependType taskDependType,
+ @RequestParam(value = "execType", required = false) CommandType execType,
+ @RequestParam(value = "warningType") WarningType warningType,
+ @RequestParam(value = "warningGroupId", required = false) Integer warningGroupId,
+ @RequestParam(value = "runMode", required = false) RunMode runMode,
+ @RequestParam(value = "processInstancePriority", required = false) Priority processInstancePriority,
+ @RequestParam(value = "workerGroup", required = false, defaultValue = "default") String workerGroup,
+ @RequestParam(value = "tenantCode", required = false, defaultValue = "default") String tenantCode,
+ @RequestParam(value = "environmentCode", required = false, defaultValue = "-1") Long environmentCode,
+ @RequestParam(value = "startParams", required = false) String startParams,
+ @RequestParam(value = "expectedParallelismNumber", required = false) Integer expectedParallelismNumber,
+ @RequestParam(value = "dryRun", defaultValue = "0", required = false) int dryRun,
+ @RequestParam(value = "testFlag", defaultValue = "0") int testFlag,
+ @RequestParam(value = "complementDependentMode", required = false) ComplementDependentMode complementDependentMode,
+ @RequestParam(value = "allLevelDependent", required = false, defaultValue = "false") boolean allLevelDependent,
+ @RequestParam(value = "executionOrder", required = false) ExecutionOrder executionOrder) {
List<Long> workflowDefinitionCodes = Arrays.stream(processDefinitionCodes.split(Constants.COMMA))
.map(Long::parseLong)
.collect(Collectors.toList());
- List<Long> result = new ArrayList<>();
+ List<Integer> result = new ArrayList<>();
for (Long workflowDefinitionCode : workflowDefinitionCodes) {
- Result<Long> triggerCodeResult = triggerWorkflowDefinition(loginUser,
+ Result<List<Integer>> workflowInstanceIds = triggerWorkflowDefinition(loginUser,
workflowDefinitionCode,
scheduleTime,
failureStrategy,
@@ -299,7 +302,7 @@ public class ExecutorController extends BaseController {
complementDependentMode,
allLevelDependent,
executionOrder);
- result.add(triggerCodeResult.getData());
+ result.addAll(workflowInstanceIds.getData());
}
return Result.success(result);
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/BackfillWorkflowExecutorDelegate.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/BackfillWorkflowExecutorDelegate.java
index 478d6b98d9..b8e57128b3 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/BackfillWorkflowExecutorDelegate.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/BackfillWorkflowExecutorDelegate.java
@@ -17,20 +17,25 @@
package org.apache.dolphinscheduler.api.executor.workflow;
+import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.validator.workflow.BackfillWorkflowDTO;
import org.apache.dolphinscheduler.common.enums.ComplementDependentMode;
import org.apache.dolphinscheduler.common.enums.ExecutionOrder;
import org.apache.dolphinscheduler.common.enums.RunMode;
+import org.apache.dolphinscheduler.common.model.Server;
import org.apache.dolphinscheduler.common.utils.DateUtils;
-import org.apache.dolphinscheduler.common.utils.JSONUtils;
-import org.apache.dolphinscheduler.dao.entity.Command;
+import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.repository.CommandDao;
-import org.apache.dolphinscheduler.extract.master.command.BackfillWorkflowCommandParam;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
+import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowBackfillTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowBackfillTriggerResponse;
+import org.apache.dolphinscheduler.registry.api.RegistryClient;
+import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
import org.apache.dolphinscheduler.service.process.ProcessService;
import java.time.ZonedDateTime;
import java.util.Collections;
-import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;
@@ -43,7 +48,7 @@ import com.google.common.collect.Lists;
@Slf4j
@Component
-public class BackfillWorkflowExecutorDelegate implements IExecutorDelegate<BackfillWorkflowDTO, Void> {
+public class BackfillWorkflowExecutorDelegate implements IExecutorDelegate<BackfillWorkflowDTO, List<Integer>> {
@Autowired
private CommandDao commandDao;
@@ -51,18 +56,20 @@ public class BackfillWorkflowExecutorDelegate implements IExecutorDelegate<Backf
@Autowired
private ProcessService processService;
+ @Autowired
+ private RegistryClient registryClient;
+
@Override
- public Void execute(final BackfillWorkflowDTO backfillWorkflowDTO) {
+ public List<Integer> execute(final BackfillWorkflowDTO backfillWorkflowDTO) {
// todo: directly call the master api to do backfill
if (backfillWorkflowDTO.getBackfillParams().getRunMode() == RunMode.RUN_MODE_SERIAL) {
- doSerialBackfillWorkflow(backfillWorkflowDTO);
+ return doSerialBackfillWorkflow(backfillWorkflowDTO);
} else {
- doParallemBackfillWorkflow(backfillWorkflowDTO);
+ return doParallemBackfillWorkflow(backfillWorkflowDTO);
}
- return null;
}
- private void doSerialBackfillWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO) {
+ private List<Integer> doSerialBackfillWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO) {
final BackfillWorkflowDTO.BackfillParamsDTO backfillParams = backfillWorkflowDTO.getBackfillParams();
final List<ZonedDateTime> backfillTimeList = backfillParams.getBackfillDateList();
if (backfillParams.getExecutionOrder() == ExecutionOrder.DESC_ORDER) {
@@ -71,17 +78,13 @@ public class BackfillWorkflowExecutorDelegate implements IExecutorDelegate<Backf
Collections.sort(backfillTimeList);
}
- final BackfillWorkflowCommandParam backfillWorkflowCommandParam = BackfillWorkflowCommandParam.builder()
- .commandParams(backfillWorkflowDTO.getStartParamList())
- .startNodes(backfillWorkflowDTO.getStartNodes())
- .backfillTimeList(backfillTimeList.stream().map(DateUtils::dateToString).collect(Collectors.toList()))
- .timeZone(DateUtils.getTimezone())
- .build();
-
- doCreateCommand(backfillWorkflowDTO, backfillWorkflowCommandParam);
+ final Integer workflowInstanceId = doBackfillWorkflow(
+ backfillWorkflowDTO,
+ backfillTimeList.stream().map(DateUtils::dateToString).collect(Collectors.toList()));
+ return Lists.newArrayList(workflowInstanceId);
}
- private void doParallemBackfillWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO) {
+ private List<Integer> doParallemBackfillWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO) {
final BackfillWorkflowDTO.BackfillParamsDTO backfillParams = backfillWorkflowDTO.getBackfillParams();
Integer expectedParallelismNumber = backfillParams.getExpectedParallelismNumber();
@@ -93,47 +96,60 @@ public class BackfillWorkflowExecutorDelegate implements IExecutorDelegate<Backf
}
log.info("In parallel mode, current expectedParallelismNumber:{}", expectedParallelismNumber);
+ final List<Integer> workflowInstanceIdList = Lists.newArrayList();
for (List<ZonedDateTime> stringDate : Lists.partition(listDate, expectedParallelismNumber)) {
- final BackfillWorkflowCommandParam backfillWorkflowCommandParam = BackfillWorkflowCommandParam.builder()
- .commandParams(backfillWorkflowDTO.getStartParamList())
- .startNodes(backfillWorkflowDTO.getStartNodes())
- .backfillTimeList(stringDate.stream().map(DateUtils::dateToString).collect(Collectors.toList()))
- .timeZone(DateUtils.getTimezone())
- .build();
- doCreateCommand(backfillWorkflowDTO, backfillWorkflowCommandParam);
+ final Integer workflowInstanceId = doBackfillWorkflow(
+ backfillWorkflowDTO,
+ stringDate.stream().map(DateUtils::dateToString).collect(Collectors.toList()));
+ workflowInstanceIdList.add(workflowInstanceId);
}
+ return workflowInstanceIdList;
}
- private void doCreateCommand(final BackfillWorkflowDTO backfillWorkflowDTO,
- final BackfillWorkflowCommandParam backfillWorkflowCommandParam) {
- List<String> backfillTimeList = backfillWorkflowCommandParam.getBackfillTimeList();
- final Command command = Command.builder()
- .commandType(backfillWorkflowDTO.getExecType())
- .processDefinitionCode(backfillWorkflowDTO.getWorkflowDefinition().getCode())
- .processDefinitionVersion(backfillWorkflowDTO.getWorkflowDefinition().getVersion())
- .executorId(backfillWorkflowDTO.getLoginUser().getId())
- .scheduleTime(DateUtils.stringToDate(backfillTimeList.get(0)))
- .commandParam(JSONUtils.toJsonString(backfillWorkflowCommandParam))
- .taskDependType(backfillWorkflowDTO.getTaskDependType())
+ private Integer doBackfillWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO,
+ final List<String> backfillTimeList) {
+ final Server masterServer = registryClient.getRandomServer(RegistryNodeType.MASTER).orElse(null);
+ if (masterServer == null) {
+ throw new ServiceException("no master server available");
+ }
+
+ final ProcessDefinition workflowDefinition = backfillWorkflowDTO.getWorkflowDefinition();
+ final WorkflowBackfillTriggerRequest backfillTriggerRequest = WorkflowBackfillTriggerRequest.builder()
+ .userId(backfillWorkflowDTO.getLoginUser().getId())
+ .backfillTimeList(backfillTimeList)
+ .workflowCode(workflowDefinition.getCode())
+ .workflowVersion(workflowDefinition.getVersion())
+ .startNodes(backfillWorkflowDTO.getStartNodes())
.failureStrategy(backfillWorkflowDTO.getFailureStrategy())
+ .taskDependType(backfillWorkflowDTO.getTaskDependType())
+ .execType(backfillWorkflowDTO.getExecType())
.warningType(backfillWorkflowDTO.getWarningType())
.warningGroupId(backfillWorkflowDTO.getWarningGroupId())
- .startTime(new Date())
- .processInstancePriority(backfillWorkflowDTO.getWorkflowInstancePriority())
- .updateTime(new Date())
+ .workflowInstancePriority(backfillWorkflowDTO.getWorkflowInstancePriority())
.workerGroup(backfillWorkflowDTO.getWorkerGroup())
.tenantCode(backfillWorkflowDTO.getTenantCode())
- .dryRun(backfillWorkflowDTO.getDryRun().getCode())
- .testFlag(backfillWorkflowDTO.getTestFlag().getCode())
+ .environmentCode(backfillWorkflowDTO.getEnvironmentCode())
+ .startParamList(backfillWorkflowDTO.getStartParamList())
+ .dryRun(backfillWorkflowDTO.getDryRun())
+ .testFlag(backfillWorkflowDTO.getTestFlag())
.build();
- commandDao.insert(command);
+
+ final WorkflowBackfillTriggerResponse backfillTriggerResponse = Clients
+ .withService(IWorkflowControlClient.class)
+ .withHost(masterServer.getHost() + ":" + masterServer.getPort())
+ .backfillTriggerWorkflow(backfillTriggerRequest);
+ if (!backfillTriggerResponse.isSuccess()) {
+ throw new ServiceException("Backfill workflow failed: " + backfillTriggerResponse.getMessage());
+ }
final BackfillWorkflowDTO.BackfillParamsDTO backfillParams = backfillWorkflowDTO.getBackfillParams();
if (backfillParams.getBackfillDependentMode() == ComplementDependentMode.ALL_DEPENDENT) {
- doBackfillDependentWorkflow(backfillWorkflowCommandParam, command);
+ doBackfillDependentWorkflow(backfillWorkflowDTO, backfillTimeList);
}
+ return backfillTriggerResponse.getWorkflowInstanceId();
}
- private void doBackfillDependentWorkflow(final BackfillWorkflowCommandParam backfillWorkflowCommandParam,
- final Command backfillCommand) {
+ private void doBackfillDependentWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO,
+ final List<String> backfillTimeList) {
+ // todo:
}
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/PauseWorkflowInstanceExecutorDelegate.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/PauseWorkflowInstanceExecutorDelegate.java
index 4036f8b212..296f9919a6 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/PauseWorkflowInstanceExecutorDelegate.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/PauseWorkflowInstanceExecutorDelegate.java
@@ -22,10 +22,10 @@ import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
-import org.apache.dolphinscheduler.extract.master.IWorkflowInstanceController;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstancePauseRequest;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstancePauseResponse;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
+import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstancePauseRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstancePauseResponse;
import lombok.extern.slf4j.Slf4j;
@@ -80,8 +80,8 @@ public class PauseWorkflowInstanceExecutorDelegate
private void pauseInMaster(ProcessInstance workflowInstance) {
try {
- final WorkflowInstancePauseResponse pauseResponse = SingletonJdkDynamicRpcClientProxyFactory
- .withService(IWorkflowInstanceController.class)
+ final WorkflowInstancePauseResponse pauseResponse = Clients
+ .withService(IWorkflowControlClient.class)
.withHost(workflowInstance.getHost())
.pauseWorkflowInstance(new WorkflowInstancePauseRequest(workflowInstance.getId()));
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/StopWorkflowInstanceExecutorDelegate.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/StopWorkflowInstanceExecutorDelegate.java
index 287c738367..80f21a0cdd 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/StopWorkflowInstanceExecutorDelegate.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/StopWorkflowInstanceExecutorDelegate.java
@@ -22,10 +22,10 @@ import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
-import org.apache.dolphinscheduler.extract.master.IWorkflowInstanceController;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStopRequest;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStopResponse;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
+import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopResponse;
import lombok.extern.slf4j.Slf4j;
@@ -81,8 +81,8 @@ public class StopWorkflowInstanceExecutorDelegate
void stopInMaster(ProcessInstance workflowInstance) {
try {
- final WorkflowInstanceStopResponse stopResponse = SingletonJdkDynamicRpcClientProxyFactory
- .withService(IWorkflowInstanceController.class)
+ final WorkflowInstanceStopResponse stopResponse = Clients
+ .withService(IWorkflowControlClient.class)
.withHost(workflowInstance.getHost())
.stopWorkflowInstance(new WorkflowInstanceStopRequest(workflowInstance.getId()));
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/TriggerWorkflowExecutorDelegate.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/TriggerWorkflowExecutorDelegate.java
index 15b31a8a31..4270f140e2 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/TriggerWorkflowExecutorDelegate.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/TriggerWorkflowExecutorDelegate.java
@@ -17,55 +17,62 @@
package org.apache.dolphinscheduler.api.executor.workflow;
+import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.validator.workflow.TriggerWorkflowDTO;
-import org.apache.dolphinscheduler.common.utils.DateUtils;
-import org.apache.dolphinscheduler.common.utils.JSONUtils;
-import org.apache.dolphinscheduler.dao.entity.Command;
-import org.apache.dolphinscheduler.dao.repository.CommandDao;
-import org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
-import org.apache.dolphinscheduler.service.process.TriggerRelationService;
+import org.apache.dolphinscheduler.common.model.Server;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
+import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowManualTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowManualTriggerResponse;
+import org.apache.dolphinscheduler.registry.api.RegistryClient;
+import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
-import java.util.Date;
+import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
+@Slf4j
@Component
-public class TriggerWorkflowExecutorDelegate implements IExecutorDelegate<TriggerWorkflowDTO, Void> {
+public class TriggerWorkflowExecutorDelegate implements IExecutorDelegate<TriggerWorkflowDTO, Integer> {
@Autowired
- private CommandDao commandDao;
-
- @Autowired
- private TriggerRelationService triggerRelationService;
+ private RegistryClient registryClient;
@Override
- public Void execute(TriggerWorkflowDTO triggerWorkflowDTO) {
- final RunWorkflowCommandParam runWorkflowCommandParam =
- RunWorkflowCommandParam.builder()
- .commandParams(triggerWorkflowDTO.getStartParamList())
- .startNodes(triggerWorkflowDTO.getStartNodes())
- .timeZone(DateUtils.getTimezone())
- .build();
- final Command command = Command.builder()
- .commandType(triggerWorkflowDTO.getExecType())
- .processDefinitionCode(triggerWorkflowDTO.getWorkflowDefinition().getCode())
- .processDefinitionVersion(triggerWorkflowDTO.getWorkflowDefinition().getVersion())
- .executorId(triggerWorkflowDTO.getLoginUser().getId())
- .commandParam(JSONUtils.toJsonString(runWorkflowCommandParam))
- .taskDependType(triggerWorkflowDTO.getTaskDependType())
+ public Integer execute(final TriggerWorkflowDTO triggerWorkflowDTO) {
+ final Server masterServer = registryClient.getRandomServer(RegistryNodeType.MASTER).orElse(null);
+ if (masterServer == null) {
+ throw new ServiceException("no master server available");
+ }
+ final WorkflowManualTriggerResponse workflowManualTriggerResponse = Clients
+ .withService(IWorkflowControlClient.class)
+ .withHost(masterServer.getHost() + ":" + masterServer.getPort())
+ .manualTriggerWorkflow(transform2WorkflowTriggerRequest(triggerWorkflowDTO));
+ if (!workflowManualTriggerResponse.isSuccess()) {
+ throw new ServiceException("Trigger workflow failed: " + workflowManualTriggerResponse.getMessage());
+ }
+ return workflowManualTriggerResponse.getWorkflowInstanceId();
+
+ }
+
+ private WorkflowManualTriggerRequest transform2WorkflowTriggerRequest(TriggerWorkflowDTO triggerWorkflowDTO) {
+ return WorkflowManualTriggerRequest.builder()
+ .userId(triggerWorkflowDTO.getLoginUser().getId())
+ .workflowDefinitionCode(triggerWorkflowDTO.getWorkflowDefinition().getCode())
+ .workflowDefinitionVersion(triggerWorkflowDTO.getWorkflowDefinition().getVersion())
+ .startNodes(triggerWorkflowDTO.getStartNodes())
.failureStrategy(triggerWorkflowDTO.getFailureStrategy())
+ .taskDependType(triggerWorkflowDTO.getTaskDependType())
.warningType(triggerWorkflowDTO.getWarningType())
.warningGroupId(triggerWorkflowDTO.getWarningGroupId())
- .startTime(new Date())
- .processInstancePriority(triggerWorkflowDTO.getWorkflowInstancePriority())
- .updateTime(new Date())
+ .workflowInstancePriority(triggerWorkflowDTO.getWorkflowInstancePriority())
.workerGroup(triggerWorkflowDTO.getWorkerGroup())
.tenantCode(triggerWorkflowDTO.getTenantCode())
- .dryRun(triggerWorkflowDTO.getDryRun().getCode())
- .testFlag(triggerWorkflowDTO.getTestFlag().getCode())
+ .environmentCode(triggerWorkflowDTO.getEnvironmentCode())
+ .startParamList(triggerWorkflowDTO.getStartParamList())
+ .dryRun(triggerWorkflowDTO.getDryRun())
+ .testFlag(triggerWorkflowDTO.getTestFlag())
.build();
- commandDao.insert(command);
- return null;
}
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java
index 306f8f87e5..30fed75cec 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java
@@ -25,19 +25,20 @@ import org.apache.dolphinscheduler.common.enums.TaskDependType;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.User;
+import java.util.List;
import java.util.Map;
public interface ExecutorService {
/**
- * Trigger the workflow and return the trigger code.
+ * Trigger the workflow and return the workflow instance id.
*/
- Long triggerWorkflowDefinition(final WorkflowTriggerRequest workflowTriggerRequest);
+ Integer triggerWorkflowDefinition(final WorkflowTriggerRequest workflowTriggerRequest);
/**
- * Backfill the workflow and return the trigger code.
+ * Backfill the workflow and return the workflow instance ids.
*/
- Long backfillWorkflowDefinition(final WorkflowBackFillRequest workflowBackFillRequest);
+ List<Integer> backfillWorkflowDefinition(final WorkflowBackFillRequest workflowBackFillRequest);
/**
* check whether the process definition can be executed
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/AlertPluginInstanceServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/AlertPluginInstanceServiceImpl.java
index 5da3ab30d6..1b69c19294 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/AlertPluginInstanceServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/AlertPluginInstanceServiceImpl.java
@@ -39,7 +39,7 @@ import org.apache.dolphinscheduler.dao.mapper.PluginDefineMapper;
import org.apache.dolphinscheduler.extract.alert.IAlertOperator;
import org.apache.dolphinscheduler.extract.alert.request.AlertSendResponse;
import org.apache.dolphinscheduler.extract.alert.request.AlertTestSendRequest;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.base.utils.Host;
import org.apache.dolphinscheduler.registry.api.RegistryClient;
import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
@@ -307,9 +307,10 @@ public class AlertPluginInstanceServiceImpl extends BaseServiceImpl implements A
AlertSendResponse alertSendResponse;
try {
- IAlertOperator alertOperator = SingletonJdkDynamicRpcClientProxyFactory
- .getProxyClient(alertServerAddress.getAddress(), IAlertOperator.class);
- alertSendResponse = alertOperator.sendTestAlert(alertTestSendRequest);
+ alertSendResponse = Clients
+ .withService(IAlertOperator.class)
+ .withHost(alertServerAddress.getAddress())
+ .sendTestAlert(alertTestSendRequest);
log.info("Send alert to: {} successfully, response: {}", alertServerAddress, alertSendResponse);
} catch (Exception e) {
log.error("Send alert: {} to: {} failed", alertTestSendRequest, alertServerAddress, e);
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
index 176f9326d1..be3172a64c 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
@@ -81,7 +81,6 @@ import org.apache.dolphinscheduler.service.command.CommandService;
import org.apache.dolphinscheduler.service.cron.CronUtils;
import org.apache.dolphinscheduler.service.exceptions.CronParseException;
import org.apache.dolphinscheduler.service.process.ProcessService;
-import org.apache.dolphinscheduler.service.process.TriggerRelationService;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.collections4.CollectionUtils;
@@ -158,9 +157,6 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
@Autowired
private WorkerGroupService workerGroupService;
- @Autowired
- private TriggerRelationService triggerRelationService;
-
@Autowired
private TenantMapper tenantMapper;
@@ -184,22 +180,19 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
@Override
@Transactional
- public Long triggerWorkflowDefinition(final WorkflowTriggerRequest workflowTriggerRequest) {
- final TriggerWorkflowDTO triggerWorkflowDTO =
- triggerWorkflowRequestTransformer.transform(workflowTriggerRequest);
+ public Integer triggerWorkflowDefinition(final WorkflowTriggerRequest triggerRequest) {
+ final TriggerWorkflowDTO triggerWorkflowDTO = triggerWorkflowRequestTransformer.transform(triggerRequest);
triggerWorkflowDTOValidator.validate(triggerWorkflowDTO);
- executorClient.triggerWorkflowDefinition().execute(triggerWorkflowDTO);
- return triggerWorkflowDTO.getTriggerCode();
+ return executorClient.triggerWorkflowDefinition().execute(triggerWorkflowDTO);
}
@Override
@Transactional
- public Long backfillWorkflowDefinition(final WorkflowBackFillRequest workflowBackFillRequest) {
+ public List<Integer> backfillWorkflowDefinition(final WorkflowBackFillRequest workflowBackFillRequest) {
final BackfillWorkflowDTO backfillWorkflowDTO =
backfillWorkflowRequestTransformer.transform(workflowBackFillRequest);
backfillWorkflowDTOValidator.validate(backfillWorkflowDTO);
- executorClient.backfillWorkflowDefinition().execute(backfillWorkflowDTO);
- return backfillWorkflowDTO.getTriggerCode();
+ return executorClient.backfillWorkflowDefinition().execute(backfillWorkflowDTO);
}
/**
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/LoggerServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/LoggerServiceImpl.java
index c0ecb0e9b5..b81e1d4642 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/LoggerServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/LoggerServiceImpl.java
@@ -34,7 +34,7 @@ import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.common.ILogService;
import org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadRequest;
import org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadResponse;
@@ -77,9 +77,9 @@ public class LoggerServiceImpl extends BaseServiceImpl implements LoggerService
* view log
*
* @param loginUser login user
- * @param taskInstId task instance id
+ * @param taskInstId task instance id
* @param skipLineNum skip line number
- * @param limit limit
+ * @param limit limit
* @return log string data
*/
@Override
@@ -107,7 +107,7 @@ public class LoggerServiceImpl extends BaseServiceImpl implements LoggerService
/**
* get log size
*
- * @param loginUser login user
+ * @param loginUser login user
* @param taskInstId task instance id
* @return log byte array
*/
@@ -179,9 +179,9 @@ public class LoggerServiceImpl extends BaseServiceImpl implements LoggerService
/**
* query log
*
- * @param taskInstance task instance
- * @param skipLineNum skip line number
- * @param limit limit
+ * @param taskInstance task instance
+ * @param skipLineNum skip line number
+ * @param limit limit
* @return log string data
*/
private String queryLog(TaskInstance taskInstance, int skipLineNum, int limit) {
@@ -202,8 +202,6 @@ public class LoggerServiceImpl extends BaseServiceImpl implements LoggerService
sb.append(head);
}
- ILogService iLogService =
- SingletonJdkDynamicRpcClientProxyFactory.getProxyClient(taskInstance.getHost(), ILogService.class);
try {
TaskInstanceLogPageQueryRequest request = TaskInstanceLogPageQueryRequest.builder()
.taskInstanceId(taskInstance.getId())
@@ -211,7 +209,10 @@ public class LoggerServiceImpl extends BaseServiceImpl implements LoggerService
.skipLineNum(skipLineNum)
.limit(limit)
.build();
- TaskInstanceLogPageQueryResponse response = iLogService.pageQueryTaskInstanceLog(request);
+ final TaskInstanceLogPageQueryResponse response = Clients
+ .withService(ILogService.class)
+ .withHost(taskInstance.getHost())
+ .pageQueryTaskInstanceLog(request);
String logContent = response.getLogContent();
if (logContent != null) {
sb.append(logContent);
@@ -239,12 +240,14 @@ public class LoggerServiceImpl extends BaseServiceImpl implements LoggerService
byte[] logBytes;
- ILogService iLogService =
- SingletonJdkDynamicRpcClientProxyFactory.getProxyClient(taskInstance.getHost(), ILogService.class);
try {
- TaskInstanceLogFileDownloadRequest request =
- new TaskInstanceLogFileDownloadRequest(taskInstance.getId(), logPath);
- TaskInstanceLogFileDownloadResponse response = iLogService.getTaskInstanceWholeLogFileBytes(request);
+ final TaskInstanceLogFileDownloadRequest request = new TaskInstanceLogFileDownloadRequest(
+ taskInstance.getId(),
+ logPath);
+ final TaskInstanceLogFileDownloadResponse response = Clients
+ .withService(ILogService.class)
+ .withHost(taskInstance.getHost())
+ .getTaskInstanceWholeLogFileBytes(request);
logBytes = response.getLogBytes();
return Bytes.concat(head, logBytes);
} catch (Exception ex) {
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/MetricsCleanUpServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/MetricsCleanUpServiceImpl.java
index f1ac0579ff..3ac661c70e 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/MetricsCleanUpServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/MetricsCleanUpServiceImpl.java
@@ -20,7 +20,7 @@ package org.apache.dolphinscheduler.api.service.impl;
import org.apache.dolphinscheduler.api.metrics.ApiServerMetrics;
import org.apache.dolphinscheduler.api.service.MetricsCleanUpService;
import org.apache.dolphinscheduler.common.model.Server;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.IWorkflowInstanceService;
import org.apache.dolphinscheduler.registry.api.RegistryClient;
import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
@@ -49,10 +49,9 @@ public class MetricsCleanUpServiceImpl implements MetricsCleanUpService {
private void cleanUpWorkflowMetrics(Server server, Long workflowDefinitionCode) {
try {
- IWorkflowInstanceService iWorkflowInstanceService =
- SingletonJdkDynamicRpcClientProxyFactory.getProxyClient(
- String.format("%s:%s", server.getHost(), server.getPort()), IWorkflowInstanceService.class);
- iWorkflowInstanceService.clearWorkflowMetrics(workflowDefinitionCode);
+ Clients.withService(IWorkflowInstanceService.class)
+ .withHost(server.getHost() + ":" + server.getPort())
+ .clearWorkflowMetrics(workflowDefinitionCode);
} catch (Exception e) {
log.error(
"Fail to clean up workflow related metrics on {} when deleting workflow definition {}, error message {}",
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java
index 49e42da561..610e9e5afa 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java
@@ -45,7 +45,7 @@ import org.apache.dolphinscheduler.dao.repository.DqExecuteResultDao;
import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.dao.utils.TaskCacheUtils;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.common.ILogService;
import org.apache.dolphinscheduler.extract.worker.IStreamingTaskInstanceOperator;
import org.apache.dolphinscheduler.extract.worker.ITaskInstanceOperator;
@@ -277,11 +277,10 @@ public class TaskInstanceServiceImpl extends BaseServiceImpl implements TaskInst
putMsg(result, Status.TASK_INSTANCE_NOT_FOUND);
return result;
}
- IStreamingTaskInstanceOperator streamingTaskInstanceOperator =
- SingletonJdkDynamicRpcClientProxyFactory
- .getProxyClient(taskInstance.getHost(), IStreamingTaskInstanceOperator.class);
- TaskInstanceTriggerSavepointResponse taskInstanceTriggerSavepointResponse =
- streamingTaskInstanceOperator.triggerSavepoint(new TaskInstanceTriggerSavepointRequest(taskInstanceId));
+ final TaskInstanceTriggerSavepointResponse taskInstanceTriggerSavepointResponse = Clients
+ .withService(IStreamingTaskInstanceOperator.class)
+ .withHost(taskInstance.getHost())
+ .triggerSavepoint(new TaskInstanceTriggerSavepointRequest(taskInstanceId));
log.info("StreamingTaskInstance trigger savepoint response: {}", taskInstanceTriggerSavepointResponse);
putMsg(result, Status.SUCCESS);
return result;
@@ -310,10 +309,10 @@ public class TaskInstanceServiceImpl extends BaseServiceImpl implements TaskInst
}
// todo: we only support streaming task for now
- ITaskInstanceOperator iTaskInstanceOperator = SingletonJdkDynamicRpcClientProxyFactory
- .getProxyClient(taskInstance.getHost(), ITaskInstanceOperator.class);
- TaskInstanceKillResponse taskInstanceKillResponse =
- iTaskInstanceOperator.killTask(new TaskInstanceKillRequest(taskInstanceId));
+ final TaskInstanceKillResponse taskInstanceKillResponse = Clients
+ .withService(ITaskInstanceOperator.class)
+ .withHost(taskInstance.getHost())
+ .killTask(new TaskInstanceKillRequest(taskInstanceId));
log.info("TaskInstance kill response: {}", taskInstanceKillResponse);
putMsg(result, Status.SUCCESS);
@@ -369,10 +368,10 @@ public class TaskInstanceServiceImpl extends BaseServiceImpl implements TaskInst
if (StringUtils.isNotBlank(taskInstance.getLogPath())) {
try {
// Remove task instance log failed will not affect the deletion of task instance
- ILogService iLogService =
- SingletonJdkDynamicRpcClientProxyFactory.getProxyClient(taskInstance.getHost(),
- ILogService.class);
- iLogService.removeTaskInstanceLog(taskInstance.getLogPath());
+ Clients
+ .withService(ILogService.class)
+ .withHost(taskInstance.getHost())
+ .removeTaskInstanceLog(taskInstance.getLogPath());
} catch (Exception ex) {
log.error("Remove task instance log error", ex);
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTO.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTO.java
index 4045c51a29..b363f97788 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTO.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTO.java
@@ -73,6 +73,4 @@ public class TriggerWorkflowDTO {
private Flag testFlag;
- private Long triggerCode;
-
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowRequestTransformer.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowRequestTransformer.java
index 9bf0a32418..2d33310673 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowRequestTransformer.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowRequestTransformer.java
@@ -21,7 +21,6 @@ import org.apache.dolphinscheduler.api.dto.workflow.WorkflowTriggerRequest;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.utils.WorkflowUtils;
import org.apache.dolphinscheduler.api.validator.ITransformer;
-import org.apache.dolphinscheduler.common.utils.CodeGenerateUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.repository.ProcessDefinitionDao;
import org.apache.dolphinscheduler.plugin.task.api.utils.PropertyUtils;
@@ -56,7 +55,6 @@ public class TriggerWorkflowRequestTransformer implements ITransformer<WorkflowT
PropertyUtils.startParamsTransformPropertyList(workflowTriggerRequest.getStartParamList()))
.dryRun(workflowTriggerRequest.getDryRun())
.testFlag(workflowTriggerRequest.getTestFlag())
- .triggerCode(CodeGenerateUtils.genCode())
.build();
ProcessDefinition workflowDefinition = processDefinitionDao
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/WorkflowInstanceExecuteFunctionControllerTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/WorkflowInstanceExecuteFunctionControllerTest.java
index d7b7fca5f2..7f3324f365 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/WorkflowInstanceExecuteFunctionControllerTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/WorkflowInstanceExecuteFunctionControllerTest.java
@@ -89,7 +89,7 @@ public class WorkflowInstanceExecuteFunctionControllerTest extends AbstractContr
final Integer version = 1;
final boolean allLevelDependent = false;
final JsonObject expectResponseContent = gson
- .fromJson("{\"code\":0,\"msg\":\"success\",\"data\":1,\"success\":true,\"failed\":false}",
+ .fromJson("{\"code\":0,\"msg\":\"success\",\"data\":[1],\"success\":true,\"failed\":false}",
JsonObject.class);
final ImmutableMap<String, Object> executeServiceResult =
@@ -123,7 +123,7 @@ public class WorkflowInstanceExecuteFunctionControllerTest extends AbstractContr
paramsMap.add("executionOrder", String.valueOf(executionOrder));
paramsMap.add("version", String.valueOf(version));
- when(executorService.triggerWorkflowDefinition(Mockito.any())).thenReturn(1L);
+ when(executorService.triggerWorkflowDefinition(Mockito.any())).thenReturn(1);
// When
final MvcResult mvcResult = mockMvc
@@ -163,7 +163,7 @@ public class WorkflowInstanceExecuteFunctionControllerTest extends AbstractContr
paramsMap.add("executionOrder", String.valueOf(executionOrder));
paramsMap.add("version", String.valueOf(version));
- when(executorService.triggerWorkflowDefinition(Mockito.any())).thenReturn(1L);
+ when(executorService.triggerWorkflowDefinition(Mockito.any())).thenReturn(1);
// When
final MvcResult mvcResult = mockMvc
@@ -203,7 +203,7 @@ public class WorkflowInstanceExecuteFunctionControllerTest extends AbstractContr
paramsMap.add("executionOrder", String.valueOf(executionOrder));
paramsMap.add("version", String.valueOf(version));
- when(executorService.triggerWorkflowDefinition(Mockito.any())).thenReturn(1L);
+ when(executorService.triggerWorkflowDefinition(Mockito.any())).thenReturn(1);
// When
final MvcResult mvcResult = mockMvc
@@ -229,7 +229,7 @@ public class WorkflowInstanceExecuteFunctionControllerTest extends AbstractContr
paramsMap.add("scheduleTime", scheduleTime);
paramsMap.add("version", String.valueOf(version));
- when(executorService.triggerWorkflowDefinition(Mockito.any())).thenReturn(1L);
+ when(executorService.triggerWorkflowDefinition(Mockito.any())).thenReturn(1);
final MvcResult mvcResult = mockMvc
.perform(post("/projects/{projectCode}/executors/start-process-instance", projectCode)
diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/CommandType.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/CommandType.java
index e54c5b1b4f..652c167ce9 100644
--- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/CommandType.java
+++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/CommandType.java
@@ -22,6 +22,7 @@ import java.util.Map;
import com.baomidou.mybatisplus.annotation.EnumValue;
+// todo: rename to WorkflowTriggerType
public enum CommandType {
/**
diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WorkflowExecutionStatus.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WorkflowExecutionStatus.java
index 7df8d451fc..caacc22deb 100644
--- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WorkflowExecutionStatus.java
+++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WorkflowExecutionStatus.java
@@ -26,6 +26,7 @@ import com.baomidou.mybatisplus.annotation.EnumValue;
public enum WorkflowExecutionStatus {
+ SUBMITTED_SUCCESS(0, "submitted"),
RUNNING_EXECUTION(1, "running"),
READY_PAUSE(2, "ready pause"),
PAUSE(3, "pause"),
@@ -45,6 +46,7 @@ public enum WorkflowExecutionStatus {
};
private static final int[] NOT_TERMINAL_STATUS = new int[]{
+ SUBMITTED_SUCCESS.getCode(),
RUNNING_EXECUTION.getCode(),
READY_PAUSE.getCode(),
READY_STOP.getCode(),
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java
index 788f9c14ae..79803907e0 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java
@@ -51,8 +51,11 @@ public class Command {
@TableField("process_definition_code")
private long processDefinitionCode;
- @TableField("executor_id")
- private int executorId;
+ @TableField("process_definition_version")
+ private int processDefinitionVersion;
+
+ @TableField("process_instance_id")
+ private int processInstanceId;
/**
* command parameter, format json
@@ -60,56 +63,66 @@ public class Command {
@TableField("command_param")
private String commandParam;
+ @TableField("process_instance_priority")
+ private Priority processInstancePriority;
+
+ @Deprecated
+ @TableField("executor_id")
+ private int executorId;
+
+ @Deprecated
@TableField("task_depend_type")
@Builder.Default
private TaskDependType taskDependType = TaskDependType.TASK_POST;
+ @Deprecated
@TableField("failure_strategy")
@Builder.Default
private FailureStrategy failureStrategy = FailureStrategy.CONTINUE;
+ @Deprecated
@TableField("warning_type")
private WarningType warningType;
+ @Deprecated
@TableField("warning_group_id")
private Integer warningGroupId;
+ @Deprecated
@TableField("schedule_time")
private Date scheduleTime;
+ @Deprecated
@TableField("start_time")
private Date startTime = new Date();
- @TableField("process_instance_priority")
- private Priority processInstancePriority;
-
+ @Deprecated
@TableField("update_time")
@Builder.Default
private Date updateTime = new Date();
+ @Deprecated
@TableField("worker_group")
private String workerGroup;
/**
* tenant code
*/
+ @Deprecated
private String tenantCode;
+ @Deprecated
@TableField("environment_code")
private Long environmentCode;
+ @Deprecated
@TableField("dry_run")
private int dryRun;
- @TableField("process_instance_id")
- private int processInstanceId;
-
- @TableField("process_definition_version")
- private int processDefinitionVersion;
-
/**
* test flag
*/
+ @Deprecated
@TableField("test_flag")
private int testFlag;
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TriggerRelation.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TriggerRelation.java
deleted file mode 100644
index 0e83ff06ef..0000000000
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TriggerRelation.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * 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.dolphinscheduler.dao.entity;
-
-import java.util.Date;
-
-import lombok.Data;
-
-import com.baomidou.mybatisplus.annotation.IdType;
-import com.baomidou.mybatisplus.annotation.TableId;
-import com.baomidou.mybatisplus.annotation.TableName;
-
-@Data
-@TableName("t_ds_trigger_relation")
-public class TriggerRelation {
-
- /**
- * id
- */
- @TableId(value = "id", type = IdType.AUTO)
- private Integer id;
-
- /**
- * trigger code
- */
- private long triggerCode;
-
- /**
- * triggerType
- */
- private int triggerType;
-
- /**
- * jobId
- */
- private Integer jobId;
-
- /**
- * create time
- */
- private Date createTime;
-
- /**
- * update time
- */
- private Date updateTime;
-
-}
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TriggerRelationMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TriggerRelationMapper.java
deleted file mode 100644
index 912ef28101..0000000000
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TriggerRelationMapper.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * 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.dolphinscheduler.dao.mapper;
-
-import org.apache.dolphinscheduler.dao.entity.TriggerRelation;
-
-import org.apache.ibatis.annotations.Param;
-
-import java.util.List;
-
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-
-/**
- * triggerRelation mapper interface
- */
-public interface TriggerRelationMapper extends BaseMapper<TriggerRelation> {
-
- /**
- * query by code and id
- * @param triggerType
- * @param jobId
- * @return
- */
- List<TriggerRelation> queryByTypeAndJobId(@Param("triggerType") Integer triggerType, @Param("jobId") int jobId);
-
- /**
- * query triggerRelation by code
- *
- * @param triggerCode triggerCode
- * @return triggerRelation
- */
- List<TriggerRelation> queryByTriggerRelationCode(@Param("triggerCode") Long triggerCode);
-
- /**
- * query triggerRelation by code
- *
- * @param triggerCode triggerCode
- * @return triggerRelation
- */
- List<TriggerRelation> queryByTriggerRelationCodeAndType(@Param("triggerCode") Long triggerCode,
- @Param("triggerType") Integer triggerType);
-
- /**
- * delete triggerRelation by code
- *
- * @param triggerCode triggerCode
- * @return int
- */
- int deleteByCode(@Param("triggerCode") Long triggerCode);
-
- /**
- * if exist update else insert
- *
- * @param triggerRelation
- */
- void upsert(@Param("triggerRelation") TriggerRelation triggerRelation);
-}
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopRequest.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/UserDao.java
similarity index 74%
copy from dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopRequest.java
copy to dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/UserDao.java
index f7ad1a38b9..5720b3aac4 100644
--- a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopRequest.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/UserDao.java
@@ -15,17 +15,9 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.extract.master.transportor;
+package org.apache.dolphinscheduler.dao.repository;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-@Data
-@AllArgsConstructor
-@NoArgsConstructor
-public class WorkflowInstanceStopRequest {
-
- private Integer workflowInstanceId;
+import org.apache.dolphinscheduler.dao.entity.User;
+public interface UserDao extends IDao<User> {
}
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopRequest.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/UserDaoImpl.java
similarity index 59%
copy from dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopRequest.java
copy to dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/UserDaoImpl.java
index f7ad1a38b9..638ad12d5b 100644
--- a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopRequest.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/UserDaoImpl.java
@@ -15,17 +15,22 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.extract.master.transportor;
+package org.apache.dolphinscheduler.dao.repository.impl;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
+import org.apache.dolphinscheduler.dao.entity.User;
+import org.apache.dolphinscheduler.dao.mapper.UserMapper;
+import org.apache.dolphinscheduler.dao.repository.BaseDao;
+import org.apache.dolphinscheduler.dao.repository.UserDao;
-@Data
-@AllArgsConstructor
-@NoArgsConstructor
-public class WorkflowInstanceStopRequest {
+import lombok.NonNull;
- private Integer workflowInstanceId;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public class UserDaoImpl extends BaseDao<User, UserMapper> implements UserDao {
+
+ public UserDaoImpl(@NonNull UserMapper userMapper) {
+ super(userMapper);
+ }
}
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TriggerRelationMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TriggerRelationMapper.xml
deleted file mode 100644
index 0cb90a768d..0000000000
--- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TriggerRelationMapper.xml
+++ /dev/null
@@ -1,61 +0,0 @@
-<?xml version="1.0" encoding="UTF-8" ?>
-<!--
- ~ 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.
- -->
-
-<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
-<mapper namespace="org.apache.dolphinscheduler.dao.mapper.TriggerRelationMapper">
- <sql id="baseSql">
- id, trigger_code, trigger_type, job_id, create_time, update_time
- </sql>
-
- <select id="queryByTypeAndJobId" resultType="org.apache.dolphinscheduler.dao.entity.TriggerRelation">
- select
- <include refid="baseSql"/>
- from t_ds_trigger_relation
- WHERE trigger_type = #{triggerType} and job_id = #{jobId}
- </select>
-
- <select id="queryByTriggerRelationCode" resultType="org.apache.dolphinscheduler.dao.entity.TriggerRelation">
- select
- <include refid="baseSql"/>
- from t_ds_trigger_relation
- WHERE trigger_code = #{triggerCode}
- </select>
-
- <select id="queryByTriggerRelationCodeAndType" resultType="org.apache.dolphinscheduler.dao.entity.TriggerRelation">
- select
- <include refid="baseSql"/>
- from t_ds_trigger_relation
- WHERE trigger_code = #{triggerCode} and trigger_type = #{triggerType}
- </select>
-
- <delete id="deleteByCode">
- delete from t_ds_trigger_relation where triggerCode = #{triggerCode}
- </delete>
-
- <insert id="upsert">
- INSERT INTO t_ds_trigger_relation (trigger_code, trigger_type, job_id, create_time, update_time) VALUES(
- #{triggerRelation.triggerCode},#{triggerRelation.triggerType},#{triggerRelation.jobId},#{triggerRelation.createTime},#{triggerRelation.updateTime})
- ON DUPLICATE KEY UPDATE update_time = #{triggerRelation.updateTime};
- </insert>
-
- <insert id="upsert" databaseId="PostgreSQL">
- INSERT INTO t_ds_trigger_relation (trigger_code, trigger_type, job_id, create_time, update_time) VALUES(
- #{triggerRelation.triggerCode},#{triggerRelation.triggerType},#{triggerRelation.jobId},#{triggerRelation.createTime},#{triggerRelation.updateTime})
- ON CONFLICT(trigger_type,job_id,trigger_code) DO UPDATE SET update_time = #{triggerRelation.updateTime};
- </insert>
-</mapper>
diff --git a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql
index cedbe1aa11..779614fd73 100644
--- a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql
+++ b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql
@@ -2128,22 +2128,6 @@ CREATE TABLE t_ds_fav_task
PRIMARY KEY (id)
);
---
--- Table structure for t_ds_trigger_relation
---
-DROP TABLE IF EXISTS `t_ds_trigger_relation`;
-CREATE TABLE t_ds_trigger_relation
-(
- id bigint(20) NOT NULL AUTO_INCREMENT,
- trigger_type int NOT NULL,
- job_id int NOT NULL,
- trigger_code bigint(20) NOT NULL,
- create_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
- update_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
- PRIMARY KEY (id),
- UNIQUE KEY t_ds_trigger_relation_UN(trigger_type,job_id,trigger_code)
-);
-
DROP TABLE IF EXISTS t_ds_relation_sub_workflow;
CREATE TABLE t_ds_relation_sub_workflow (
diff --git a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql
index b998919e27..770e592331 100644
--- a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql
+++ b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql
@@ -2107,22 +2107,6 @@ CREATE TABLE `t_ds_fav_task`
AUTO_INCREMENT = 1
DEFAULT CHARSET = utf8 COLLATE = utf8_bin;
--- ----------------------------
--- Table structure for t_ds_trigger_relation
--- ----------------------------
-DROP TABLE IF EXISTS `t_ds_trigger_relation`;
-CREATE TABLE `t_ds_trigger_relation` (
- `id` bigint(20) NOT NULL AUTO_INCREMENT,
- `trigger_type` int(11) NOT NULL DEFAULT '0' COMMENT '0 process 1 task',
- `trigger_code` bigint(20) NOT NULL,
- `job_id` bigint(20) NOT NULL,
- `create_time` datetime DEFAULT NULL,
- `update_time` datetime DEFAULT NULL,
- PRIMARY KEY (`id`),
- KEY `t_ds_trigger_relation_trigger_code_IDX` (`trigger_code`),
- UNIQUE KEY `t_ds_trigger_relation_UN` (`trigger_type`,`job_id`,`trigger_code`)
-) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COLLATE = utf8_bin;
-
DROP TABLE IF EXISTS `t_ds_relation_sub_workflow`;
CREATE TABLE `t_ds_relation_sub_workflow` (
diff --git a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql
index aaced9dfe6..b8046385f7 100644
--- a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql
+++ b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql
@@ -2098,21 +2098,6 @@ CREATE TABLE t_ds_fav_task
PRIMARY KEY (id)
);
--- ----------------------------
--- Table structure for t_ds_trigger_relation
--- ----------------------------
-DROP TABLE IF EXISTS t_ds_trigger_relation;
-CREATE TABLE t_ds_trigger_relation (
- id serial NOT NULL,
- trigger_type int NOT NULL,
- trigger_code bigint NOT NULL,
- job_id bigint NOT NULL,
- create_time timestamp DEFAULT NULL,
- update_time timestamp DEFAULT NULL,
- PRIMARY KEY (id),
- CONSTRAINT t_ds_trigger_relation_unique UNIQUE (trigger_type,job_id,trigger_code)
-);
-
DROP TABLE IF EXISTS t_ds_relation_sub_workflow;
CREATE TABLE t_ds_relation_sub_workflow (
id serial NOT NULL,
diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.3.0_schema/mysql/dolphinscheduler_ddl.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.3.0_schema/mysql/dolphinscheduler_ddl.sql
index 3c3f27fecd..cb0bdaf1b1 100644
--- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.3.0_schema/mysql/dolphinscheduler_ddl.sql
+++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.3.0_schema/mysql/dolphinscheduler_ddl.sql
@@ -125,3 +125,5 @@ d//
delimiter ;
CALL drop_column_t_ds_alert_plugin_instance;
DROP PROCEDURE drop_column_t_ds_alert_plugin_instance;
+
+DROP TABLE IF EXISTS `t_ds_trigger_relation`;
diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.3.0_schema/postgresql/dolphinscheduler_ddl.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.3.0_schema/postgresql/dolphinscheduler_ddl.sql
index 56b410756d..8fd0f00746 100644
--- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.3.0_schema/postgresql/dolphinscheduler_ddl.sql
+++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.3.0_schema/postgresql/dolphinscheduler_ddl.sql
@@ -123,3 +123,5 @@ d//
select drop_column_t_ds_alert_plugin_instance();
DROP FUNCTION IF EXISTS drop_column_t_ds_alert_plugin_instance();
+
+DROP TABLE IF EXISTS t_ds_trigger_relation;
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TriggerRelationMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TriggerRelationMapperTest.java
deleted file mode 100644
index 7e31b64a23..0000000000
--- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TriggerRelationMapperTest.java
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * 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.dolphinscheduler.dao.mapper;
-
-import static com.google.common.truth.Truth.assertThat;
-
-import org.apache.dolphinscheduler.common.enums.ApiTriggerType;
-import org.apache.dolphinscheduler.common.utils.DateUtils;
-import org.apache.dolphinscheduler.dao.BaseDaoTest;
-import org.apache.dolphinscheduler.dao.entity.TriggerRelation;
-
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-import org.springframework.beans.factory.annotation.Autowired;
-
-/**
- * trigger mapper test
- */
-public class TriggerRelationMapperTest extends BaseDaoTest {
-
- @Autowired
- TriggerRelationMapper triggerRelationMapper;
-
- /**
- * test insert
- *
- * @return
- */
- @Test
- public void testInsert() {
- TriggerRelation expectedObj = createTriggerRelation();
- Assertions.assertTrue(expectedObj.getId() > 0);
- }
-
- /**
- * test select by id
- *
- * @return
- */
- @Test
- public void testSelectById() {
- TriggerRelation expectRelation = createTriggerRelation();
- TriggerRelation actualRelation = triggerRelationMapper.selectById(expectRelation.getId());
- Assertions.assertEquals(expectRelation, actualRelation);
- }
-
- /**
- * test select by type and job id
- *
- * @return
- */
- @Test
- public void testQueryByTypeAndJobId() {
- TriggerRelation expectRelation = createTriggerRelation();
- assertThat(
- triggerRelationMapper.queryByTypeAndJobId(expectRelation.getTriggerType(), expectRelation.getJobId()))
- .containsExactly(expectRelation);
- }
-
- /**
- * test select by trigger code
- *
- * @return
- */
- @Test
- public void testQueryByTriggerRelationCode() {
- TriggerRelation expectRelation = createTriggerRelation();
- assertThat(triggerRelationMapper.queryByTriggerRelationCode(expectRelation.getTriggerCode()))
- .containsExactly(expectRelation);
- }
-
- /**
- * test select by type and trigger code
- *
- * @return
- */
- @Test
- public void testQueryByTriggerRelationCodeAndType() {
- TriggerRelation expectRelation = createTriggerRelation();
- assertThat(triggerRelationMapper.queryByTriggerRelationCodeAndType(expectRelation.getTriggerCode(),
- expectRelation.getTriggerType())).containsExactly(expectRelation);
- }
-
- @Test
- public void testUpsert() {
- TriggerRelation expectRelation = createTriggerRelation();
- triggerRelationMapper.upsert(expectRelation);
- assertThat(triggerRelationMapper.selectById(expectRelation.getId())).isEqualTo(expectRelation);
- }
-
- /**
- * test delete
- */
- @Test
- public void testDelete() {
- TriggerRelation expectRelation = createTriggerRelation();
- triggerRelationMapper.deleteById(expectRelation.getId());
- assertThat(triggerRelationMapper.selectById(expectRelation.getId())).isNull();
- }
-
- /**
- * create TriggerRelation and insert
- *
- * @return TriggerRelation
- * @throws Exception
- */
- private TriggerRelation createTriggerRelation() {
- TriggerRelation triggerRelation = new TriggerRelation();
- triggerRelation.setTriggerCode(4567890);
- triggerRelation.setTriggerType(ApiTriggerType.COMMAND.getCode());
- triggerRelation.setJobId(99);
- triggerRelation.setCreateTime(DateUtils.getCurrentDate());
- triggerRelation.setUpdateTime(DateUtils.getCurrentDate());
-
- triggerRelationMapper.insert(triggerRelation);
- return triggerRelation;
- }
-
-}
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/SingletonJdkDynamicRpcClientProxyFactory.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/Clients.java
similarity index 71%
rename from dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/SingletonJdkDynamicRpcClientProxyFactory.java
rename to dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/Clients.java
index fc48d90c65..c4741e0a3b 100644
--- a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/SingletonJdkDynamicRpcClientProxyFactory.java
+++ b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/Clients.java
@@ -22,15 +22,21 @@ import org.apache.dolphinscheduler.extract.base.config.NettyClientConfig;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
+/**
+ * The factory class for creating a dynamic proxy client.
+ * <pre>
+ * final IService proxyClient = Clients
+ * .withService(IService.class)
+ * .withHost(serverAddress);
+ * </pre>
+ */
@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public class SingletonJdkDynamicRpcClientProxyFactory {
-
- private static final JdkDynamicRpcClientProxyFactory INSTANCE = new JdkDynamicRpcClientProxyFactory(
- NettyRemotingClientFactory.buildNettyRemotingClient(new NettyClientConfig()));
+public class Clients {
- public static <T> T getProxyClient(String serverAddress, Class<T> clazz) {
- return INSTANCE.getProxyClient(serverAddress, clazz);
- }
+ private static final JdkDynamicRpcClientProxyFactory jdkDynamicRpcClientProxyFactory =
+ new JdkDynamicRpcClientProxyFactory(
+ NettyRemotingClientFactory.buildNettyRemotingClient(
+ new NettyClientConfig()));
public static <T> JdkDynamicRpcClientProxyBuilder<T> withService(Class<T> serviceClazz) {
return new JdkDynamicRpcClientProxyBuilder<>(serviceClazz);
@@ -45,7 +51,7 @@ public class SingletonJdkDynamicRpcClientProxyFactory {
}
public T withHost(String serviceHost) {
- return getProxyClient(serviceHost, serviceClazz);
+ return jdkDynamicRpcClientProxyFactory.getProxyClient(serviceHost, serviceClazz);
}
}
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/test/java/org/apache/dolphinscheduler/extract/base/client/SingletonJdkDynamicRpcClientProxyFactoryTest.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/test/java/org/apache/dolphinscheduler/extract/base/client/ClientsTest.java
similarity index 88%
rename from dolphinscheduler-extract/dolphinscheduler-extract-base/src/test/java/org/apache/dolphinscheduler/extract/base/client/SingletonJdkDynamicRpcClientProxyFactoryTest.java
rename to dolphinscheduler-extract/dolphinscheduler-extract-base/src/test/java/org/apache/dolphinscheduler/extract/base/client/ClientsTest.java
index 92ed49934c..72a3889b13 100644
--- a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/test/java/org/apache/dolphinscheduler/extract/base/client/SingletonJdkDynamicRpcClientProxyFactoryTest.java
+++ b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/test/java/org/apache/dolphinscheduler/extract/base/client/ClientsTest.java
@@ -34,7 +34,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-public class SingletonJdkDynamicRpcClientProxyFactoryTest {
+public class ClientsTest {
private SpringServerMethodInvokerDiscovery springServerMethodInvokerDiscovery;
@@ -55,15 +55,17 @@ public class SingletonJdkDynamicRpcClientProxyFactoryTest {
@Test
public void getProxyClient() {
- IService proxyClient =
- SingletonJdkDynamicRpcClientProxyFactory.getProxyClient(serverAddress, IService.class);
+ IService proxyClient = Clients
+ .withService(IService.class)
+ .withHost(serverAddress);
Assertions.assertNotNull(proxyClient);
}
@Test
public void testPing() {
- IService proxyClient =
- SingletonJdkDynamicRpcClientProxyFactory.getProxyClient(serverAddress, IService.class);
+ IService proxyClient = Clients
+ .withService(IService.class)
+ .withHost(serverAddress);
assertEquals("pong", proxyClient.ping("ping"));
MethodInvocationException methodInvocationException =
@@ -73,8 +75,9 @@ public class SingletonJdkDynamicRpcClientProxyFactoryTest {
@Test
public void testVoid() {
- IService proxyClient =
- SingletonJdkDynamicRpcClientProxyFactory.getProxyClient(serverAddress, IService.class);
+ IService proxyClient = Clients
+ .withService(IService.class)
+ .withHost(serverAddress);
assertDoesNotThrow(proxyClient::voidMethod);
}
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/IWorkflowControlClient.java b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/IWorkflowControlClient.java
new file mode 100644
index 0000000000..be1fc1f85e
--- /dev/null
+++ b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/IWorkflowControlClient.java
@@ -0,0 +1,53 @@
+/*
+ * 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.dolphinscheduler.extract.master;
+
+import org.apache.dolphinscheduler.extract.base.RpcMethod;
+import org.apache.dolphinscheduler.extract.base.RpcService;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowBackfillTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowBackfillTriggerResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstancePauseRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstancePauseResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowManualTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowManualTriggerResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowScheduleTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowScheduleTriggerResponse;
+
+/**
+ * Workflow instance controller used to do control operation for workflow instance.
+ */
+@RpcService
+public interface IWorkflowControlClient {
+
+ @RpcMethod
+ WorkflowManualTriggerResponse manualTriggerWorkflow(final WorkflowManualTriggerRequest workflowManualTriggerRequest);
+
+ @RpcMethod
+ WorkflowBackfillTriggerResponse backfillTriggerWorkflow(final WorkflowBackfillTriggerRequest workflowBackfillTriggerRequest);
+
+ @RpcMethod
+ WorkflowScheduleTriggerResponse scheduleTriggerWorkflow(final WorkflowScheduleTriggerRequest workflowScheduleTriggerRequest);
+
+ @RpcMethod
+ WorkflowInstancePauseResponse pauseWorkflowInstance(final WorkflowInstancePauseRequest workflowInstancePauseRequest);
+
+ @RpcMethod
+ WorkflowInstanceStopResponse stopWorkflowInstance(final WorkflowInstanceStopRequest workflowInstanceStopRequest);
+}
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/IWorkflowInstanceController.java b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/IWorkflowInstanceController.java
deleted file mode 100644
index 043efaa83c..0000000000
--- a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/IWorkflowInstanceController.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.dolphinscheduler.extract.master;
-
-import org.apache.dolphinscheduler.extract.base.RpcMethod;
-import org.apache.dolphinscheduler.extract.base.RpcService;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstancePauseRequest;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstancePauseResponse;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStopRequest;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStopResponse;
-
-/**
- * Workflow instance controller used to do control operation for workflow instance.
- */
-@RpcService
-public interface IWorkflowInstanceController {
-
- @RpcMethod
- WorkflowInstancePauseResponse pauseWorkflowInstance(WorkflowInstancePauseRequest workflowInstancePauseRequest);
-
- @RpcMethod
- WorkflowInstanceStopResponse stopWorkflowInstance(WorkflowInstanceStopRequest workflowInstanceStopRequest);
-
-}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTO.java b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowBackfillTriggerRequest.java
similarity index 66%
copy from dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTO.java
copy to dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowBackfillTriggerRequest.java
index 4045c51a29..d30c166bb9 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTO.java
+++ b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowBackfillTriggerRequest.java
@@ -15,19 +15,17 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.api.validator.workflow;
+package org.apache.dolphinscheduler.extract.master.transportor.workflow;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.FailureStrategy;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.Priority;
-import org.apache.dolphinscheduler.common.enums.RunMode;
import org.apache.dolphinscheduler.common.enums.TaskDependType;
import org.apache.dolphinscheduler.common.enums.WarningType;
-import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
-import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import java.util.ArrayList;
import java.util.List;
import lombok.AllArgsConstructor;
@@ -39,27 +37,33 @@ import lombok.NoArgsConstructor;
@Builder
@NoArgsConstructor
@AllArgsConstructor
-public class TriggerWorkflowDTO {
+public class WorkflowBackfillTriggerRequest {
- private User loginUser;
+ private Integer userId;
- private ProcessDefinition workflowDefinition;
+ private List<String> backfillTimeList;
+
+ private Long workflowCode;
+
+ private Integer workflowVersion;
private List<Long> startNodes;
- private FailureStrategy failureStrategy;
+ @Builder.Default
+ private FailureStrategy failureStrategy = FailureStrategy.CONTINUE;
- private TaskDependType taskDependType;
+ @Builder.Default
+ private TaskDependType taskDependType = TaskDependType.TASK_POST;
private CommandType execType;
- private WarningType warningType;
+ @Builder.Default
+ private WarningType warningType = WarningType.NONE;
private Integer warningGroupId;
- private RunMode runMode;
-
- private Priority workflowInstancePriority;
+ @Builder.Default
+ private Priority workflowInstancePriority = Priority.MEDIUM;
private String workerGroup;
@@ -67,12 +71,13 @@ public class TriggerWorkflowDTO {
private Long environmentCode;
- private List<Property> startParamList;
-
- private Flag dryRun;
+ @Builder.Default
+ private List<Property> startParamList = new ArrayList<>();
- private Flag testFlag;
+ @Builder.Default
+ private Flag dryRun = Flag.NO;
- private Long triggerCode;
+ @Builder.Default
+ private Flag testFlag = Flag.NO;
}
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopResponse.java b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowBackfillTriggerResponse.java
similarity index 60%
copy from dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopResponse.java
copy to dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowBackfillTriggerResponse.java
index 22ae5b6891..5ca09228e1 100644
--- a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopResponse.java
+++ b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowBackfillTriggerResponse.java
@@ -15,25 +15,37 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.extract.master.transportor;
+package org.apache.dolphinscheduler.extract.master.transportor.workflow;
import lombok.AllArgsConstructor;
+import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
-@AllArgsConstructor
+@Builder
@NoArgsConstructor
-public class WorkflowInstanceStopResponse {
+@AllArgsConstructor
+public class WorkflowBackfillTriggerResponse {
private boolean success;
+
private String message;
- public static WorkflowInstanceStopResponse success() {
- return new WorkflowInstanceStopResponse(true, null);
+ private Integer workflowInstanceId;
+
+ public static WorkflowBackfillTriggerResponse fail(String message) {
+ return WorkflowBackfillTriggerResponse.builder()
+ .success(false)
+ .message(message)
+ .build();
}
- public static WorkflowInstanceStopResponse fail(String message) {
- return new WorkflowInstanceStopResponse(false, message);
+ public static WorkflowBackfillTriggerResponse success(Integer workflowInstanceId) {
+ return WorkflowBackfillTriggerResponse.builder()
+ .success(true)
+ .workflowInstanceId(workflowInstanceId)
+ .build();
}
+
}
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstancePauseRequest.java b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowInstancePauseRequest.java
similarity index 93%
rename from dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstancePauseRequest.java
rename to dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowInstancePauseRequest.java
index 1ddd34c113..e5b10829e9 100644
--- a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstancePauseRequest.java
+++ b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowInstancePauseRequest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.extract.master.transportor;
+package org.apache.dolphinscheduler.extract.master.transportor.workflow;
import lombok.AllArgsConstructor;
import lombok.Data;
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstancePauseResponse.java b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowInstancePauseResponse.java
similarity index 94%
copy from dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstancePauseResponse.java
copy to dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowInstancePauseResponse.java
index 7b2eb86802..958fd97bc4 100644
--- a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstancePauseResponse.java
+++ b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowInstancePauseResponse.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.extract.master.transportor;
+package org.apache.dolphinscheduler.extract.master.transportor.workflow;
import lombok.AllArgsConstructor;
import lombok.Data;
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopRequest.java b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowInstanceStopRequest.java
similarity index 93%
copy from dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopRequest.java
copy to dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowInstanceStopRequest.java
index f7ad1a38b9..a30e657f9d 100644
--- a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopRequest.java
+++ b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowInstanceStopRequest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.extract.master.transportor;
+package org.apache.dolphinscheduler.extract.master.transportor.workflow;
import lombok.AllArgsConstructor;
import lombok.Data;
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopResponse.java b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowInstanceStopResponse.java
similarity index 94%
copy from dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopResponse.java
copy to dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowInstanceStopResponse.java
index 22ae5b6891..010b3732e7 100644
--- a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopResponse.java
+++ b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowInstanceStopResponse.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.extract.master.transportor;
+package org.apache.dolphinscheduler.extract.master.transportor.workflow;
import lombok.AllArgsConstructor;
import lombok.Data;
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTO.java b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowManualTriggerRequest.java
similarity index 65%
copy from dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTO.java
copy to dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowManualTriggerRequest.java
index 4045c51a29..83848009c3 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTO.java
+++ b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowManualTriggerRequest.java
@@ -15,19 +15,16 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.api.validator.workflow;
+package org.apache.dolphinscheduler.extract.master.transportor.workflow;
-import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.FailureStrategy;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.Priority;
-import org.apache.dolphinscheduler.common.enums.RunMode;
import org.apache.dolphinscheduler.common.enums.TaskDependType;
import org.apache.dolphinscheduler.common.enums.WarningType;
-import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
-import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import java.util.ArrayList;
import java.util.List;
import lombok.AllArgsConstructor;
@@ -39,27 +36,29 @@ import lombok.NoArgsConstructor;
@Builder
@NoArgsConstructor
@AllArgsConstructor
-public class TriggerWorkflowDTO {
+public class WorkflowManualTriggerRequest {
- private User loginUser;
+ private Integer userId;
- private ProcessDefinition workflowDefinition;
+ private Long workflowDefinitionCode;
- private List<Long> startNodes;
+ private Integer workflowDefinitionVersion;
- private FailureStrategy failureStrategy;
+ private List<Long> startNodes;
- private TaskDependType taskDependType;
+ @Builder.Default
+ private FailureStrategy failureStrategy = FailureStrategy.CONTINUE;
- private CommandType execType;
+ @Builder.Default
+ private TaskDependType taskDependType = TaskDependType.TASK_POST;
- private WarningType warningType;
+ @Builder.Default
+ private WarningType warningType = WarningType.NONE;
private Integer warningGroupId;
- private RunMode runMode;
-
- private Priority workflowInstancePriority;
+ @Builder.Default
+ private Priority workflowInstancePriority = Priority.MEDIUM;
private String workerGroup;
@@ -67,12 +66,12 @@ public class TriggerWorkflowDTO {
private Long environmentCode;
- private List<Property> startParamList;
-
- private Flag dryRun;
-
- private Flag testFlag;
+ @Builder.Default
+ private List<Property> startParamList = new ArrayList<>();
- private Long triggerCode;
+ @Builder.Default
+ private Flag dryRun = Flag.NO;
+ @Builder.Default
+ private Flag testFlag = Flag.NO;
}
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstancePauseResponse.java b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowManualTriggerResponse.java
similarity index 60%
rename from dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstancePauseResponse.java
rename to dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowManualTriggerResponse.java
index 7b2eb86802..4cf4e1669a 100644
--- a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstancePauseResponse.java
+++ b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowManualTriggerResponse.java
@@ -15,25 +15,36 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.extract.master.transportor;
+package org.apache.dolphinscheduler.extract.master.transportor.workflow;
import lombok.AllArgsConstructor;
+import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
-@AllArgsConstructor
+@Builder
@NoArgsConstructor
-public class WorkflowInstancePauseResponse {
+@AllArgsConstructor
+public class WorkflowManualTriggerResponse {
private boolean success;
+
private String message;
- public static WorkflowInstancePauseResponse success() {
- return new WorkflowInstancePauseResponse(true, null);
+ private Integer workflowInstanceId;
+
+ public static WorkflowManualTriggerResponse fail(String message) {
+ return WorkflowManualTriggerResponse.builder()
+ .success(false)
+ .message(message)
+ .build();
}
- public static WorkflowInstancePauseResponse fail(String message) {
- return new WorkflowInstancePauseResponse(false, message);
+ public static WorkflowManualTriggerResponse success(Integer workflowInstanceId) {
+ return WorkflowManualTriggerResponse.builder()
+ .success(true)
+ .workflowInstanceId(workflowInstanceId)
+ .build();
}
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTO.java b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowScheduleTriggerRequest.java
similarity index 61%
copy from dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTO.java
copy to dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowScheduleTriggerRequest.java
index 4045c51a29..9c49ad7d59 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTO.java
+++ b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowScheduleTriggerRequest.java
@@ -15,20 +15,15 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.api.validator.workflow;
+package org.apache.dolphinscheduler.extract.master.transportor.workflow;
-import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.FailureStrategy;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.Priority;
-import org.apache.dolphinscheduler.common.enums.RunMode;
import org.apache.dolphinscheduler.common.enums.TaskDependType;
import org.apache.dolphinscheduler.common.enums.WarningType;
-import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
-import org.apache.dolphinscheduler.dao.entity.User;
-import org.apache.dolphinscheduler.plugin.task.api.model.Property;
-import java.util.List;
+import java.util.Date;
import lombok.AllArgsConstructor;
import lombok.Builder;
@@ -39,27 +34,31 @@ import lombok.NoArgsConstructor;
@Builder
@NoArgsConstructor
@AllArgsConstructor
-public class TriggerWorkflowDTO {
+public class WorkflowScheduleTriggerRequest {
- private User loginUser;
+ private Integer userId;
- private ProcessDefinition workflowDefinition;
+ private Date scheduleTIme;
- private List<Long> startNodes;
+ private String timezoneId;
- private FailureStrategy failureStrategy;
+ private Long workflowCode;
- private TaskDependType taskDependType;
+ private Integer workflowVersion;
- private CommandType execType;
+ @Builder.Default
+ private FailureStrategy failureStrategy = FailureStrategy.CONTINUE;
- private WarningType warningType;
+ @Builder.Default
+ private TaskDependType taskDependType = TaskDependType.TASK_POST;
- private Integer warningGroupId;
+ @Builder.Default
+ private WarningType warningType = WarningType.NONE;
- private RunMode runMode;
+ private Integer warningGroupId;
- private Priority workflowInstancePriority;
+ @Builder.Default
+ private Priority workflowInstancePriority = Priority.MEDIUM;
private String workerGroup;
@@ -67,12 +66,10 @@ public class TriggerWorkflowDTO {
private Long environmentCode;
- private List<Property> startParamList;
-
- private Flag dryRun;
-
- private Flag testFlag;
+ @Builder.Default
+ private Flag dryRun = Flag.NO;
- private Long triggerCode;
+ @Builder.Default
+ private Flag testFlag = Flag.NO;
}
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopResponse.java b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowScheduleTriggerResponse.java
similarity index 60%
rename from dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopResponse.java
rename to dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowScheduleTriggerResponse.java
index 22ae5b6891..1ca855f6d2 100644
--- a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopResponse.java
+++ b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/workflow/WorkflowScheduleTriggerResponse.java
@@ -15,25 +15,37 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.extract.master.transportor;
+package org.apache.dolphinscheduler.extract.master.transportor.workflow;
import lombok.AllArgsConstructor;
+import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
-@AllArgsConstructor
+@Builder
@NoArgsConstructor
-public class WorkflowInstanceStopResponse {
+@AllArgsConstructor
+public class WorkflowScheduleTriggerResponse {
private boolean success;
+
private String message;
- public static WorkflowInstanceStopResponse success() {
- return new WorkflowInstanceStopResponse(true, null);
+ private Integer workflowInstanceId;
+
+ public static WorkflowScheduleTriggerResponse fail(String message) {
+ return WorkflowScheduleTriggerResponse.builder()
+ .success(false)
+ .message(message)
+ .build();
}
- public static WorkflowInstanceStopResponse fail(String message) {
- return new WorkflowInstanceStopResponse(false, message);
+ public static WorkflowScheduleTriggerResponse success(Integer workflowInstanceId) {
+ return WorkflowScheduleTriggerResponse.builder()
+ .success(true)
+ .workflowInstanceId(workflowInstanceId)
+ .build();
}
+
}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/TaskGroupCoordinator.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/TaskGroupCoordinator.java
index dc1e41f797..98969072e6 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/TaskGroupCoordinator.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/TaskGroupCoordinator.java
@@ -32,7 +32,7 @@ import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
import org.apache.dolphinscheduler.dao.repository.TaskGroupDao;
import org.apache.dolphinscheduler.dao.repository.TaskGroupQueueDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskInstanceController;
import org.apache.dolphinscheduler.extract.master.transportor.TaskGroupSlotAcquireSuccessNotifyRequest;
import org.apache.dolphinscheduler.extract.master.transportor.TaskGroupSlotAcquireSuccessNotifyResponse;
@@ -471,7 +471,7 @@ public class TaskGroupCoordinator extends BaseDaemonThread {
.build();
TaskGroupSlotAcquireSuccessNotifyResponse taskGroupSlotAcquireSuccessNotifyResponse =
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskInstanceController.class)
.withHost(processInstance.getHost())
.notifyTaskGroupSlotAcquireSuccess(taskGroupSlotAcquireSuccessNotifyRequest);
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/CommandEngine.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/CommandEngine.java
index 694143e9b5..f5db7e3235 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/CommandEngine.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/CommandEngine.java
@@ -33,6 +33,7 @@ import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.config.MasterServerLoadProtection;
import org.apache.dolphinscheduler.server.master.engine.IWorkflowRepository;
import org.apache.dolphinscheduler.server.master.engine.WorkflowEventBusCoordinator;
+import org.apache.dolphinscheduler.server.master.engine.exceptions.CommandDuplicateHandleException;
import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowStartLifecycleEvent;
import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.WorkflowExecutionRunnableFactory;
@@ -177,6 +178,12 @@ public class CommandEngine extends BaseDaemonThread implements AutoCloseable {
}
private Void bootstrapError(Command command, Throwable throwable) {
+ if (throwable instanceof CommandDuplicateHandleException) {
+ log.warn("Handle command failed, the command: {} has been handled by other master",
+ command,
+ throwable);
+ return null;
+ }
log.error("Failed bootstrap command {} ", JSONUtils.toPrettyJsonString(command), throwable);
commandService.moveToErrorCommand(command, ExceptionUtils.getStackTrace(throwable));
return null;
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/AbstractCommandHandler.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/AbstractCommandHandler.java
index db61cd8782..a596a607ff 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/AbstractCommandHandler.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/AbstractCommandHandler.java
@@ -40,7 +40,6 @@ import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteContext.W
import org.apache.commons.collections4.CollectionUtils;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
@@ -67,9 +66,7 @@ public abstract class AbstractCommandHandler implements ICommandHandler {
@Override
public WorkflowExecutionRunnable handleCommand(final Command command) {
- final WorkflowExecuteContextBuilder workflowExecuteContextBuilder = WorkflowExecuteContext
- .builder()
- .withWorkflowInstanceLifecycleListeners(workflowLifecycleListeners)
+ final WorkflowExecuteContextBuilder workflowExecuteContextBuilder = WorkflowExecuteContext.builder()
.withCommand(command);
assembleWorkflowDefinition(workflowExecuteContextBuilder);
@@ -94,7 +91,8 @@ public abstract class AbstractCommandHandler implements ICommandHandler {
protected void assembleWorkflowInstanceLifecycleListeners(
final WorkflowExecuteContextBuilder workflowExecuteContextBuilder) {
- workflowExecuteContextBuilder.setWorkflowInstanceLifecycleListeners(new ArrayList<>());
+ workflowExecuteContextBuilder.setWorkflowInstanceLifecycleListeners(
+ workflowExecuteContextBuilder.getWorkflowInstanceLifecycleListeners());
}
protected void assembleWorkflowDefinition(
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/BackfillWorkflowCommandHandler.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/BackfillWorkflowCommandHandler.java
index ef33cbfa1b..b8870c1298 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/BackfillWorkflowCommandHandler.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/BackfillWorkflowCommandHandler.java
@@ -18,14 +18,8 @@
package org.apache.dolphinscheduler.server.master.engine.command.handler;
import org.apache.dolphinscheduler.common.enums.CommandType;
-import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
-import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
-import org.apache.dolphinscheduler.server.master.config.MasterConfig;
-import org.apache.dolphinscheduler.server.master.engine.workflow.listener.WorkflowSuccessLifecycleListener;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;
/**
@@ -35,21 +29,6 @@ import org.springframework.stereotype.Component;
@Component
public class BackfillWorkflowCommandHandler extends RunWorkflowCommandHandler {
- @Autowired
- private ProcessInstanceDao processInstanceDao;
-
- @Autowired
- private TaskInstanceDao taskInstanceDao;
-
- @Autowired
- private MasterConfig masterConfig;
-
- @Autowired
- private ApplicationContext applicationContext;
-
- @Autowired
- private WorkflowSuccessLifecycleListener workflowSuccessLifecycleListener;
-
@Override
public CommandType commandType() {
return CommandType.COMPLEMENT_DATA;
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/ReRunWorkflowCommandHandler.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/ReRunWorkflowCommandHandler.java
index 6717bd1e2c..ddb00133e6 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/ReRunWorkflowCommandHandler.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/ReRunWorkflowCommandHandler.java
@@ -24,6 +24,7 @@ import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
+import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteContext.WorkflowExecuteContextBuilder;
import java.util.Date;
@@ -48,6 +49,9 @@ public class ReRunWorkflowCommandHandler extends RunWorkflowCommandHandler {
@Autowired
private ApplicationContext applicationContext;
+ @Autowired
+ private MasterConfig masterConfig;
+
/**
* Generate the repeat running workflow instance.
* <p> Will use the origin workflow instance, but will update the following fields. Need to note we cannot not
@@ -71,6 +75,7 @@ public class ReRunWorkflowCommandHandler extends RunWorkflowCommandHandler {
workflowInstance.setStateWithDesc(WorkflowExecutionStatus.RUNNING_EXECUTION, command.getCommandType().name());
workflowInstance.setCommandType(command.getCommandType());
workflowInstance.setRestartTime(new Date());
+ workflowInstance.setHost(masterConfig.getMasterAddress());
workflowInstance.setEndTime(null);
workflowInstance.setRunTimes(workflowInstance.getRunTimes() + 1);
workflowInstanceDao.updateById(workflowInstance);
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/RunWorkflowCommandHandler.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/RunWorkflowCommandHandler.java
index a7074a7898..98ae824542 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/RunWorkflowCommandHandler.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/RunWorkflowCommandHandler.java
@@ -18,18 +18,13 @@
package org.apache.dolphinscheduler.server.master.engine.command.handler;
import org.apache.dolphinscheduler.common.enums.CommandType;
-import org.apache.dolphinscheduler.common.enums.Flag;
-import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
-import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.Command;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
-import org.apache.dolphinscheduler.dao.utils.EnvironmentUtils;
-import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
import org.apache.dolphinscheduler.extract.master.command.ICommandParam;
import org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
@@ -43,9 +38,7 @@ import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteContext.W
import org.apache.dolphinscheduler.service.expand.CuringParamsService;
import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.ObjectUtils;
-import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -86,45 +79,15 @@ public class RunWorkflowCommandHandler extends AbstractCommandHandler {
protected void assembleWorkflowInstance(final WorkflowExecuteContextBuilder workflowExecuteContextBuilder) {
final ProcessDefinition workflowDefinition = workflowExecuteContextBuilder.getWorkflowDefinition();
final Command command = workflowExecuteContextBuilder.getCommand();
+ final ProcessInstance workflowInstance = processInstanceDao.queryById(command.getProcessInstanceId());
- final ProcessInstance processInstance = new ProcessInstance();
- processInstance.setProcessDefinitionCode(workflowDefinition.getCode());
- processInstance.setProcessDefinitionVersion(workflowDefinition.getVersion());
- processInstance.setProjectCode(workflowDefinition.getProjectCode());
- processInstance.setStateWithDesc(WorkflowExecutionStatus.RUNNING_EXECUTION, command.getCommandType().name());
- processInstance.setRecovery(Flag.NO);
- processInstance.setStartTime(new Date());
- processInstance.setRestartTime(processInstance.getStartTime());
- processInstance.setEndTime(null);
- processInstance.setRunTimes(1);
- processInstance.setName(String.join("-", workflowDefinition.getName(), DateUtils.getCurrentTimeStamp()));
- processInstance.setHost(masterConfig.getMasterAddress());
- processInstance.setCommandType(command.getCommandType());
- processInstance.setCommandParam(command.getCommandParam());
- processInstance.setTaskDependType(command.getTaskDependType());
- processInstance.setFailureStrategy(command.getFailureStrategy());
- processInstance.setWarningType(ObjectUtils.defaultIfNull(command.getWarningType(), WarningType.NONE));
- processInstance.setWarningGroupId(command.getWarningGroupId());
- processInstance.setScheduleTime(command.getScheduleTime());
- // todo: merge the global params or add startup params
- // or can we merge this after
-
- processInstance.setGlobalParams(mergeCommandParamsWithWorkflowParams(command, workflowDefinition));
- processInstance.setExecutorId(command.getExecutorId());
- processInstance.setExecutorName(null);
- processInstance.setTenantCode(command.getTenantCode());
- processInstance.setIsSubProcess(Flag.NO);
- processInstance.addHistoryCmd(command.getCommandType());
- processInstance.setProcessInstancePriority(command.getProcessInstancePriority());
- processInstance.setWorkerGroup(WorkerGroupUtils.getWorkerGroupOrDefault(command.getWorkerGroup()));
- processInstance.setEnvironmentCode(EnvironmentUtils.getEnvironmentCodeOrDefault(command.getEnvironmentCode()));
- processInstance.setTimeout(workflowDefinition.getTimeout());
- processInstance.setVarPool(null);
- processInstance.setDryRun(command.getDryRun());
- processInstance.setTestFlag(command.getTestFlag());
- processInstanceDao.insert(processInstance);
-
- workflowExecuteContextBuilder.setWorkflowInstance(processInstance);
+ workflowInstance.setStateWithDesc(WorkflowExecutionStatus.RUNNING_EXECUTION, command.getCommandType().name());
+ workflowInstance.setHost(masterConfig.getMasterAddress());
+ workflowInstance.setCommandParam(command.getCommandParam());
+ workflowInstance.setGlobalParams(mergeCommandParamsWithWorkflowParams(command, workflowDefinition));
+ processInstanceDao.upsertProcessInstance(workflowInstance);
+
+ workflowExecuteContextBuilder.setWorkflowInstance(workflowInstance);
}
@Override
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/ScheduleWorkflowCommandHandler.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/ScheduleWorkflowCommandHandler.java
index 2c7871d54b..02b720fe0e 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/ScheduleWorkflowCommandHandler.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/ScheduleWorkflowCommandHandler.java
@@ -18,13 +18,8 @@
package org.apache.dolphinscheduler.server.master.engine.command.handler;
import org.apache.dolphinscheduler.common.enums.CommandType;
-import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
-import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.scheduler.api.SchedulerApi;
-import org.apache.dolphinscheduler.server.master.config.MasterConfig;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;
/**
@@ -33,18 +28,6 @@ import org.springframework.stereotype.Component;
@Component
public class ScheduleWorkflowCommandHandler extends RunWorkflowCommandHandler {
- @Autowired
- private ProcessInstanceDao processInstanceDao;
-
- @Autowired
- private TaskInstanceDao taskInstanceDao;
-
- @Autowired
- private MasterConfig masterConfig;
-
- @Autowired
- private ApplicationContext applicationContext;
-
@Override
public CommandType commandType() {
return CommandType.SCHEDULER;
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/client/LogicTaskExecutorClientDelegator.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/client/LogicTaskExecutorClientDelegator.java
index 09d6142615..7a941f9e30 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/client/LogicTaskExecutorClientDelegator.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/client/LogicTaskExecutorClientDelegator.java
@@ -20,7 +20,7 @@ package org.apache.dolphinscheduler.server.master.engine.task.client;
import static com.google.common.base.Preconditions.checkArgument;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ILogicTaskInstanceOperator;
import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskKillRequest;
import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskKillResponse;
@@ -51,7 +51,7 @@ public class LogicTaskExecutorClientDelegator implements ITaskExecutorClientDele
final String taskName = taskInstance.getName();
checkArgument(StringUtils.isNotEmpty(executorHost), "Executor host is empty");
- final LogicTaskPauseResponse pauseResponse = SingletonJdkDynamicRpcClientProxyFactory
+ final LogicTaskPauseResponse pauseResponse = Clients
.withService(ILogicTaskInstanceOperator.class)
.withHost(taskInstance.getHost())
.pauseLogicTask(new LogicTaskPauseRequest(taskInstance.getId()));
@@ -69,7 +69,7 @@ public class LogicTaskExecutorClientDelegator implements ITaskExecutorClientDele
final String taskName = taskInstance.getName();
checkArgument(StringUtils.isNotEmpty(executorHost), "Executor host is empty");
- final LogicTaskKillResponse killResponse = SingletonJdkDynamicRpcClientProxyFactory
+ final LogicTaskKillResponse killResponse = Clients
.withService(ILogicTaskInstanceOperator.class)
.withHost(taskInstance.getHost())
.killLogicTask(new LogicTaskKillRequest(taskInstance.getId()));
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/client/PhysicalTaskExecutorClientDelegator.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/client/PhysicalTaskExecutorClientDelegator.java
index 45fdf18634..2db50f84c9 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/client/PhysicalTaskExecutorClientDelegator.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/client/PhysicalTaskExecutorClientDelegator.java
@@ -20,7 +20,7 @@ package org.apache.dolphinscheduler.server.master.engine.task.client;
import static com.google.common.base.Preconditions.checkArgument;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.worker.ITaskInstanceOperator;
import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstanceKillRequest;
import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstanceKillResponse;
@@ -50,7 +50,7 @@ public class PhysicalTaskExecutorClientDelegator implements ITaskExecutorClientD
final String taskName = taskInstance.getName();
checkArgument(StringUtils.isNotEmpty(executorHost), "Executor host is empty");
- final TaskInstancePauseResponse pauseResponse = SingletonJdkDynamicRpcClientProxyFactory
+ final TaskInstancePauseResponse pauseResponse = Clients
.withService(ITaskInstanceOperator.class)
.withHost(taskInstance.getHost())
.pauseTask(new TaskInstancePauseRequest(taskInstance.getId()));
@@ -68,7 +68,7 @@ public class PhysicalTaskExecutorClientDelegator implements ITaskExecutorClientD
final String taskName = taskInstance.getName();
checkArgument(StringUtils.isNotEmpty(executorHost), "Executor host is empty");
- final TaskInstanceKillResponse killResponse = SingletonJdkDynamicRpcClientProxyFactory
+ final TaskInstanceKillResponse killResponse = Clients
.withService(ITaskInstanceOperator.class)
.withHost(executorHost)
.killTask(new TaskInstanceKillRequest(taskInstance.getId()));
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java
index e95052fae4..905632cc24 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java
@@ -23,7 +23,7 @@ import static org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionSta
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.worker.ITaskInstanceOperator;
import org.apache.dolphinscheduler.extract.worker.transportor.TakeOverTaskRequest;
import org.apache.dolphinscheduler.extract.worker.transportor.TakeOverTaskResponse;
@@ -256,7 +256,7 @@ public abstract class AbstractTaskStateAction implements ITaskStateAction {
.taskInstanceId(taskExecutionRunnable.getTaskInstance().getId())
.workflowHost(masterConfig.getMasterAddress())
.build();
- final TakeOverTaskResponse takeOverTaskResponse = SingletonJdkDynamicRpcClientProxyFactory
+ final TakeOverTaskResponse takeOverTaskResponse = Clients
.withService(ITaskInstanceOperator.class)
.withHost(taskExecutionRunnable.getTaskInstance().getHost())
.takeOverTask(takeOverTaskRequest);
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/listener/WorkflowSuccessLifecycleListener.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/listener/WorkflowSuccessLifecycleListener.java
index f8394b0e38..af4d76a3e2 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/listener/WorkflowSuccessLifecycleListener.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/listener/WorkflowSuccessLifecycleListener.java
@@ -21,14 +21,16 @@ import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
-import org.apache.dolphinscheduler.dao.entity.Command;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.repository.CommandDao;
import org.apache.dolphinscheduler.extract.master.command.BackfillWorkflowCommandParam;
import org.apache.dolphinscheduler.extract.master.command.ICommandParam;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowBackfillTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowBackfillTriggerResponse;
import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.AbstractWorkflowLifecycleLifecycleEvent;
import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.WorkflowLifecycleEventType;
import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
+import org.apache.dolphinscheduler.server.master.engine.workflow.trigger.WorkflowBackfillTrigger;
import org.apache.commons.collections4.CollectionUtils;
@@ -43,6 +45,9 @@ import org.springframework.stereotype.Component;
@Component
public class WorkflowSuccessLifecycleListener implements IWorkflowLifecycleListener {
+ @Autowired
+ private WorkflowBackfillTrigger workflowBackfillTrigger;
+
@Autowired
private CommandDao commandDao;
@@ -76,30 +81,30 @@ public class WorkflowSuccessLifecycleListener implements IWorkflowLifecycleListe
if (CollectionUtils.isEmpty(backfillTimeList)) {
return;
}
- final BackfillWorkflowCommandParam nextCommandParam = BackfillWorkflowCommandParam.builder()
+ final WorkflowBackfillTriggerRequest backfillTriggerRequest = WorkflowBackfillTriggerRequest.builder()
+ .userId(workflowInstance.getExecutorId())
.backfillTimeList(backfillTimeList)
- .commandParams(commandParam.getCommandParams())
- .timeZone(commandParam.getTimeZone())
- .build();
-
- final Command command = Command.builder()
- .commandType(CommandType.COMPLEMENT_DATA)
- .processDefinitionCode(workflowInstance.getProcessDefinitionCode())
- .processDefinitionVersion(workflowInstance.getProcessDefinitionVersion())
- .executorId(workflowInstance.getExecutorId())
+ .workflowCode(workflowInstance.getProcessDefinitionCode())
+ .workflowVersion(workflowInstance.getProcessDefinitionVersion())
+ .startNodes(commandParam.getStartNodes())
.failureStrategy(workflowInstance.getFailureStrategy())
.taskDependType(workflowInstance.getTaskDependType())
- .commandParam(JSONUtils.toJsonString(nextCommandParam))
- .scheduleTime(DateUtils.stringToDate(backfillTimeList.get(0)))
+ .execType(CommandType.COMPLEMENT_DATA)
.warningType(workflowInstance.getWarningType())
.warningGroupId(workflowInstance.getWarningGroupId())
+ .workflowInstancePriority(workflowInstance.getProcessInstancePriority())
.workerGroup(workflowInstance.getWorkerGroup())
.tenantCode(workflowInstance.getTenantCode())
.environmentCode(workflowInstance.getEnvironmentCode())
- .processInstancePriority(workflowInstance.getProcessInstancePriority())
+ .startParamList(commandParam.getCommandParams())
+ .dryRun(Flag.of(workflowInstance.getDryRun()))
+ .testFlag(Flag.of(workflowInstance.getTestFlag()))
.build();
-
- commandDao.insert(command);
+ final WorkflowBackfillTriggerResponse backfillTriggerResponse =
+ workflowBackfillTrigger.triggerWorkflow(backfillTriggerRequest);
+ if (!backfillTriggerResponse.isSuccess()) {
+ log.warn("Backfill workflow failed: {}", backfillTriggerResponse.getMessage());
+ }
}
@Override
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/IWorkflowExecutionRunnable.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/IWorkflowExecutionRunnable.java
index 4f85b6095f..abef61abc1 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/IWorkflowExecutionRunnable.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/IWorkflowExecutionRunnable.java
@@ -31,34 +31,75 @@ public interface IWorkflowExecutionRunnable {
/**
* Get the id of the WorkflowExecutionRunnable.
*/
- int getId();
+ default int getId() {
+ return getWorkflowInstance().getId();
+ }
/**
* Get the name of the WorkflowExecutionRunnable.
*/
- String getName();
+ default String getName() {
+ return getWorkflowInstance().getName();
+ }
+ /**
+ * Pause the WorkflowExecutionRunnable.
+ */
+ void pause();
+
+ /**
+ * Whether the workflow is ready to pause.
+ */
+ default boolean isWorkflowReadyPause() {
+ final WorkflowExecutionStatus workflowExecutionStatus = getWorkflowInstance().getState();
+ return workflowExecutionStatus == WorkflowExecutionStatus.READY_PAUSE;
+ }
+
+ /**
+ * Stop the WorkflowExecutionRunnable.
+ */
+ void stop();
+
+ /**
+ * Whether the workflow is ready to stop.
+ */
+ default boolean isWorkflowReadyStop() {
+ final WorkflowExecutionStatus workflowExecutionStatus = getWorkflowInstance().getState();
+ return workflowExecutionStatus == WorkflowExecutionStatus.READY_STOP;
+ }
+
+ /**
+ * Get the WorkflowExecuteContext belongs to the WorkflowExecutionRunnable.
+ */
IWorkflowExecuteContext getWorkflowExecuteContext();
/**
* Get the WorkflowInstance belongs to the WorkflowExecutionRunnable.
*/
- ProcessInstance getWorkflowInstance();
+ default ProcessInstance getWorkflowInstance() {
+ return getWorkflowExecuteContext().getWorkflowInstance();
+ }
/**
* Get the state of the WorkflowExecutionRunnable.
*/
- WorkflowExecutionStatus getState();
+ default WorkflowExecutionStatus getState() {
+ return getWorkflowInstance().getState();
+ }
/**
* Get the WorkflowEventBus belongs to the Workflow instance.
*/
- WorkflowEventBus getWorkflowEventBus();
+ default WorkflowEventBus getWorkflowEventBus() {
+ return getWorkflowExecuteContext().getWorkflowEventBus();
+ }
/**
* Get the WorkflowExecutionGraph belongs to the Workflow instance.
*/
- IWorkflowExecutionGraph getWorkflowExecutionGraph();
+ default IWorkflowExecutionGraph getWorkflowExecutionGraph() {
+ return getWorkflowExecuteContext().getWorkflowExecutionGraph();
+ }
/**
* Get the WorkflowInstanceLifecycleListeners belongs to the Workflow instance.
@@ -66,12 +107,8 @@ public interface IWorkflowExecutionRunnable {
List<IWorkflowLifecycleListener> getWorkflowLifecycleListeners();
/**
- * Whether the workflow is ready to pause.
+ * Register a WorkflowInstanceLifecycleListener to the Workflow instance.
*/
- boolean isWorkflowReadyPause();
-
- boolean isWorkflowReadyStop();
-
void registerWorkflowInstanceLifecycleListener(IWorkflowLifecycleListener listener);
}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/WorkflowExecutionRunnable.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/WorkflowExecutionRunnable.java
index c24ce17677..73e17aac74 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/WorkflowExecutionRunnable.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/WorkflowExecutionRunnable.java
@@ -17,10 +17,10 @@
package org.apache.dolphinscheduler.server.master.engine.workflow.runnable;
-import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
-import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
-import org.apache.dolphinscheduler.server.master.engine.WorkflowEventBus;
-import org.apache.dolphinscheduler.server.master.engine.graph.IWorkflowExecutionGraph;
+import static com.google.common.base.Preconditions.checkArgument;
+
+import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowPauseLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowStopLifecycleEvent;
import org.apache.dolphinscheduler.server.master.engine.workflow.listener.IWorkflowLifecycleListener;
import org.apache.dolphinscheduler.server.master.runner.IWorkflowExecuteContext;
@@ -37,56 +37,23 @@ public class WorkflowExecutionRunnable implements IWorkflowExecutionRunnable {
@Getter
private final IWorkflowExecuteContext workflowExecuteContext;
- private final ProcessInstance workflowInstance;
-
- @Getter
- private final IWorkflowExecutionGraph workflowExecutionGraph;
-
- @Getter
- private final WorkflowEventBus workflowEventBus;
-
@Getter
private final List<IWorkflowLifecycleListener> workflowInstanceLifecycleListeners;
public WorkflowExecutionRunnable(WorkflowExecutionRunnableBuilder workflowExecutionRunnableBuilder) {
final ApplicationContext applicationContext = workflowExecutionRunnableBuilder.getApplicationContext();
this.workflowExecuteContext = workflowExecutionRunnableBuilder.getWorkflowExecuteContextBuilder().build();
- this.workflowInstance = workflowExecuteContext.getWorkflowInstance();
- this.workflowExecutionGraph = workflowExecuteContext.getWorkflowExecutionGraph();
- this.workflowEventBus = workflowExecuteContext.getWorkflowEventBus();
this.workflowInstanceLifecycleListeners = workflowExecuteContext.getWorkflowInstanceLifecycleListeners();
}
@Override
- public int getId() {
- return workflowInstance.getId();
- }
-
- @Override
- public String getName() {
- return workflowInstance.getName();
- }
-
- @Override
- public boolean isWorkflowReadyPause() {
- final WorkflowExecutionStatus workflowExecutionStatus = workflowInstance.getState();
- return workflowExecutionStatus == WorkflowExecutionStatus.READY_PAUSE;
- }
-
- @Override
- public boolean isWorkflowReadyStop() {
- final WorkflowExecutionStatus workflowExecutionStatus = workflowInstance.getState();
- return workflowExecutionStatus == WorkflowExecutionStatus.READY_STOP;
- }
-
- @Override
- public ProcessInstance getWorkflowInstance() {
- return workflowExecuteContext.getWorkflowInstance();
+ public void pause() {
+ getWorkflowEventBus().publish(WorkflowPauseLifecycleEvent.of(this));
}
@Override
- public WorkflowExecutionStatus getState() {
- return workflowInstance.getState();
+ public void stop() {
+ getWorkflowEventBus().publish(WorkflowStopLifecycleEvent.of(this));
}
@Override
@@ -96,6 +63,7 @@ public class WorkflowExecutionRunnable implements IWorkflowExecutionRunnable {
@Override
public void registerWorkflowInstanceLifecycleListener(IWorkflowLifecycleListener listener) {
+ checkArgument(listener != null, "listener cannot be null");
workflowInstanceLifecycleListeners.add(listener);
}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/WorkflowExecutionRunnableFactory.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/WorkflowExecutionRunnableFactory.java
index 33f8231a50..640083f030 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/WorkflowExecutionRunnableFactory.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/WorkflowExecutionRunnableFactory.java
@@ -23,7 +23,6 @@ import org.apache.dolphinscheduler.dao.repository.CommandDao;
import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
import org.apache.dolphinscheduler.server.master.engine.command.ICommandHandler;
import org.apache.dolphinscheduler.server.master.engine.exceptions.CommandDuplicateHandleException;
-import org.apache.dolphinscheduler.server.master.engine.workflow.listener.IWorkflowLifecycleListener;
import java.util.List;
@@ -46,9 +45,6 @@ public class WorkflowExecutionRunnableFactory {
@Autowired
private CommandDao commandDao;
- @Autowired
- private List<IWorkflowLifecycleListener> workflowLifecycleListeners;
-
/**
* Generate WorkflowExecutionRunnable from command.
* <p> We use transaction here to make sure that the command will be handled only once. Since in some case if the
@@ -57,9 +53,7 @@ public class WorkflowExecutionRunnableFactory {
@Transactional
public IWorkflowExecutionRunnable createWorkflowExecuteRunnable(Command command) {
deleteCommandOrThrow(command);
- final IWorkflowExecutionRunnable workflowExecutionRunnable = doCreateWorkflowExecutionRunnable(command);
- workflowLifecycleListeners.forEach(workflowExecutionRunnable::registerWorkflowInstanceLifecycleListener);
- return workflowExecutionRunnable;
+ return doCreateWorkflowExecutionRunnable(command);
}
/**
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/IWorkflowStateAction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/IWorkflowStateAction.java
index 1d5ae0b7b1..b2f12702d5 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/IWorkflowStateAction.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/IWorkflowStateAction.java
@@ -33,6 +33,7 @@ import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkf
* Represents the action to be taken when a workflow is in a certain state and receive a target event.
* <p> Each {@link WorkflowExecutionStatus} should have a corresponding {@link IWorkflowStateAction} implementation.
*
+ * @see WorkflowSubmittedStateAction
* @see WorkflowRunningStateAction
* @see WorkflowReadyPauseStateAction
* @see WorkflowPausedStateAction
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowStateActionFactory.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowStateActionFactory.java
index f65da11abf..3f01aff25f 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowStateActionFactory.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowStateActionFactory.java
@@ -18,7 +18,6 @@
package org.apache.dolphinscheduler.server.master.engine.workflow.statemachine;
import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
-import org.apache.dolphinscheduler.server.master.engine.WorkflowCacheRepository;
import java.util.Arrays;
import java.util.HashMap;
@@ -32,8 +31,7 @@ public class WorkflowStateActionFactory {
private final Map<WorkflowExecutionStatus, IWorkflowStateAction> workflowStateActionMap = new HashMap<>();
- public WorkflowStateActionFactory(List<IWorkflowStateAction> workflowStateActions,
- WorkflowCacheRepository workflowCacheRepository) {
+ public WorkflowStateActionFactory(List<IWorkflowStateAction> workflowStateActions) {
workflowStateActions.forEach(
workflowStateAction -> workflowStateActionMap.put(workflowStateAction.matchState(),
workflowStateAction));
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowSubmittedStateAction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowSubmittedStateAction.java
new file mode 100644
index 0000000000..9243b600fc
--- /dev/null
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowSubmittedStateAction.java
@@ -0,0 +1,121 @@
+/*
+ * 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.dolphinscheduler.server.master.engine.workflow.statemachine;
+
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
+import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowFailedLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowFinalizeLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowPauseLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowPausedLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowStartLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowStopLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowStoppedLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowSucceedLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowTopologyLogicalTransitionWithTaskFinishLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.springframework.stereotype.Component;
+
+/**
+ * The workflow submitted state should transform to running state when handle command.
+ */
+@Slf4j
+@Component
+public class WorkflowSubmittedStateAction extends AbstractWorkflowStateAction {
+
+ @Override
+ public void startEventAction(final IWorkflowExecutionRunnable workflowExecutionRunnable,
+ final WorkflowStartLifecycleEvent workflowStartEvent) {
+ throwExceptionIfStateIsNotMatch(workflowExecutionRunnable);
+ logWarningIfCannotDoAction(workflowExecutionRunnable, workflowStartEvent);
+ }
+
+ @Override
+ public void topologyLogicalTransitionEventAction(
+ final IWorkflowExecutionRunnable workflowExecutionRunnable,
+ final WorkflowTopologyLogicalTransitionWithTaskFinishLifecycleEvent workflowTopologyLogicalTransitionWithTaskFinishEvent) {
+ throwExceptionIfStateIsNotMatch(workflowExecutionRunnable);
+ logWarningIfCannotDoAction(workflowExecutionRunnable, workflowTopologyLogicalTransitionWithTaskFinishEvent);
+ }
+
+ @Override
+ public void pauseEventAction(final IWorkflowExecutionRunnable workflowExecutionRunnable,
+ final WorkflowPauseLifecycleEvent workflowPauseEvent) {
+ throwExceptionIfStateIsNotMatch(workflowExecutionRunnable);
+ logWarningIfCannotDoAction(workflowExecutionRunnable, workflowPauseEvent);
+ }
+
+ @Override
+ public void pausedEventAction(final IWorkflowExecutionRunnable workflowExecutionRunnable,
+ final WorkflowPausedLifecycleEvent workflowPausedEvent) {
+ throwExceptionIfStateIsNotMatch(workflowExecutionRunnable);
+ logWarningIfCannotDoAction(workflowExecutionRunnable, workflowPausedEvent);
+ }
+
+ @Override
+ public void stopEventAction(final IWorkflowExecutionRunnable workflowExecutionRunnable,
+ final WorkflowStopLifecycleEvent workflowStopEvent) {
+ throwExceptionIfStateIsNotMatch(workflowExecutionRunnable);
+ logWarningIfCannotDoAction(workflowExecutionRunnable, workflowStopEvent);
+ }
+
+ @Override
+ public void stoppedEventAction(final IWorkflowExecutionRunnable workflowExecutionRunnable,
+ final WorkflowStoppedLifecycleEvent workflowStoppedEvent) {
+ throwExceptionIfStateIsNotMatch(workflowExecutionRunnable);
+ logWarningIfCannotDoAction(workflowExecutionRunnable, workflowStoppedEvent);
+ }
+
+ @Override
+ public void succeedEventAction(final IWorkflowExecutionRunnable workflowExecutionRunnable,
+ final WorkflowSucceedLifecycleEvent workflowSucceedEvent) {
+ throwExceptionIfStateIsNotMatch(workflowExecutionRunnable);
+ logWarningIfCannotDoAction(workflowExecutionRunnable, workflowSucceedEvent);
+ }
+
+ @Override
+ public void failedEventAction(final IWorkflowExecutionRunnable workflowExecutionRunnable,
+ final WorkflowFailedLifecycleEvent workflowFailedEvent) {
+ throwExceptionIfStateIsNotMatch(workflowExecutionRunnable);
+ logWarningIfCannotDoAction(workflowExecutionRunnable, workflowFailedEvent);
+ }
+
+ @Override
+ public void finalizeEventAction(final IWorkflowExecutionRunnable workflowExecutionRunnable,
+ final WorkflowFinalizeLifecycleEvent workflowFinalizeEvent) {
+ throwExceptionIfStateIsNotMatch(workflowExecutionRunnable);
+ logWarningIfCannotDoAction(workflowExecutionRunnable, workflowFinalizeEvent);
+ }
+
+ @Override
+ public WorkflowExecutionStatus matchState() {
+ return WorkflowExecutionStatus.SUBMITTED_SUCCESS;
+ }
+
+ /**
+ * The running state can only finish with success/failure.
+ */
+ @Override
+ protected void emitWorkflowFinishedEventIfApplicable(IWorkflowExecutionRunnable workflowExecutionRunnable) {
+ throw new IllegalStateException(
+ "The workflow " + workflowExecutionRunnable.getName() +
+ "is submitted, shouldn't emit workflow finished event");
+ }
+}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/AbstractWorkflowTrigger.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/AbstractWorkflowTrigger.java
new file mode 100644
index 0000000000..1d58ab9339
--- /dev/null
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/AbstractWorkflowTrigger.java
@@ -0,0 +1,86 @@
+/*
+ * 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.dolphinscheduler.server.master.engine.workflow.trigger;
+
+import org.apache.dolphinscheduler.dao.entity.Command;
+import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
+import org.apache.dolphinscheduler.dao.entity.ProcessDefinitionLog;
+import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
+import org.apache.dolphinscheduler.dao.entity.User;
+import org.apache.dolphinscheduler.dao.repository.CommandDao;
+import org.apache.dolphinscheduler.dao.repository.ProcessDefinitionLogDao;
+import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
+import org.apache.dolphinscheduler.dao.repository.UserDao;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.transaction.annotation.Transactional;
+
+@Slf4j
+public abstract class AbstractWorkflowTrigger<TriggerRequest, TriggerResponse>
+ implements
+ IWorkflowTrigger<TriggerRequest, TriggerResponse> {
+
+ @Autowired
+ private ProcessDefinitionLogDao workflowDefinitionDao;
+
+ @Autowired
+ private ProcessInstanceDao workflowInstanceDao;
+
+ @Autowired
+ private UserDao userDao;
+
+ @Autowired
+ private CommandDao commandDao;
+
+ @Override
+ @Transactional
+ public TriggerResponse triggerWorkflow(final TriggerRequest triggerRequest) {
+ final ProcessInstance workflowInstance = constructWorkflowInstance(triggerRequest);
+ workflowInstanceDao.insert(workflowInstance);
+
+ final Command command = constructTriggerCommand(triggerRequest, workflowInstance);
+ commandDao.insert(command);
+
+ return onTriggerSuccess(workflowInstance);
+ }
+
+ protected abstract ProcessInstance constructWorkflowInstance(final TriggerRequest triggerRequest);
+
+ protected abstract Command constructTriggerCommand(final TriggerRequest triggerRequest,
+ final ProcessInstance workflowInstance);
+
+ protected abstract TriggerResponse onTriggerSuccess(final ProcessInstance workflowInstance);
+
+ protected ProcessDefinition getProcessDefinition(final Long workflowCode, final Integer workflowVersion) {
+ final ProcessDefinitionLog workflow = workflowDefinitionDao.queryByDefinitionCodeAndVersion(
+ workflowCode, workflowVersion);
+ if (workflow == null) {
+ throw new IllegalStateException(
+ "Workflow definition not found: " + workflowCode + " version " + workflowVersion);
+ }
+ return workflow;
+ }
+
+ protected User getExecutorUser(final Integer userId) {
+ return userDao.queryOptionalById(userId)
+ .orElseThrow(() -> new IllegalStateException("User not found: " + userId));
+ }
+
+}
diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopRequest.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/IWorkflowTrigger.java
similarity index 65%
rename from dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopRequest.java
rename to dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/IWorkflowTrigger.java
index f7ad1a38b9..97f98ffadc 100644
--- a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowInstanceStopRequest.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/IWorkflowTrigger.java
@@ -15,17 +15,16 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.extract.master.transportor;
+package org.apache.dolphinscheduler.server.master.engine.workflow.trigger;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-@Data
-@AllArgsConstructor
-@NoArgsConstructor
-public class WorkflowInstanceStopRequest {
+/**
+ * The trigger interface of the workflow, used to trigger the workflow and generate the workflow instance.
+ *
+ * @param <TriggerRequest> trigger request
+ * @param <TriggerResponse> trigger response
+ */
+public interface IWorkflowTrigger<TriggerRequest, TriggerResponse> {
- private Integer workflowInstanceId;
+ TriggerResponse triggerWorkflow(final TriggerRequest triggerRequest);
}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/WorkflowBackfillTrigger.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/WorkflowBackfillTrigger.java
new file mode 100644
index 0000000000..8318e59d37
--- /dev/null
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/WorkflowBackfillTrigger.java
@@ -0,0 +1,114 @@
+/*
+ * 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.dolphinscheduler.server.master.engine.workflow.trigger;
+
+import org.apache.dolphinscheduler.common.enums.CommandType;
+import org.apache.dolphinscheduler.common.enums.Flag;
+import org.apache.dolphinscheduler.common.enums.WarningType;
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
+import org.apache.dolphinscheduler.common.utils.DateUtils;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.dao.entity.Command;
+import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
+import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
+import org.apache.dolphinscheduler.dao.utils.EnvironmentUtils;
+import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
+import org.apache.dolphinscheduler.extract.master.command.BackfillWorkflowCommandParam;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowBackfillTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowBackfillTriggerResponse;
+
+import org.apache.commons.lang3.ObjectUtils;
+
+import java.util.Date;
+import java.util.List;
+
+import org.springframework.stereotype.Component;
+
+/**
+ * Backfill trigger of the workflow, used to trigger the workflow and generate the workflow instance in the backfill way.
+ */
+@Component
+public class WorkflowBackfillTrigger
+ extends
+ AbstractWorkflowTrigger<WorkflowBackfillTriggerRequest, WorkflowBackfillTriggerResponse> {
+
+ @Override
+ protected ProcessInstance constructWorkflowInstance(WorkflowBackfillTriggerRequest backfillTriggerRequest) {
+ final CommandType commandType = CommandType.COMPLEMENT_DATA;
+ final Long workflowCode = backfillTriggerRequest.getWorkflowCode();
+ final Integer workflowVersion = backfillTriggerRequest.getWorkflowVersion();
+ final List<String> backfillTimeList = backfillTriggerRequest.getBackfillTimeList();
+ final ProcessDefinition workflowDefinition = getProcessDefinition(workflowCode, workflowVersion);
+
+ final ProcessInstance workflowInstance = new ProcessInstance();
+ workflowInstance.setProcessDefinitionCode(workflowDefinition.getCode());
+ workflowInstance.setProcessDefinitionVersion(workflowDefinition.getVersion());
+ workflowInstance.setProjectCode(workflowDefinition.getProjectCode());
+ workflowInstance.setCommandType(commandType);
+ workflowInstance.setStateWithDesc(WorkflowExecutionStatus.SUBMITTED_SUCCESS, commandType.name());
+ workflowInstance.setRecovery(Flag.NO);
+ workflowInstance.setScheduleTime(DateUtils.stringToDate(backfillTimeList.get(0)));
+ workflowInstance.setStartTime(new Date());
+ workflowInstance.setRestartTime(workflowInstance.getStartTime());
+ workflowInstance.setRunTimes(1);
+ workflowInstance.setName(String.join("-", workflowDefinition.getName(), DateUtils.getCurrentTimeStamp()));
+ workflowInstance.setTaskDependType(backfillTriggerRequest.getTaskDependType());
+ workflowInstance.setFailureStrategy(backfillTriggerRequest.getFailureStrategy());
+ workflowInstance
+ .setWarningType(ObjectUtils.defaultIfNull(backfillTriggerRequest.getWarningType(), WarningType.NONE));
+ workflowInstance.setWarningGroupId(backfillTriggerRequest.getWarningGroupId());
+ workflowInstance.setExecutorId(backfillTriggerRequest.getUserId());
+ workflowInstance.setExecutorName(getExecutorUser(backfillTriggerRequest.getUserId()).getUserName());
+ workflowInstance.setTenantCode(backfillTriggerRequest.getTenantCode());
+ workflowInstance.setIsSubProcess(Flag.NO);
+ workflowInstance.addHistoryCmd(commandType);
+ workflowInstance.setProcessInstancePriority(backfillTriggerRequest.getWorkflowInstancePriority());
+ workflowInstance
+ .setWorkerGroup(WorkerGroupUtils.getWorkerGroupOrDefault(backfillTriggerRequest.getWorkerGroup()));
+ workflowInstance.setEnvironmentCode(
+ EnvironmentUtils.getEnvironmentCodeOrDefault(backfillTriggerRequest.getEnvironmentCode()));
+ workflowInstance.setTimeout(workflowDefinition.getTimeout());
+ workflowInstance.setDryRun(backfillTriggerRequest.getDryRun().getCode());
+ workflowInstance.setTestFlag(backfillTriggerRequest.getTestFlag().getCode());
+ return workflowInstance;
+ }
+
+ @Override
+ protected Command constructTriggerCommand(WorkflowBackfillTriggerRequest backfillTriggerRequest,
+ ProcessInstance workflowInstance) {
+ final BackfillWorkflowCommandParam backfillWorkflowCommandParam = BackfillWorkflowCommandParam.builder()
+ .commandParams(backfillTriggerRequest.getStartParamList())
+ .startNodes(backfillTriggerRequest.getStartNodes())
+ .timeZone(DateUtils.getTimezone())
+ .backfillTimeList(backfillTriggerRequest.getBackfillTimeList())
+ .build();
+ return Command.builder()
+ .commandType(backfillTriggerRequest.getExecType())
+ .processDefinitionCode(backfillTriggerRequest.getWorkflowCode())
+ .processDefinitionVersion(backfillTriggerRequest.getWorkflowVersion())
+ .processInstanceId(workflowInstance.getId())
+ .processInstancePriority(workflowInstance.getProcessInstancePriority())
+ .commandParam(JSONUtils.toJsonString(backfillWorkflowCommandParam))
+ .build();
+ }
+
+ @Override
+ protected WorkflowBackfillTriggerResponse onTriggerSuccess(ProcessInstance workflowInstance) {
+ return WorkflowBackfillTriggerResponse.success(workflowInstance.getId());
+ }
+}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/WorkflowManualTrigger.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/WorkflowManualTrigger.java
new file mode 100644
index 0000000000..b59df155bd
--- /dev/null
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/WorkflowManualTrigger.java
@@ -0,0 +1,111 @@
+/*
+ * 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.dolphinscheduler.server.master.engine.workflow.trigger;
+
+import org.apache.dolphinscheduler.common.enums.CommandType;
+import org.apache.dolphinscheduler.common.enums.Flag;
+import org.apache.dolphinscheduler.common.enums.WarningType;
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
+import org.apache.dolphinscheduler.common.utils.DateUtils;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.dao.entity.Command;
+import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
+import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
+import org.apache.dolphinscheduler.dao.utils.EnvironmentUtils;
+import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
+import org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowManualTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowManualTriggerResponse;
+
+import org.apache.commons.lang3.ObjectUtils;
+
+import java.util.Date;
+
+import org.springframework.stereotype.Component;
+
+/**
+ * Manual trigger of the workflow, used to trigger the workflow and generate the workflow instance in the manual way.
+ */
+@Component
+public class WorkflowManualTrigger
+ extends
+ AbstractWorkflowTrigger<WorkflowManualTriggerRequest, WorkflowManualTriggerResponse> {
+
+ @Override
+ protected ProcessInstance constructWorkflowInstance(final WorkflowManualTriggerRequest workflowManualTriggerRequest) {
+ final CommandType commandType = CommandType.START_PROCESS;
+ final Long workflowCode = workflowManualTriggerRequest.getWorkflowDefinitionCode();
+ final Integer workflowVersion = workflowManualTriggerRequest.getWorkflowDefinitionVersion();
+ final ProcessDefinition workflowDefinition = getProcessDefinition(workflowCode, workflowVersion);
+
+ final ProcessInstance workflowInstance = new ProcessInstance();
+ workflowInstance.setProcessDefinitionCode(workflowDefinition.getCode());
+ workflowInstance.setProcessDefinitionVersion(workflowDefinition.getVersion());
+ workflowInstance.setProjectCode(workflowDefinition.getProjectCode());
+ workflowInstance.setCommandType(commandType);
+ workflowInstance.setStateWithDesc(WorkflowExecutionStatus.SUBMITTED_SUCCESS, commandType.name());
+ workflowInstance.setRecovery(Flag.NO);
+ workflowInstance.setStartTime(new Date());
+ workflowInstance.setRestartTime(workflowInstance.getStartTime());
+ workflowInstance.setRunTimes(1);
+ workflowInstance.setName(String.join("-", workflowDefinition.getName(), DateUtils.getCurrentTimeStamp()));
+ workflowInstance.setTaskDependType(workflowManualTriggerRequest.getTaskDependType());
+ workflowInstance.setFailureStrategy(workflowManualTriggerRequest.getFailureStrategy());
+ workflowInstance.setWarningType(
+ ObjectUtils.defaultIfNull(workflowManualTriggerRequest.getWarningType(), WarningType.NONE));
+ workflowInstance.setWarningGroupId(workflowManualTriggerRequest.getWarningGroupId());
+ workflowInstance.setExecutorId(workflowManualTriggerRequest.getUserId());
+ workflowInstance.setExecutorName(getExecutorUser(workflowManualTriggerRequest.getUserId()).getUserName());
+ workflowInstance.setTenantCode(workflowManualTriggerRequest.getTenantCode());
+ workflowInstance.setIsSubProcess(Flag.NO);
+ workflowInstance.addHistoryCmd(commandType);
+ workflowInstance.setProcessInstancePriority(workflowManualTriggerRequest.getWorkflowInstancePriority());
+ workflowInstance.setWorkerGroup(
+ WorkerGroupUtils.getWorkerGroupOrDefault(workflowManualTriggerRequest.getWorkerGroup()));
+ workflowInstance.setEnvironmentCode(
+ EnvironmentUtils.getEnvironmentCodeOrDefault(workflowManualTriggerRequest.getEnvironmentCode()));
+ workflowInstance.setTimeout(workflowDefinition.getTimeout());
+ workflowInstance.setDryRun(workflowManualTriggerRequest.getDryRun().getCode());
+ workflowInstance.setTestFlag(workflowManualTriggerRequest.getTestFlag().getCode());
+ return workflowInstance;
+ }
+
+ @Override
+ protected Command constructTriggerCommand(final WorkflowManualTriggerRequest workflowManualTriggerRequest,
+ final ProcessInstance workflowInstance) {
+ final RunWorkflowCommandParam runWorkflowCommandParam = RunWorkflowCommandParam.builder()
+ .commandParams(workflowManualTriggerRequest.getStartParamList())
+ .startNodes(workflowManualTriggerRequest.getStartNodes())
+ .timeZone(DateUtils.getTimezone())
+ .build();
+ return Command.builder()
+ .commandType(CommandType.START_PROCESS)
+ .processDefinitionCode(workflowManualTriggerRequest.getWorkflowDefinitionCode())
+ .processDefinitionVersion(workflowManualTriggerRequest.getWorkflowDefinitionVersion())
+ .processInstanceId(workflowInstance.getId())
+ .processInstancePriority(workflowInstance.getProcessInstancePriority())
+ .commandParam(JSONUtils.toJsonString(runWorkflowCommandParam))
+ .build();
+ }
+
+ @Override
+ protected WorkflowManualTriggerResponse onTriggerSuccess(ProcessInstance workflowInstance) {
+ return WorkflowManualTriggerResponse.success(workflowInstance.getId());
+ }
+
+}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/WorkflowScheduleTrigger.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/WorkflowScheduleTrigger.java
new file mode 100644
index 0000000000..4753f99a8f
--- /dev/null
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/trigger/WorkflowScheduleTrigger.java
@@ -0,0 +1,106 @@
+/*
+ * 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.dolphinscheduler.server.master.engine.workflow.trigger;
+
+import org.apache.dolphinscheduler.common.enums.CommandType;
+import org.apache.dolphinscheduler.common.enums.Flag;
+import org.apache.dolphinscheduler.common.enums.WarningType;
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
+import org.apache.dolphinscheduler.common.utils.DateUtils;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.dao.entity.Command;
+import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
+import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
+import org.apache.dolphinscheduler.dao.utils.EnvironmentUtils;
+import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
+import org.apache.dolphinscheduler.extract.master.command.ScheduleWorkflowCommandParam;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowScheduleTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowScheduleTriggerResponse;
+
+import org.apache.commons.lang3.ObjectUtils;
+
+import java.util.Date;
+
+import org.springframework.stereotype.Component;
+
+@Component
+public class WorkflowScheduleTrigger
+ extends
+ AbstractWorkflowTrigger<WorkflowScheduleTriggerRequest, WorkflowScheduleTriggerResponse> {
+
+ @Override
+ protected ProcessInstance constructWorkflowInstance(WorkflowScheduleTriggerRequest scheduleTriggerRequest) {
+ final CommandType commandType = CommandType.SCHEDULER;
+ final Long workflowCode = scheduleTriggerRequest.getWorkflowCode();
+ final Integer workflowVersion = scheduleTriggerRequest.getWorkflowVersion();
+ final ProcessDefinition workflowDefinition = getProcessDefinition(workflowCode, workflowVersion);
+
+ final ProcessInstance workflowInstance = new ProcessInstance();
+ workflowInstance.setProcessDefinitionCode(workflowDefinition.getCode());
+ workflowInstance.setProcessDefinitionVersion(workflowDefinition.getVersion());
+ workflowInstance.setProjectCode(workflowDefinition.getProjectCode());
+ workflowInstance.setCommandType(commandType);
+ workflowInstance.setStateWithDesc(WorkflowExecutionStatus.SUBMITTED_SUCCESS, commandType.name());
+ workflowInstance.setRecovery(Flag.NO);
+ workflowInstance.setScheduleTime(scheduleTriggerRequest.getScheduleTIme());
+ workflowInstance.setStartTime(new Date());
+ workflowInstance.setRestartTime(workflowInstance.getStartTime());
+ workflowInstance.setRunTimes(1);
+ workflowInstance.setName(String.join("-", workflowDefinition.getName(), DateUtils.getCurrentTimeStamp()));
+ workflowInstance.setTaskDependType(scheduleTriggerRequest.getTaskDependType());
+ workflowInstance.setFailureStrategy(scheduleTriggerRequest.getFailureStrategy());
+ workflowInstance
+ .setWarningType(ObjectUtils.defaultIfNull(scheduleTriggerRequest.getWarningType(), WarningType.NONE));
+ workflowInstance.setWarningGroupId(scheduleTriggerRequest.getWarningGroupId());
+ workflowInstance.setExecutorId(scheduleTriggerRequest.getUserId());
+ workflowInstance.setExecutorName(getExecutorUser(scheduleTriggerRequest.getUserId()).getUserName());
+ workflowInstance.setTenantCode(scheduleTriggerRequest.getTenantCode());
+ workflowInstance.setIsSubProcess(Flag.NO);
+ workflowInstance.addHistoryCmd(commandType);
+ workflowInstance.setProcessInstancePriority(scheduleTriggerRequest.getWorkflowInstancePriority());
+ workflowInstance
+ .setWorkerGroup(WorkerGroupUtils.getWorkerGroupOrDefault(scheduleTriggerRequest.getWorkerGroup()));
+ workflowInstance.setEnvironmentCode(
+ EnvironmentUtils.getEnvironmentCodeOrDefault(scheduleTriggerRequest.getEnvironmentCode()));
+ workflowInstance.setTimeout(workflowDefinition.getTimeout());
+ workflowInstance.setDryRun(scheduleTriggerRequest.getDryRun().getCode());
+ workflowInstance.setTestFlag(scheduleTriggerRequest.getTestFlag().getCode());
+ return workflowInstance;
+ }
+
+ @Override
+ protected Command constructTriggerCommand(final WorkflowScheduleTriggerRequest scheduleTriggerRequest,
+ final ProcessInstance workflowInstance) {
+ final ScheduleWorkflowCommandParam scheduleWorkflowCommandParam = ScheduleWorkflowCommandParam.builder()
+ .timeZone(scheduleTriggerRequest.getTimezoneId())
+ .build();
+ return Command.builder()
+ .commandType(CommandType.SCHEDULER)
+ .processDefinitionCode(scheduleTriggerRequest.getWorkflowCode())
+ .processDefinitionVersion(scheduleTriggerRequest.getWorkflowVersion())
+ .processInstanceId(workflowInstance.getId())
+ .processInstancePriority(workflowInstance.getProcessInstancePriority())
+ .commandParam(JSONUtils.toJsonString(scheduleWorkflowCommandParam))
+ .build();
+ }
+
+ @Override
+ protected WorkflowScheduleTriggerResponse onTriggerSuccess(ProcessInstance workflowInstance) {
+ return WorkflowScheduleTriggerResponse.success(workflowInstance.getId());
+ }
+}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutionEventListenerImpl.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutionEventListenerImpl.java
index e700764d83..b788337033 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutionEventListenerImpl.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutionEventListenerImpl.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.master.rpc;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.transportor.ITaskExecutionEvent;
import org.apache.dolphinscheduler.extract.master.transportor.TaskExecutionDispatchEvent;
@@ -73,7 +73,7 @@ public class TaskExecutionEventListenerImpl implements ITaskExecutionEventListen
// instance state.
// The logic task doesn't need to send ack
if (!TaskTypeUtils.isLogicTask(taskExecutionRunnable.getTaskDefinition().getTaskType())) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskInstanceExecutionEventAckListener.class)
.withHost(taskExecutionDispatchEvent.getTaskInstanceHost())
.handleTaskInstanceDispatchedEventAck(
@@ -99,7 +99,7 @@ public class TaskExecutionEventListenerImpl implements ITaskExecutionEventListen
// instance state.
// The logic task doesn't need to send ack
if (!TaskTypeUtils.isLogicTask(taskExecutionRunnable.getTaskDefinition().getTaskType())) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskInstanceExecutionEventAckListener.class)
.withHost(taskExecutionRunningEvent.getTaskInstanceHost())
.handleTaskInstanceExecutionRunningEventAck(
@@ -122,7 +122,7 @@ public class TaskExecutionEventListenerImpl implements ITaskExecutionEventListen
// So once the master failover and we take over the task instance success, then we should fetch the latest task
// instance state.
if (!TaskTypeUtils.isLogicTask(taskExecutionRunnable.getTaskDefinition().getTaskType())) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskInstanceExecutionEventAckListener.class)
.withHost(taskExecutionSuccessEvent.getTaskInstanceHost())
.handleTaskExecutionSuccessEventAck(
@@ -143,7 +143,7 @@ public class TaskExecutionEventListenerImpl implements ITaskExecutionEventListen
// So once the master failover and we take over the task instance success, then we should fetch the latest task
// instance state.
if (!TaskTypeUtils.isLogicTask(taskExecutionRunnable.getTaskDefinition().getTaskType())) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskInstanceExecutionEventAckListener.class)
.withHost(taskExecutionFailedEvent.getTaskInstanceHost())
.handleTaskExecutionFailedEventAck(
@@ -164,7 +164,7 @@ public class TaskExecutionEventListenerImpl implements ITaskExecutionEventListen
// So once the master failover and we take over the task instance success, then we should fetch the latest task
// instance state.
if (!TaskTypeUtils.isLogicTask(taskExecutionRunnable.getTaskDefinition().getTaskType())) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskInstanceExecutionEventAckListener.class)
.withHost(taskExecutionKilledEvent.getTaskInstanceHost())
.handleTaskExecutionKilledEventAck(
@@ -182,7 +182,7 @@ public class TaskExecutionEventListenerImpl implements ITaskExecutionEventListen
// So once the master failover and we take over the task instance success, then we should fetch the latest task
// instance state.
if (!TaskTypeUtils.isLogicTask(taskExecutionRunnable.getTaskDefinition().getTaskType())) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskInstanceExecutionEventAckListener.class)
.withHost(taskExecutionPausedEvent.getTaskInstanceHost())
.handleTaskExecutionPausedEventAck(
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/WorkflowControlClient.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/WorkflowControlClient.java
new file mode 100644
index 0000000000..eca632ac45
--- /dev/null
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/WorkflowControlClient.java
@@ -0,0 +1,126 @@
+/*
+ * 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.dolphinscheduler.server.master.rpc;
+
+import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowBackfillTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowBackfillTriggerResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstancePauseRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstancePauseResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowManualTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowManualTriggerResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowScheduleTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowScheduleTriggerResponse;
+import org.apache.dolphinscheduler.server.master.engine.WorkflowCacheRepository;
+import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
+import org.apache.dolphinscheduler.server.master.engine.workflow.trigger.WorkflowBackfillTrigger;
+import org.apache.dolphinscheduler.server.master.engine.workflow.trigger.WorkflowManualTrigger;
+import org.apache.dolphinscheduler.server.master.engine.workflow.trigger.WorkflowScheduleTrigger;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+public class WorkflowControlClient implements IWorkflowControlClient {
+
+ @Autowired
+ private WorkflowManualTrigger workflowManualTrigger;
+
+ @Autowired
+ private WorkflowBackfillTrigger workflowBackfillTrigger;
+
+ @Autowired
+ private WorkflowScheduleTrigger workflowScheduleTrigger;
+
+ @Autowired
+ private WorkflowCacheRepository workflowRepository;
+
+ @Override
+ public WorkflowManualTriggerResponse manualTriggerWorkflow(final WorkflowManualTriggerRequest manualTriggerRequest) {
+ try {
+ return workflowManualTrigger.triggerWorkflow(manualTriggerRequest);
+ } catch (Exception ex) {
+ log.error("Handle workflowTriggerRequest: {} failed", manualTriggerRequest, ex);
+ return WorkflowManualTriggerResponse.fail("Trigger workflow failed: " + ExceptionUtils.getMessage(ex));
+ }
+ }
+
+ @Override
+ public WorkflowBackfillTriggerResponse backfillTriggerWorkflow(final WorkflowBackfillTriggerRequest backfillTriggerRequest) {
+ try {
+ return workflowBackfillTrigger.triggerWorkflow(backfillTriggerRequest);
+ } catch (Exception ex) {
+ log.error("Handle workflowBackfillTriggerRequest: {} failed", backfillTriggerRequest, ex);
+ return WorkflowBackfillTriggerResponse.fail("Backfill workflow failed: " + ExceptionUtils.getMessage(ex));
+ }
+ }
+
+ @Override
+ public WorkflowScheduleTriggerResponse scheduleTriggerWorkflow(WorkflowScheduleTriggerRequest workflowScheduleTriggerRequest) {
+ try {
+ return workflowScheduleTrigger.triggerWorkflow(workflowScheduleTriggerRequest);
+ } catch (Exception ex) {
+ log.error("Handle workflowScheduleTriggerRequest: {} failed", workflowScheduleTriggerRequest, ex);
+ return WorkflowScheduleTriggerResponse
+ .fail("Schedule trigger workflow failed: " + ExceptionUtils.getMessage(ex));
+ }
+ }
+
+ @Override
+ public WorkflowInstancePauseResponse pauseWorkflowInstance(final WorkflowInstancePauseRequest workflowInstancePauseRequest) {
+ try {
+ final Integer workflowInstanceId = workflowInstancePauseRequest.getWorkflowInstanceId();
+ final IWorkflowExecutionRunnable workflow = workflowRepository.get(workflowInstanceId);
+ if (workflow == null) {
+ return WorkflowInstancePauseResponse.fail(
+ "Cannot find the WorkflowExecuteRunnable: " + workflowInstanceId);
+ }
+ workflow.pause();
+ return WorkflowInstancePauseResponse.success();
+ } catch (Exception ex) {
+ log.error("Handle workflowInstancePauseRequest: {} failed", workflowInstancePauseRequest, ex);
+ return WorkflowInstancePauseResponse.fail(
+ "Pause workflow instance failed: " + ExceptionUtils.getMessage(ex));
+ }
+ }
+
+ @Override
+ public WorkflowInstanceStopResponse stopWorkflowInstance(final WorkflowInstanceStopRequest workflowInstanceStopRequest) {
+ try {
+ final Integer workflowInstanceId = workflowInstanceStopRequest.getWorkflowInstanceId();
+ final IWorkflowExecutionRunnable workflow = workflowRepository.get(workflowInstanceId);
+ if (workflow == null) {
+ return WorkflowInstanceStopResponse
+ .fail("Cannot find the WorkflowExecuteRunnable: " + workflowInstanceId);
+ }
+ workflow.stop();
+ return WorkflowInstanceStopResponse.success();
+ } catch (Exception ex) {
+ log.error("Handle workflowInstanceStopRequest: {} failed", workflowInstanceStopRequest, ex);
+ return WorkflowInstanceStopResponse.fail(
+ "Stop workflow instance failed:" + ExceptionUtils.getMessage(ex));
+ }
+ }
+}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/WorkflowInstanceControllerImpl.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/WorkflowInstanceControllerImpl.java
deleted file mode 100644
index 5c8ac87d00..0000000000
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/WorkflowInstanceControllerImpl.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * 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.dolphinscheduler.server.master.rpc;
-
-import org.apache.dolphinscheduler.extract.master.IWorkflowInstanceController;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstancePauseRequest;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstancePauseResponse;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStopRequest;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStopResponse;
-import org.apache.dolphinscheduler.server.master.engine.WorkflowCacheRepository;
-import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowPauseLifecycleEvent;
-import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowStopLifecycleEvent;
-import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
-
-import org.apache.commons.lang3.exception.ExceptionUtils;
-
-import lombok.extern.slf4j.Slf4j;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-@Slf4j
-@Service
-public class WorkflowInstanceControllerImpl implements IWorkflowInstanceController {
-
- @Autowired
- private WorkflowCacheRepository workflowCacheRepository;
-
- @Override
- public WorkflowInstancePauseResponse pauseWorkflowInstance(final WorkflowInstancePauseRequest workflowInstancePauseRequest) {
- try {
- final Integer workflowInstanceId = workflowInstancePauseRequest.getWorkflowInstanceId();
- final IWorkflowExecutionRunnable workflowExecutionRunnable =
- workflowCacheRepository.get(workflowInstanceId);
- if (workflowExecutionRunnable == null) {
- return WorkflowInstancePauseResponse
- .fail("Cannot find the WorkflowExecuteRunnable: " + workflowInstanceId);
- }
- workflowExecutionRunnable.getWorkflowEventBus()
- .publish(WorkflowPauseLifecycleEvent.of(workflowExecutionRunnable));
- return WorkflowInstancePauseResponse.success();
- } catch (Exception ex) {
- log.error("Handle workflowInstancePauseRequest: {} failed", workflowInstancePauseRequest, ex);
- return WorkflowInstancePauseResponse.fail(ExceptionUtils.getMessage(ex));
- }
- }
-
- @Override
- public WorkflowInstanceStopResponse stopWorkflowInstance(final WorkflowInstanceStopRequest workflowInstanceStopRequest) {
- try {
- final Integer workflowInstanceId = workflowInstanceStopRequest.getWorkflowInstanceId();
- final IWorkflowExecutionRunnable workflowExecutionRunnable =
- workflowCacheRepository.get(workflowInstanceId);
- if (workflowExecutionRunnable == null) {
- return WorkflowInstanceStopResponse
- .fail("Cannot find the WorkflowExecuteRunnable: " + workflowInstanceId);
- }
- workflowExecutionRunnable.getWorkflowEventBus()
- .publish(WorkflowStopLifecycleEvent.of(workflowExecutionRunnable));
- return WorkflowInstanceStopResponse.success();
- } catch (Exception ex) {
- log.error("Handle workflowInstanceStopRequest: {} failed", workflowInstanceStopRequest, ex);
- return WorkflowInstanceStopResponse.fail(ExceptionUtils.getMessage(ex));
- }
- }
-}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteContext.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteContext.java
index 380194b5ed..703edf17b0 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteContext.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteContext.java
@@ -77,11 +77,6 @@ public class WorkflowExecuteContext implements IWorkflowExecuteContext {
return this;
}
- public WorkflowExecuteContextBuilder withWorkflowInstanceLifecycleListeners(List<IWorkflowLifecycleListener> workflowLifecycleListeners) {
- this.workflowInstanceLifecycleListeners = workflowLifecycleListeners;
- return this;
- }
-
public WorkflowExecuteContext build() {
return new WorkflowExecuteContext(
command,
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/dispatcher/WorkerTaskDispatcher.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/dispatcher/WorkerTaskDispatcher.java
index 802efa063c..9ff7f75772 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/dispatcher/WorkerTaskDispatcher.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/dispatcher/WorkerTaskDispatcher.java
@@ -19,7 +19,7 @@ package org.apache.dolphinscheduler.server.master.runner.dispatcher;
import static com.google.common.base.Preconditions.checkNotNull;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.base.utils.Host;
import org.apache.dolphinscheduler.extract.worker.ITaskInstanceOperator;
import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstanceDispatchRequest;
@@ -51,7 +51,7 @@ public class WorkerTaskDispatcher extends BaseTaskDispatcher {
final String taskName = taskExecutionContext.getTaskName();
final String workerAddress = taskExecutionContext.getHost();
try {
- final TaskInstanceDispatchResponse taskInstanceDispatchResponse = SingletonJdkDynamicRpcClientProxyFactory
+ final TaskInstanceDispatchResponse taskInstanceDispatchResponse = Clients
.withService(ITaskInstanceOperator.class)
.withHost(workerAddress)
.dispatchTask(new TaskInstanceDispatchRequest(ITaskExecutionRunnable.getTaskExecutionContext()));
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecuteDispatchEventSender.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecuteDispatchEventSender.java
index eebfe8c7a1..a284a30792 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecuteDispatchEventSender.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecuteDispatchEventSender.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.master.runner.message;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.transportor.TaskExecutionDispatchEvent;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
@@ -31,7 +31,7 @@ public class LogicTaskInstanceExecuteDispatchEventSender
@Override
public void sendMessage(TaskExecutionDispatchEvent taskExecutionDispatchEvent) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskExecutionEventListener.class)
.withHost(taskExecutionDispatchEvent.getWorkflowInstanceHost())
.onTaskInstanceDispatched(taskExecutionDispatchEvent);
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecuteRunningEventSender.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecuteRunningEventSender.java
index 1735b27c9a..762f19e29a 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecuteRunningEventSender.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecuteRunningEventSender.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.master.runner.message;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.transportor.TaskExecutionRunningEvent;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
@@ -33,7 +33,7 @@ public class LogicTaskInstanceExecuteRunningEventSender
@Override
public void sendMessage(TaskExecutionRunningEvent taskInstanceExecutionRunningEvent) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskExecutionEventListener.class)
.withHost(taskInstanceExecutionRunningEvent.getWorkflowInstanceHost())
.onTaskInstanceExecutionRunning(taskInstanceExecutionRunningEvent);
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionFailedEventSender.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionFailedEventSender.java
index 2022bdc3fb..5902919cf5 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionFailedEventSender.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionFailedEventSender.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.master.runner.message;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.transportor.TaskExecutionFailedEvent;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
@@ -31,7 +31,7 @@ public class LogicTaskInstanceExecutionFailedEventSender
@Override
public void sendMessage(final TaskExecutionFailedEvent message) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskExecutionEventListener.class)
.withHost(message.getWorkflowInstanceHost())
.onTaskInstanceExecutionFailed(message);
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionKilledEventSender.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionKilledEventSender.java
index 70d2c939c3..ec3b77e3ad 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionKilledEventSender.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionKilledEventSender.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.master.runner.message;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.transportor.TaskExecutionKilledEvent;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
@@ -31,7 +31,7 @@ public class LogicTaskInstanceExecutionKilledEventSender
@Override
public void sendMessage(final TaskExecutionKilledEvent message) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskExecutionEventListener.class)
.withHost(message.getWorkflowInstanceHost())
.onTaskInstanceExecutionKilled(message);
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionPausedEventSender.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionPausedEventSender.java
index d2d4ec593f..1b9718c939 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionPausedEventSender.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionPausedEventSender.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.master.runner.message;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.transportor.TaskExecutionPausedEvent;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
@@ -31,7 +31,7 @@ public class LogicTaskInstanceExecutionPausedEventSender
@Override
public void sendMessage(final TaskExecutionPausedEvent taskExecutionPausedEvent) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskExecutionEventListener.class)
.withHost(taskExecutionPausedEvent.getWorkflowInstanceHost())
.onTaskInstanceExecutionPaused(taskExecutionPausedEvent);
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionSuccessEventSender.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionSuccessEventSender.java
index 655caa087f..fc29520bf9 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionSuccessEventSender.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionSuccessEventSender.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.master.runner.message;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.transportor.TaskExecutionSuccessEvent;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
@@ -31,7 +31,7 @@ public class LogicTaskInstanceExecutionSuccessEventSender
@Override
public void sendMessage(final TaskExecutionSuccessEvent message) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskExecutionEventListener.class)
.withHost(message.getWorkflowInstanceHost())
.onTaskInstanceExecutionSuccess(message);
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/dynamic/DynamicLogicTask.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/dynamic/DynamicLogicTask.java
index 5da82f571c..581ab9c646 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/dynamic/DynamicLogicTask.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/dynamic/DynamicLogicTask.java
@@ -30,10 +30,10 @@ import org.apache.dolphinscheduler.dao.mapper.CommandMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
-import org.apache.dolphinscheduler.extract.master.IWorkflowInstanceController;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStopRequest;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStopResponse;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
+import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopResponse;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.model.DynamicInputParameter;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
@@ -283,8 +283,8 @@ public class DynamicLogicTask extends BaseAsyncLogicTask<DynamicParameters> {
private void doKillRunningSubWorkflowInstances(List<ProcessInstance> runningSubProcessInstanceList) throws MasterTaskExecuteException {
for (ProcessInstance subProcessInstance : runningSubProcessInstanceList) {
try {
- WorkflowInstanceStopResponse workflowInstanceStopResponse = SingletonJdkDynamicRpcClientProxyFactory
- .withService(IWorkflowInstanceController.class)
+ WorkflowInstanceStopResponse workflowInstanceStopResponse = Clients
+ .withService(IWorkflowControlClient.class)
.withHost(subProcessInstance.getHost())
.stopWorkflowInstance(new WorkflowInstanceStopRequest(subProcessInstance.getId()));
if (workflowInstanceStopResponse.isSuccess()) {
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/subworkflow/SubWorkflowLogicTask.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/subworkflow/SubWorkflowLogicTask.java
index 176cb3ab4e..d1fa605aa3 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/subworkflow/SubWorkflowLogicTask.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/subworkflow/SubWorkflowLogicTask.java
@@ -25,14 +25,14 @@ import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.repository.CommandDao;
import org.apache.dolphinscheduler.dao.repository.ProcessDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
-import org.apache.dolphinscheduler.extract.master.IWorkflowInstanceController;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
+import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
import org.apache.dolphinscheduler.extract.master.command.ICommandParam;
import org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstancePauseRequest;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstancePauseResponse;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStopRequest;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStopResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstancePauseRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstancePauseResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopResponse;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.parameters.SubProcessParameters;
import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
@@ -99,8 +99,8 @@ public class SubWorkflowLogicTask extends BaseAsyncLogicTask<SubProcessParameter
taskExecutionContext.getTaskInstanceId());
try {
- WorkflowInstancePauseResponse pauseResponse = SingletonJdkDynamicRpcClientProxyFactory
- .withService(IWorkflowInstanceController.class)
+ WorkflowInstancePauseResponse pauseResponse = Clients
+ .withService(IWorkflowControlClient.class)
.withHost(subProcessInstance.getHost())
.pauseWorkflowInstance(new WorkflowInstancePauseRequest(subProcessInstance.getId()));
if (pauseResponse.isSuccess()) {
@@ -128,8 +128,8 @@ public class SubWorkflowLogicTask extends BaseAsyncLogicTask<SubProcessParameter
return;
}
try {
- WorkflowInstanceStopResponse stopResponse = SingletonJdkDynamicRpcClientProxyFactory
- .withService(IWorkflowInstanceController.class)
+ WorkflowInstanceStopResponse stopResponse = Clients
+ .withService(IWorkflowControlClient.class)
.withHost(subProcessInstance.getHost())
.stopWorkflowInstance(new WorkflowInstanceStopRequest(subProcessInstance.getId()));
if (stopResponse.isSuccess()) {
diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/Repository.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/Repository.java
index 9966ee3d9d..106399749d 100644
--- a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/Repository.java
+++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/Repository.java
@@ -53,6 +53,10 @@ public class Repository {
.collect(Collectors.toList());
}
+ public ProcessInstance queryWorkflowInstance(final Integer workflowInstanceId) {
+ return workflowInstanceDao.queryById(workflowInstanceId);
+ }
+
/**
* Return the list of task instances for a given workflow definition in ascending order of their IDs.
*/
@@ -65,4 +69,14 @@ public class Repository {
.collect(Collectors.toList());
}
+ /**
+ * Return the list of task instances for a given workflow definition in ascending order of their IDs.
+ */
+ public List<TaskInstance> queryTaskInstance(final Integer workflowInstanceId) {
+ return taskInstanceDao.queryByWorkflowInstanceId(workflowInstanceId)
+ .stream()
+ .sorted(Comparator.comparingInt(TaskInstance::getId))
+ .collect(Collectors.toList());
+ }
+
}
diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/WorkflowOperator.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/WorkflowOperator.java
index 63fb0c0cd4..062d2ef287 100644
--- a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/WorkflowOperator.java
+++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/WorkflowOperator.java
@@ -17,29 +17,27 @@
package org.apache.dolphinscheduler.server.master.it;
-import org.apache.dolphinscheduler.common.enums.CommandType;
-import org.apache.dolphinscheduler.common.utils.DateUtils;
-import org.apache.dolphinscheduler.common.utils.JSONUtils;
-import org.apache.dolphinscheduler.dao.entity.Command;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.Schedule;
-import org.apache.dolphinscheduler.dao.repository.CommandDao;
-import org.apache.dolphinscheduler.extract.master.IWorkflowInstanceController;
+import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
import org.apache.dolphinscheduler.extract.master.command.BackfillWorkflowCommandParam;
import org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstancePauseRequest;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstancePauseResponse;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStopRequest;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStopResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowBackfillTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowBackfillTriggerResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstancePauseRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstancePauseResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowManualTriggerRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowManualTriggerResponse;
import org.apache.dolphinscheduler.scheduler.api.SchedulerApi;
-import java.util.Date;
-
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
+import org.assertj.core.api.Assertions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -47,28 +45,42 @@ import org.springframework.stereotype.Component;
public class WorkflowOperator {
@Autowired
- private CommandDao commandDao;
-
- @Autowired
- private IWorkflowInstanceController workflowInstanceController;
+ private IWorkflowControlClient workflowInstanceController;
@Autowired
private SchedulerApi schedulerApi;
+ public Integer manualTriggerWorkflow(final WorkflowTriggerDTO workflowTriggerDTO) {
+ final WorkflowManualTriggerRequest workflowManualTriggerRequest = WorkflowManualTriggerRequest.builder()
+ .userId(workflowTriggerDTO.workflowDefinition.getUserId())
+ .workflowDefinitionCode(workflowTriggerDTO.workflowDefinition.getCode())
+ .workflowDefinitionVersion(workflowTriggerDTO.workflowDefinition.getVersion())
+ .startNodes(workflowTriggerDTO.getRunWorkflowCommandParam().getStartNodes())
+ .startParamList(workflowTriggerDTO.getRunWorkflowCommandParam().getCommandParams())
+ .build();
+
+ final WorkflowManualTriggerResponse manualTriggerWorkflowResponse =
+ workflowInstanceController.manualTriggerWorkflow(workflowManualTriggerRequest);
+ Assertions.assertThat(manualTriggerWorkflowResponse.isSuccess()).isTrue();
+
+ return manualTriggerWorkflowResponse.getWorkflowInstanceId();
+ }
+
public void backfillWorkflow(final WorkflowBackfillDTO workflowBackfillDTO) {
final ProcessDefinition workflowDefinition = workflowBackfillDTO.getWorkflow();
- final Command command = Command.builder()
- .commandType(CommandType.COMPLEMENT_DATA)
- .processDefinitionCode(workflowDefinition.getCode())
- .processDefinitionVersion(workflowDefinition.getVersion())
- .executorId(workflowDefinition.getUserId())
- .scheduleTime(DateUtils.stringToDate(
- workflowBackfillDTO.getBackfillWorkflowCommandParam().getBackfillTimeList().get(0)))
- .commandParam(JSONUtils.toJsonString(workflowBackfillDTO.getBackfillWorkflowCommandParam()))
- .startTime(new Date())
- .updateTime(new Date())
+
+ final WorkflowBackfillTriggerRequest backfillTriggerRequest = WorkflowBackfillTriggerRequest.builder()
+ .userId(workflowDefinition.getUserId())
+ .workflowCode(workflowDefinition.getCode())
+ .workflowVersion(workflowDefinition.getVersion())
+ .startNodes(workflowBackfillDTO.getBackfillWorkflowCommandParam().getStartNodes())
+ .startParamList(workflowBackfillDTO.getBackfillWorkflowCommandParam().getCommandParams())
+ .backfillTimeList(workflowBackfillDTO.getBackfillWorkflowCommandParam().getBackfillTimeList())
.build();
- commandDao.insert(command);
+ final WorkflowBackfillTriggerResponse backfillTriggerResponse =
+ workflowInstanceController.backfillTriggerWorkflow(backfillTriggerRequest);
+
+ Assertions.assertThat(backfillTriggerResponse.isSuccess()).isTrue();
}
public void schedulingWorkflow(final WorkflowSchedulingDTO workflowSchedulingDTO) {
@@ -77,21 +89,6 @@ public class WorkflowOperator {
schedulerApi.insertOrUpdateScheduleTask(project.getId(), schedule);
}
- public void triggerWorkflow(final WorkflowTriggerDTO workflowTriggerDTO) {
- final ProcessDefinition workflowDefinition = workflowTriggerDTO.getWorkflowDefinition();
- final RunWorkflowCommandParam runWorkflowCommandParam = workflowTriggerDTO.getRunWorkflowCommandParam();
- final Command command = Command.builder()
- .commandType(CommandType.START_PROCESS)
- .processDefinitionCode(workflowDefinition.getCode())
- .processDefinitionVersion(workflowDefinition.getVersion())
- .executorId(workflowDefinition.getUserId())
- .commandParam(JSONUtils.toJsonString(runWorkflowCommandParam))
- .startTime(new Date())
- .updateTime(new Date())
- .build();
- commandDao.insert(command);
- }
-
public WorkflowInstancePauseResponse pauseWorkflowInstance(Integer workflowInstanceId) {
final WorkflowInstancePauseRequest workflowInstancePauseRequest =
new WorkflowInstancePauseRequest(workflowInstanceId);
diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/cases/WorkflowInstancePauseIT.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/cases/WorkflowInstancePauseIT.java
index 52f0fe44f2..31b39fd345 100644
--- a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/cases/WorkflowInstancePauseIT.java
+++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/cases/WorkflowInstancePauseIT.java
@@ -21,7 +21,6 @@ import static com.google.common.truth.Truth.assertThat;
import static org.awaitility.Awaitility.await;
import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
-import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
@@ -67,30 +66,37 @@ public class WorkflowInstancePauseIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ final Integer workflowInstanceId = workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.pollInterval(Duration.ofMillis(100))
.atMost(Duration.ofMinutes(1))
- .untilAsserted(() -> Assertions.assertThat(repository.queryWorkflowInstance(workflow)).hasSize(1));
-
- // Make sure the task instance is running, otherwise will pause the submitted task instance
- // In fact, this case might pause the first three tasks if the first three tasks are not dispatched.
- ThreadUtils.sleep(1_000);
+ .untilAsserted(() -> {
+ Assertions
+ .assertThat(repository.queryWorkflowInstance(workflowInstanceId))
+ .satisfies(workflowInstance -> {
+ assertThat(workflowInstance.getState())
+ .isEqualTo(WorkflowExecutionStatus.RUNNING_EXECUTION);
+ });
+ Assertions
+ .assertThat(repository.queryTaskInstance(workflowInstanceId))
+ .satisfiesExactly(taskInstance -> {
+ assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ });
+ });
- final Integer workflowInstanceId = repository.queryWorkflowInstance(workflow).get(0).getId();
assertThat(workflowOperator.pauseWorkflowInstance(workflowInstanceId).isSuccess());
await()
.atMost(Duration.ofMinutes(1))
.untilAsserted(() -> {
Assertions
- .assertThat(repository.queryWorkflowInstance(workflow))
- .satisfiesExactly(workflowInstance -> {
+ .assertThat(repository.queryWorkflowInstance(workflowInstanceId))
+ .satisfies(workflowInstance -> {
assertThat(workflowInstance.getState()).isEqualTo(WorkflowExecutionStatus.SUCCESS);
});
Assertions
- .assertThat(repository.queryTaskInstance(workflow))
+ .assertThat(repository.queryTaskInstance(workflowInstanceId))
.satisfiesExactly(taskInstance -> {
assertThat(taskInstance.getName()).isEqualTo("A");
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
@@ -111,30 +117,38 @@ public class WorkflowInstancePauseIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ final Integer workflowInstanceId = workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.pollInterval(Duration.ofMillis(100))
.atMost(Duration.ofMinutes(1))
- .untilAsserted(() -> Assertions.assertThat(repository.queryWorkflowInstance(workflow)).hasSize(1));
-
- // Make sure the task instance is running, otherwise will pause the submitted task instance
- // In fact, this case might pause the first three tasks if the first three tasks are not dispatched.
- ThreadUtils.sleep(1_000);
+ .untilAsserted(() -> {
+ Assertions
+ .assertThat(repository.queryWorkflowInstance(workflowInstanceId))
+ .satisfies(workflowInstance -> {
+ assertThat(workflowInstance.getState())
+ .isEqualTo(WorkflowExecutionStatus.RUNNING_EXECUTION);
+ });
+ Assertions
+ .assertThat(
+ repository.queryTaskInstance(workflowInstanceId))
+ .satisfiesExactly(taskInstance -> {
+ assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ });
+ });
- final Integer workflowInstanceId = repository.queryWorkflowInstance(workflow).get(0).getId();
assertThat(workflowOperator.pauseWorkflowInstance(workflowInstanceId).isSuccess());
await()
.atMost(Duration.ofMinutes(1))
.untilAsserted(() -> {
Assertions
- .assertThat(repository.queryWorkflowInstance(workflow))
- .satisfiesExactly(workflowInstance -> {
+ .assertThat(repository.queryWorkflowInstance(workflowInstanceId))
+ .satisfies(workflowInstance -> {
assertThat(workflowInstance.getState()).isEqualTo(WorkflowExecutionStatus.FAILURE);
});
Assertions
- .assertThat(repository.queryTaskInstance(workflow))
+ .assertThat(repository.queryTaskInstance(workflowInstanceId))
.satisfiesExactly(taskInstance -> {
assertThat(taskInstance.getName()).isEqualTo("A");
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.FAILURE);
@@ -154,30 +168,45 @@ public class WorkflowInstancePauseIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ final Integer workflowInstanceId = workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.pollInterval(Duration.ofMillis(100))
.atMost(Duration.ofMinutes(1))
- .untilAsserted(() -> Assertions.assertThat(repository.queryWorkflowInstance(workflow)).hasSize(1));
-
- // Make sure the task instance is running, otherwise will pause the submitted task instance
- // In fact, this case might pause the first three tasks if the first three tasks are not dispatched.
- ThreadUtils.sleep(1_000);
+ .untilAsserted(() -> {
+ Assertions.assertThat(repository.queryWorkflowInstance(workflowInstanceId))
+ .satisfies(workflowInstance -> {
+ assertThat(workflowInstance.getState())
+ .isEqualTo(WorkflowExecutionStatus.RUNNING_EXECUTION);
+ });
+ Assertions
+ .assertThat(repository.queryTaskInstance(workflowInstanceId))
+ .anySatisfy(taskInstance -> {
+ assertThat(taskInstance.getName()).isEqualTo("A1");
+ assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ })
+ .anySatisfy(taskInstance -> {
+ assertThat(taskInstance.getName()).isEqualTo("B1");
+ assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ })
+ .anySatisfy(taskInstance -> {
+ assertThat(taskInstance.getName()).isEqualTo("C1");
+ assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ });
+ });
- final Integer workflowInstanceId = repository.queryWorkflowInstance(workflow).get(0).getId();
assertThat(workflowOperator.pauseWorkflowInstance(workflowInstanceId).isSuccess());
await()
.atMost(Duration.ofMinutes(1))
.untilAsserted(() -> {
Assertions
- .assertThat(repository.queryWorkflowInstance(workflow))
- .satisfiesExactly(workflowInstance -> {
+ .assertThat(repository.queryWorkflowInstance(workflowInstanceId))
+ .satisfies(workflowInstance -> {
assertThat(workflowInstance.getState()).isEqualTo(WorkflowExecutionStatus.PAUSE);
});
Assertions
- .assertThat(repository.queryTaskInstance(workflow))
+ .assertThat(repository.queryTaskInstance(workflowInstanceId))
.hasSize(6)
.anySatisfy(taskInstance -> {
assertThat(taskInstance.getName()).isEqualTo("A1");
diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/cases/WorkflowInstanceStopIT.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/cases/WorkflowInstanceStopIT.java
index f47cf19f3e..b944d3e98e 100644
--- a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/cases/WorkflowInstanceStopIT.java
+++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/cases/WorkflowInstanceStopIT.java
@@ -24,7 +24,7 @@ import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
-import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStopResponse;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopResponse;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.server.master.AbstractMasterIntegrationTest;
import org.apache.dolphinscheduler.server.master.engine.IWorkflowRepository;
@@ -68,14 +68,24 @@ public class WorkflowInstanceStopIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ final Integer workflowInstanceId = workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.pollInterval(Duration.ofMillis(100))
.atMost(Duration.ofMinutes(1))
- .untilAsserted(() -> Assertions.assertThat(repository.queryWorkflowInstance(workflow)).hasSize(1));
+ .untilAsserted(() -> {
+ Assertions
+ .assertThat(repository.queryWorkflowInstance(workflowInstanceId).getState())
+ .isEqualTo(WorkflowExecutionStatus.RUNNING_EXECUTION);
+ Assertions
+ .assertThat(repository.queryTaskInstance(workflowInstanceId))
+ .satisfiesExactly(
+ taskInstance -> {
+ assertThat(taskInstance.getState())
+ .isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ });
+ });
- final Integer workflowInstanceId = repository.queryWorkflowInstance(workflow).get(0).getId();
Assertions
.assertThat(workflowOperator.stopWorkflowInstance(workflowInstanceId))
.matches(WorkflowInstanceStopResponse::isSuccess);
@@ -84,13 +94,10 @@ public class WorkflowInstanceStopIT extends AbstractMasterIntegrationTest {
.atMost(Duration.ofMinutes(1))
.untilAsserted(() -> {
Assertions
- .assertThat(repository.queryWorkflowInstance(workflow))
- .satisfiesExactly(
- workflowInstance -> {
- assertThat(workflowInstance.getState()).isEqualTo(WorkflowExecutionStatus.STOP);
- });
+ .assertThat(repository.queryWorkflowInstance(workflowInstanceId).getState())
+ .isEqualTo(WorkflowExecutionStatus.STOP);
Assertions
- .assertThat(repository.queryTaskInstance(workflow))
+ .assertThat(repository.queryTaskInstance(workflowInstanceId))
.satisfiesExactly(
taskInstance -> {
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.KILL);
@@ -110,27 +117,36 @@ public class WorkflowInstanceStopIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ final Integer workflowInstanceId = workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.pollInterval(Duration.ofMillis(100))
.atMost(Duration.ofMinutes(1))
- .untilAsserted(() -> Assertions.assertThat(repository.queryWorkflowInstance(workflow)).hasSize(1));
+ .untilAsserted(() -> {
+ Assertions
+ .assertThat(repository.queryWorkflowInstance(workflowInstanceId).getState())
+ .isEqualTo(WorkflowExecutionStatus.RUNNING_EXECUTION);
+ Assertions
+ .assertThat(repository.queryTaskInstance(workflowInstanceId))
+ .satisfiesExactly(
+ taskInstance -> {
+ assertThat(taskInstance.getState())
+ .isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ });
+ });
- final Integer workflowInstanceId = repository.queryWorkflowInstance(workflow).get(0).getId();
assertThat(workflowOperator.stopWorkflowInstance(workflowInstanceId).isSuccess());
-
await()
.atMost(Duration.ofMinutes(1))
.untilAsserted(() -> {
Assertions
- .assertThat(repository.queryWorkflowInstance(workflow))
- .satisfiesExactly(
+ .assertThat(repository.queryWorkflowInstance(workflowInstanceId))
+ .satisfies(
workflowInstance -> {
assertThat(workflowInstance.getState()).isEqualTo(WorkflowExecutionStatus.STOP);
});
Assertions
- .assertThat(repository.queryTaskInstance(workflow))
+ .assertThat(repository.queryTaskInstance(workflowInstanceId))
.satisfiesExactly(
taskInstance -> {
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.KILL);
@@ -150,29 +166,29 @@ public class WorkflowInstanceStopIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ final Integer workflowInstanceId = workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.atMost(Duration.ofMinutes(1))
- .untilAsserted(() -> Assertions.assertThat(repository.queryWorkflowInstance(workflow)).hasSize(1));
+ .untilAsserted(() -> Assertions
+ .assertThat(repository.queryWorkflowInstance(workflowInstanceId).getState())
+ .isEqualTo(WorkflowExecutionStatus.RUNNING_EXECUTION)
+
+ );
// make sure the task has been dispatched to the executor
ThreadUtils.sleep(2_000);
-
- final Integer workflowInstanceId = repository.queryWorkflowInstance(workflow).get(0).getId();
assertThat(workflowOperator.stopWorkflowInstance(workflowInstanceId).isSuccess());
await()
.atMost(Duration.ofMinutes(1))
.untilAsserted(() -> {
Assertions
- .assertThat(repository.queryWorkflowInstance(workflow))
- .satisfiesExactly(
- workflowInstance -> {
- assertThat(workflowInstance.getState()).isEqualTo(WorkflowExecutionStatus.STOP);
- });
+ .assertThat(repository.queryWorkflowInstance(workflowInstanceId).getState())
+ .isEqualTo(WorkflowExecutionStatus.STOP);
+
Assertions
- .assertThat(repository.queryTaskInstance(workflow))
+ .assertThat(repository.queryTaskInstance(workflowInstanceId))
.hasSize(3)
.anySatisfy(
taskInstance -> {
diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/cases/WorkflowStartIT.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/cases/WorkflowStartIT.java
index bd70599cbd..90c69a62bd 100644
--- a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/cases/WorkflowStartIT.java
+++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/it/cases/WorkflowStartIT.java
@@ -78,7 +78,7 @@ public class WorkflowStartIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.atMost(Duration.ofMinutes(1))
@@ -109,7 +109,7 @@ public class WorkflowStartIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.atMost(Duration.ofMinutes(1))
@@ -154,7 +154,7 @@ public class WorkflowStartIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(runWorkflowCommandParam)
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.atMost(Duration.ofMinutes(1))
@@ -190,7 +190,7 @@ public class WorkflowStartIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.atMost(Duration.ofMinutes(1))
@@ -221,7 +221,7 @@ public class WorkflowStartIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.atMost(Duration.ofMinutes(3))
@@ -274,7 +274,7 @@ public class WorkflowStartIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.atMost(Duration.ofMinutes(1))
@@ -313,7 +313,7 @@ public class WorkflowStartIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.atMost(Duration.ofMinutes(1))
@@ -344,7 +344,7 @@ public class WorkflowStartIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.atMost(Duration.ofMinutes(1))
@@ -382,7 +382,7 @@ public class WorkflowStartIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.atMost(Duration.ofMinutes(1))
@@ -421,7 +421,7 @@ public class WorkflowStartIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(new RunWorkflowCommandParam())
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.atMost(Duration.ofMinutes(1))
@@ -491,7 +491,7 @@ public class WorkflowStartIT extends AbstractMasterIntegrationTest {
.workflowDefinition(workflow)
.runWorkflowCommandParam(runWorkflowCommandParam)
.build();
- workflowOperator.triggerWorkflow(workflowTriggerDTO);
+ workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
await()
.atMost(Duration.ofMinutes(1))
diff --git a/dolphinscheduler-microbench/src/main/java/org/apache/dolphinscheduler/microbench/rpc/RpcBenchMarkTest.java b/dolphinscheduler-microbench/src/main/java/org/apache/dolphinscheduler/microbench/rpc/RpcBenchMarkTest.java
index 496983118f..e9c14face6 100644
--- a/dolphinscheduler-microbench/src/main/java/org/apache/dolphinscheduler/microbench/rpc/RpcBenchMarkTest.java
+++ b/dolphinscheduler-microbench/src/main/java/org/apache/dolphinscheduler/microbench/rpc/RpcBenchMarkTest.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.microbench.rpc;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.base.config.NettyServerConfig;
import org.apache.dolphinscheduler.extract.base.server.SpringServerMethodInvokerDiscovery;
import org.apache.dolphinscheduler.microbench.base.AbstractBaseBenchmark;
@@ -56,8 +56,9 @@ public class RpcBenchMarkTest extends AbstractBaseBenchmark {
springServerMethodInvokerDiscovery = new SpringServerMethodInvokerDiscovery(nettyServerConfig);
springServerMethodInvokerDiscovery.postProcessAfterInitialization(new IServiceImpl(), "iServiceImpl");
springServerMethodInvokerDiscovery.start();
- iService =
- SingletonJdkDynamicRpcClientProxyFactory.getProxyClient("localhost:12345", IService.class);
+ iService = Clients
+ .withService(IService.class)
+ .withHost("localhost:12345");
}
@Benchmark
diff --git a/dolphinscheduler-registry/dolphinscheduler-registry-api/src/main/java/org/apache/dolphinscheduler/registry/api/RegistryClient.java b/dolphinscheduler-registry/dolphinscheduler-registry-api/src/main/java/org/apache/dolphinscheduler/registry/api/RegistryClient.java
index 53ae5f9449..987c75b588 100644
--- a/dolphinscheduler-registry/dolphinscheduler-registry-api/src/main/java/org/apache/dolphinscheduler/registry/api/RegistryClient.java
+++ b/dolphinscheduler-registry/dolphinscheduler-registry-api/src/main/java/org/apache/dolphinscheduler/registry/api/RegistryClient.java
@@ -28,6 +28,8 @@ import org.apache.dolphinscheduler.common.model.WorkerHeartBeat;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
import java.io.IOException;
@@ -39,6 +41,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import lombok.NonNull;
@@ -129,6 +132,15 @@ public class RegistryClient {
return serverList;
}
+ public Optional<Server> getRandomServer(final RegistryNodeType registryNodeType) {
+ final List<Server> serverList = getServerList(registryNodeType);
+ if (CollectionUtils.isEmpty(serverList)) {
+ return Optional.empty();
+ }
+ final Server server = serverList.get(RandomUtils.nextInt(0, serverList.size()));
+ return Optional.ofNullable(server);
+ }
+
/**
* Return server host:port -> value
*/
diff --git a/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/pom.xml b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/pom.xml
index 27e5795f5f..3c6571d34f 100644
--- a/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/pom.xml
+++ b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/pom.xml
@@ -40,6 +40,11 @@
<artifactId>dolphinscheduler-meter</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.dolphinscheduler</groupId>
+ <artifactId>dolphinscheduler-extract-master</artifactId>
+ </dependency>
+
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-quartz</artifactId>
diff --git a/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/ProcessScheduleTask.java b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/ProcessScheduleTask.java
index db75914de5..cc3ac49dd0 100644
--- a/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/ProcessScheduleTask.java
+++ b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/ProcessScheduleTask.java
@@ -17,14 +17,14 @@
package org.apache.dolphinscheduler.scheduler.quartz;
-import org.apache.dolphinscheduler.common.enums.CommandType;
+import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
-import org.apache.dolphinscheduler.common.utils.JSONUtils;
-import org.apache.dolphinscheduler.dao.entity.Command;
+import org.apache.dolphinscheduler.common.enums.TaskDependType;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.Schedule;
import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
-import org.apache.dolphinscheduler.extract.master.command.ScheduleWorkflowCommandParam;
+import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowScheduleTriggerRequest;
import org.apache.dolphinscheduler.service.command.CommandService;
import org.apache.dolphinscheduler.service.process.ProcessService;
@@ -50,6 +50,9 @@ public class ProcessScheduleTask extends QuartzJobBean {
@Autowired
private CommandService commandService;
+ @Autowired
+ private IWorkflowControlClient workflowInstanceController;
+
@Counted(value = "ds.master.quartz.job.executed")
@Timed(value = "ds.master.quartz.job.execution.time", percentiles = {0.5, 0.75, 0.95, 0.99}, histogram = true)
@Override
@@ -85,28 +88,24 @@ public class ProcessScheduleTask extends QuartzJobBean {
return;
}
- Command command = new Command();
- command.setCommandType(CommandType.SCHEDULER);
- command.setExecutorId(schedule.getUserId());
- command.setFailureStrategy(schedule.getFailureStrategy());
- command.setProcessDefinitionCode(schedule.getProcessDefinitionCode());
- command.setScheduleTime(scheduledFireTime);
- command.setStartTime(fireTime);
- command.setWarningGroupId(schedule.getWarningGroupId());
- String workerGroup = WorkerGroupUtils.getWorkerGroupOrDefault(schedule.getWorkerGroup());
- command.setWorkerGroup(workerGroup);
- command.setTenantCode(schedule.getTenantCode());
- command.setEnvironmentCode(schedule.getEnvironmentCode());
- command.setWarningType(schedule.getWarningType());
- command.setProcessInstancePriority(schedule.getProcessInstancePriority());
- command.setProcessDefinitionVersion(processDefinition.getVersion());
-
- final ScheduleWorkflowCommandParam scheduleWorkflowCommandParam = ScheduleWorkflowCommandParam.builder()
- .timeZone(schedule.getTimezoneId())
+ final WorkflowScheduleTriggerRequest scheduleTriggerRequest = WorkflowScheduleTriggerRequest.builder()
+ .userId(schedule.getUserId())
+ .scheduleTIme(scheduledFireTime)
+ .timezoneId(schedule.getTimezoneId())
+ .workflowCode(processDefinition.getCode())
+ .workflowVersion(processDefinition.getVersion())
+ .failureStrategy(schedule.getFailureStrategy())
+ .taskDependType(TaskDependType.TASK_POST)
+ .warningType(schedule.getWarningType())
+ .warningGroupId(schedule.getWarningGroupId())
+ .workflowInstancePriority(schedule.getProcessInstancePriority())
+ .workerGroup(WorkerGroupUtils.getWorkerGroupOrDefault(schedule.getWorkerGroup()))
+ .tenantCode(schedule.getTenantCode())
+ .environmentCode(schedule.getEnvironmentCode())
+ .dryRun(Flag.NO)
+ .testFlag(Flag.NO)
.build();
- command.setCommandParam(JSONUtils.toJsonString(scheduleWorkflowCommandParam));
-
- commandService.createCommand(command);
+ workflowInstanceController.scheduleTriggerWorkflow(scheduleTriggerRequest);
}
private void deleteJob(JobExecutionContext context, int projectId, int scheduleId) {
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java
index 84f1609acc..ba9a214df9 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java
@@ -105,7 +105,7 @@ import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.dao.utils.DqRuleUtils;
import org.apache.dolphinscheduler.dao.utils.EnvironmentUtils;
import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.common.ILogService;
import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
@@ -261,9 +261,6 @@ public class ProcessServiceImpl implements ProcessService {
@Autowired
private CommandService commandService;
- @Autowired
- private TriggerRelationService triggerRelationService;
-
/**
* find process instance detail by id
*
@@ -362,9 +359,10 @@ public class ProcessServiceImpl implements ProcessService {
if (StringUtils.isEmpty(taskInstance.getHost()) || StringUtils.isEmpty(taskLogPath)) {
continue;
}
- ILogService iLogService =
- SingletonJdkDynamicRpcClientProxyFactory.getProxyClient(taskInstance.getHost(), ILogService.class);
- iLogService.removeTaskInstanceLog(taskLogPath);
+ Clients
+ .withService(ILogService.class)
+ .withHost(taskInstance.getHost())
+ .removeTaskInstanceLog(taskLogPath);
}
}
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/TriggerRelationService.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/TriggerRelationService.java
deleted file mode 100644
index 6de2506afd..0000000000
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/TriggerRelationService.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * 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.dolphinscheduler.service.process;
-
-import org.apache.dolphinscheduler.common.enums.ApiTriggerType;
-import org.apache.dolphinscheduler.dao.entity.TriggerRelation;
-
-import java.util.List;
-
-import org.springframework.stereotype.Component;
-
-/**
- * Trigger relation operator to db，because operator command process instance
- */
-@Component
-public interface TriggerRelationService {
-
- void saveTriggerToDb(ApiTriggerType type, Long triggerCode, Integer jobId);
-
- List<TriggerRelation> queryByTypeAndJobId(ApiTriggerType apiTriggerType, int jobId);
-
- int saveCommandTrigger(Integer commandId, Integer processInstanceId);
-
- int saveProcessInstanceTrigger(Integer commandId, Integer processInstanceId);
-}
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/TriggerRelationServiceImpl.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/TriggerRelationServiceImpl.java
deleted file mode 100644
index 0344ea87ea..0000000000
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/TriggerRelationServiceImpl.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * 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.dolphinscheduler.service.process;
-
-import org.apache.dolphinscheduler.common.enums.ApiTriggerType;
-import org.apache.dolphinscheduler.dao.entity.TriggerRelation;
-import org.apache.dolphinscheduler.dao.mapper.TriggerRelationMapper;
-
-import org.apache.commons.collections4.CollectionUtils;
-
-import java.util.Date;
-import java.util.List;
-
-import lombok.extern.slf4j.Slf4j;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-/**
- * Trigger relation operator to db
- */
-@Slf4j
-@Component
-public class TriggerRelationServiceImpl implements TriggerRelationService {
-
- @Autowired
- private TriggerRelationMapper triggerRelationMapper;
-
- @Override
- public void saveTriggerToDb(ApiTriggerType type, Long triggerCode, Integer jobId) {
- TriggerRelation triggerRelation = new TriggerRelation();
- triggerRelation.setTriggerType(type.getCode());
- triggerRelation.setJobId(jobId);
- triggerRelation.setTriggerCode(triggerCode);
- triggerRelation.setCreateTime(new Date());
- triggerRelation.setUpdateTime(new Date());
- triggerRelationMapper.upsert(triggerRelation);
- }
-
- @Override
- public List<TriggerRelation> queryByTypeAndJobId(ApiTriggerType apiTriggerType, int jobId) {
- return triggerRelationMapper.queryByTypeAndJobId(apiTriggerType.getCode(), jobId);
- }
-
- @Override
- public int saveCommandTrigger(Integer commandId, Integer processInstanceId) {
- List<TriggerRelation> existTriggers = queryByTypeAndJobId(ApiTriggerType.PROCESS, processInstanceId);
- if (CollectionUtils.isEmpty(existTriggers)) {
- return 0;
- }
- existTriggers.forEach(triggerRelation -> saveTriggerToDb(ApiTriggerType.COMMAND,
- triggerRelation.getTriggerCode(), commandId));
- int triggerRelationSize = existTriggers.size();
- if (triggerRelationSize > 1) {
- // Fix https://github.com/apache/dolphinscheduler/issues/15864
- // This case shouldn't happen
- log.error("The PROCESS TriggerRelation of command: {} is more than one", commandId);
- }
- return existTriggers.size();
- }
-
- @Override
- public int saveProcessInstanceTrigger(Integer commandId, Integer processInstanceId) {
- List<TriggerRelation> existTriggers = queryByTypeAndJobId(ApiTriggerType.COMMAND, commandId);
- if (CollectionUtils.isEmpty(existTriggers)) {
- return 0;
- }
- existTriggers.forEach(triggerRelation -> saveTriggerToDb(ApiTriggerType.PROCESS,
- triggerRelation.getTriggerCode(), processInstanceId));
- int triggerRelationSize = existTriggers.size();
- if (triggerRelationSize > 1) {
- // Fix https://github.com/apache/dolphinscheduler/issues/15864
- // This case shouldn't happen
- log.error("The COMMAND TriggerRelation of command: {} is more than one", commandId);
- }
- return existTriggers.size();
- }
-
-}
diff --git a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
index 28084f54ad..267d9f18cc 100644
--- a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
+++ b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
@@ -167,9 +167,6 @@ public class ProcessServiceTest {
@Mock
TaskPluginManager taskPluginManager;
- @Mock
- private TriggerRelationService triggerRelationService;
-
@Test
public void testGetUserById() {
User user = new User();
diff --git a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/TriggerRelationServiceTest.java b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/TriggerRelationServiceTest.java
deleted file mode 100644
index 4f3bbb0bc7..0000000000
--- a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/TriggerRelationServiceTest.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * 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.dolphinscheduler.service.process;
-
-import static com.google.common.truth.Truth.assertThat;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.when;
-
-import org.apache.dolphinscheduler.common.enums.ApiTriggerType;
-import org.apache.dolphinscheduler.dao.entity.TriggerRelation;
-import org.apache.dolphinscheduler.dao.mapper.TriggerRelationMapper;
-
-import java.util.Date;
-
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.InjectMocks;
-import org.mockito.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
-import org.mockito.junit.jupiter.MockitoSettings;
-import org.mockito.quality.Strictness;
-
-import com.google.common.collect.Lists;
-
-/**
- * Trigger Relation Service Test
- */
-@ExtendWith(MockitoExtension.class)
-@MockitoSettings(strictness = Strictness.LENIENT)
-public class TriggerRelationServiceTest {
-
- @InjectMocks
- private TriggerRelationServiceImpl triggerRelationService;
- @Mock
- private TriggerRelationMapper triggerRelationMapper;
-
- @Test
- public void saveTriggerToDb() {
- doNothing().when(triggerRelationMapper).upsert(any());
- triggerRelationService.saveTriggerToDb(ApiTriggerType.COMMAND, 1234567890L, 100);
- }
-
- @Test
- public void queryByTypeAndJobId() {
- doNothing().when(triggerRelationMapper).upsert(any());
- when(triggerRelationMapper.queryByTypeAndJobId(ApiTriggerType.PROCESS.getCode(), 100))
- .thenReturn(Lists.newArrayList(getTriggerTdoDb()));
-
- assertThat(triggerRelationService.queryByTypeAndJobId(ApiTriggerType.PROCESS, 100)).hasSize(1);
- assertThat(triggerRelationService.queryByTypeAndJobId(ApiTriggerType.PROCESS, 200)).isEmpty();
- }
-
- @Test
- public void saveCommandTrigger() {
- doNothing().when(triggerRelationMapper).upsert(any());
- when(triggerRelationMapper.queryByTypeAndJobId(ApiTriggerType.PROCESS.getCode(), 100))
- .thenReturn(Lists.newArrayList(getTriggerTdoDb()));
- assertThat(triggerRelationService.saveCommandTrigger(1234567890, 100)).isAtLeast(1);
- assertThat(triggerRelationService.saveCommandTrigger(1234567890, 200)).isEqualTo(0);
-
- }
-
- @Test
- public void saveProcessInstanceTrigger() {
- doNothing().when(triggerRelationMapper).upsert(any());
- when(triggerRelationMapper.queryByTypeAndJobId(ApiTriggerType.COMMAND.getCode(), 100))
- .thenReturn(Lists.newArrayList(getTriggerTdoDb()));
- assertThat(triggerRelationService.saveProcessInstanceTrigger(100, 1234567890)).isAtLeast(1);
- assertThat(triggerRelationService.saveProcessInstanceTrigger(200, 1234567890)).isEqualTo(0);
- }
-
- private TriggerRelation getTriggerTdoDb() {
- TriggerRelation triggerRelation = new TriggerRelation();
- triggerRelation.setTriggerType(ApiTriggerType.PROCESS.getCode());
- triggerRelation.setJobId(100);
- triggerRelation.setTriggerCode(1234567890L);
- triggerRelation.setCreateTime(new Date());
- triggerRelation.setUpdateTime(new Date());
- return triggerRelation;
- }
-}
diff --git a/dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag-startup-param.tsx b/dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag-startup-param.tsx
index 0887f2324f..67649705a3 100644
--- a/dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag-startup-param.tsx
+++ b/dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag-startup-param.tsx
@@ -117,10 +117,9 @@ export default defineComponent({
<span class={styles.tab}>
{t('project.workflow.complement_range')}:
</span>
- {this.commandParam && this.commandParam.complementStartDate ? (
+ {this.commandParam && this.commandParam.backfillTimeList ? (
<span class={styles.content}>
- {this.commandParam.complementStartDate}-
- {this.commandParam.complementEndDate}
+ {this.commandParam.backfillTimeList.join(',')}
</span>
) : (
'-'
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionDispatchEventSender.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionDispatchEventSender.java
index 18ddbdcec7..a6fe00e267 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionDispatchEventSender.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionDispatchEventSender.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.worker.message;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.transportor.ITaskExecutionEvent;
import org.apache.dolphinscheduler.extract.master.transportor.TaskExecutionDispatchEvent;
@@ -32,7 +32,7 @@ public class TaskExecutionDispatchEventSender
@Override
public void sendEvent(TaskExecutionDispatchEvent taskExecutionDispatchEvent) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskExecutionEventListener.class)
.withHost(taskExecutionDispatchEvent.getWorkflowInstanceHost())
.onTaskInstanceDispatched(taskExecutionDispatchEvent);
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionFailedEventSender.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionFailedEventSender.java
index a19443530e..68a49c2bfc 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionFailedEventSender.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionFailedEventSender.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.worker.message;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.transportor.ITaskExecutionEvent;
import org.apache.dolphinscheduler.extract.master.transportor.TaskExecutionFailedEvent;
@@ -32,7 +32,7 @@ public class TaskExecutionFailedEventSender
@Override
public void sendEvent(TaskExecutionFailedEvent message) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskExecutionEventListener.class)
.withHost(message.getWorkflowInstanceHost())
.onTaskInstanceExecutionFailed(message);
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionKilledEventSender.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionKilledEventSender.java
index 7aba8f7913..7bd5568cb2 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionKilledEventSender.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionKilledEventSender.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.worker.message;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.transportor.ITaskExecutionEvent;
import org.apache.dolphinscheduler.extract.master.transportor.TaskExecutionKilledEvent;
@@ -32,7 +32,7 @@ public class TaskExecutionKilledEventSender
@Override
public void sendEvent(TaskExecutionKilledEvent message) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskExecutionEventListener.class)
.withHost(message.getWorkflowInstanceHost())
.onTaskInstanceExecutionKilled(message);
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionPausedEventSender.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionPausedEventSender.java
index 2a416f0473..ed5ecacc39 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionPausedEventSender.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionPausedEventSender.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.worker.message;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.transportor.ITaskExecutionEvent;
import org.apache.dolphinscheduler.extract.master.transportor.TaskExecutionPausedEvent;
@@ -32,7 +32,7 @@ public class TaskExecutionPausedEventSender
@Override
public void sendEvent(TaskExecutionPausedEvent taskExecutionPausedEvent) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskExecutionEventListener.class)
.withHost(taskExecutionPausedEvent.getWorkflowInstanceHost())
.onTaskInstanceExecutionPaused(taskExecutionPausedEvent);
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionSuccessEventSender.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionSuccessEventSender.java
index f1bdd0f775..b07a0539b7 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionSuccessEventSender.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecutionSuccessEventSender.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.worker.message;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.transportor.ITaskExecutionEvent;
import org.apache.dolphinscheduler.extract.master.transportor.TaskExecutionSuccessEvent;
@@ -32,7 +32,7 @@ public class TaskExecutionSuccessEventSender
@Override
public void sendEvent(TaskExecutionSuccessEvent message) {
- SingletonJdkDynamicRpcClientProxyFactory
+ Clients
.withService(ITaskExecutionEventListener.class)
.withHost(message.getWorkflowInstanceHost())
.onTaskInstanceExecutionSuccess(message);
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskInstanceExecutionRunningEventSender.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskInstanceExecutionRunningEventSender.java
index 884eea3db3..9ee907a3b2 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskInstanceExecutionRunningEventSender.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskInstanceExecutionRunningEventSender.java
@@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.worker.message;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.ITaskExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.transportor.ITaskExecutionEvent;
import org.apache.dolphinscheduler.extract.master.transportor.TaskExecutionRunningEvent;
@@ -33,12 +33,10 @@ public class TaskInstanceExecutionRunningEventSender
TaskInstanceExecutionEventSender<TaskExecutionRunningEvent> {
@Override
- public void sendEvent(TaskExecutionRunningEvent taskInstanceExecutionRunningEvent) {
- ITaskExecutionEventListener iTaskExecutionEventListener =
- SingletonJdkDynamicRpcClientProxyFactory
- .getProxyClient(taskInstanceExecutionRunningEvent.getWorkflowInstanceHost(),
- ITaskExecutionEventListener.class);
- iTaskExecutionEventListener.onTaskInstanceExecutionRunning(taskInstanceExecutionRunningEvent);
+ public void sendEvent(final TaskExecutionRunningEvent taskInstanceExecutionRunningEvent) {
+ Clients.withService(ITaskExecutionEventListener.class)
+ .withHost(taskInstanceExecutionRunningEvent.getWorkflowInstanceHost())
+ .onTaskInstanceExecutionRunning(taskInstanceExecutionRunningEvent);
}
@Override
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutor.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutor.java
index 05d6b87286..131a3d14fb 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutor.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutor.java
@@ -29,7 +29,7 @@ import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.extract.alert.IAlertOperator;
import org.apache.dolphinscheduler.extract.alert.request.AlertSendRequest;
import org.apache.dolphinscheduler.extract.alert.request.AlertSendResponse;
-import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.base.utils.Host;
import org.apache.dolphinscheduler.extract.master.transportor.ITaskExecutionEvent;
import org.apache.dolphinscheduler.plugin.datasource.api.utils.CommonUtils;
@@ -265,9 +265,10 @@ public abstract class WorkerTaskExecutor implements Runnable {
task.getExitStatus() == TaskExecutionStatus.SUCCESS ? WarningType.SUCCESS.getCode()
: WarningType.FAILURE.getCode());
try {
- IAlertOperator alertOperator = SingletonJdkDynamicRpcClientProxyFactory
- .getProxyClient(alertServerAddress.getAddress(), IAlertOperator.class);
- AlertSendResponse alertSendResponse = alertOperator.sendAlert(alertSendRequest);
+ final AlertSendResponse alertSendResponse = Clients
+ .withService(IAlertOperator.class)
+ .withHost(alertServerAddress.getAddress())
+ .sendAlert(alertSendRequest);
log.info("Send alert to: {} successfully, response: {}", alertServerAddress, alertSendResponse);
} catch (Exception e) {
log.error("Send alert: {} to: {} failed", alertSendRequest, alertServerAddress, e);