Merge branch 'dev' into dev_wenjun_removeDuplicateDeclaration

SbloodyS · web-flow · commit 2b8f7c48c5a2 · 2026-02-22T14:43:04.000+08:00
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
@@ -163,6 +163,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
     @Transactional
     public Integer triggerWorkflowDefinition(final WorkflowTriggerRequest triggerRequest) {
         final TriggerWorkflowDTO triggerWorkflowDTO = triggerWorkflowRequestTransformer.transform(triggerRequest);
+        // todo: use validator chain
         triggerWorkflowDTOValidator.validate(triggerWorkflowDTO);
         return executorClient.triggerWorkflowDefinition().execute(triggerWorkflowDTO);
     }
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java
@@ -31,6 +31,7 @@
 import org.apache.dolphinscheduler.api.service.SchedulerService;
 import org.apache.dolphinscheduler.api.utils.PageInfo;
 import org.apache.dolphinscheduler.api.utils.Result;
+import org.apache.dolphinscheduler.api.validator.TenantExistValidator;
 import org.apache.dolphinscheduler.api.vo.ScheduleVO;
 import org.apache.dolphinscheduler.common.constants.Constants;
 import org.apache.dolphinscheduler.common.enums.FailureStrategy;
@@ -43,13 +44,11 @@
 import org.apache.dolphinscheduler.dao.entity.Environment;
 import org.apache.dolphinscheduler.dao.entity.Project;
 import org.apache.dolphinscheduler.dao.entity.Schedule;
-import org.apache.dolphinscheduler.dao.entity.Tenant;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
 import org.apache.dolphinscheduler.dao.mapper.EnvironmentMapper;
 import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
 import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
-import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
 import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionMapper;
 import org.apache.dolphinscheduler.scheduler.api.SchedulerApi;
 import org.apache.dolphinscheduler.service.cron.CronUtils;
@@ -107,7 +106,7 @@ public class SchedulerServiceImpl extends BaseServiceImpl implements SchedulerSe
     private EnvironmentMapper environmentMapper;
 
     @Autowired
-    private TenantMapper tenantMapper;
+    private TenantExistValidator tenantExistValidator;
 
     /**
      * save schedule
@@ -166,7 +165,7 @@ public Map<String, Object> insertSchedule(User loginUser,
         Schedule scheduleObj = new Schedule();
         Date now = new Date();
 
-        checkValidTenant(tenantCode);
+        tenantExistValidator.validate(tenantCode);
 
         scheduleObj.setTenantCode(tenantCode);
         scheduleObj.setProjectName(project.getName());
@@ -276,7 +275,7 @@ public Schedule createSchedulesV2(User loginUser,
                     scheduleExists.getId());
         }
 
-        checkValidTenant(scheduleCreateRequest.getTenantCode());
+        tenantExistValidator.validate(scheduleCreateRequest.getTenantCode());
 
         Schedule schedule = scheduleCreateRequest.convert2Schedule();
         Environment environment = environmentMapper.queryByEnvironmentCode(schedule.getEnvironmentCode());
@@ -759,7 +758,7 @@ private void updateSchedule(Map<String, Object> result, Schedule schedule, Workf
 
         Date now = new Date();
 
-        checkValidTenant(tenantCode);
+        tenantExistValidator.validate(tenantCode);
         schedule.setTenantCode(tenantCode);
 
         // updateWorkflowInstance param
@@ -818,17 +817,4 @@ private void updateSchedule(Map<String, Object> result, Schedule schedule, Workf
         putMsg(result, Status.SUCCESS);
     }
 
-    /**
-     * check valid tenant
-     *
-     * @param tenantCode
-     */
-    private void checkValidTenant(String tenantCode) {
-        if (!Constants.DEFAULT.equals(tenantCode)) {
-            Tenant tenant = tenantMapper.queryByTenantCode(tenantCode);
-            if (tenant == null) {
-                throw new ServiceException(Status.TENANT_NOT_EXIST, tenantCode);
-            }
-        }
-    }
 }
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/TenantExistValidator.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/TenantExistValidator.java
@@ -0,0 +1,46 @@
+/*
+ * 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.api.validator;
+
+import org.apache.dolphinscheduler.dao.repository.TenantDao;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.springframework.stereotype.Component;
+
+/**
+ * This validator is used to validate whether the tenant exists.
+ * <p> If the tenant does not exist, an {@link IllegalArgumentException} will be thrown. </p>
+ */
+@Slf4j
+@Component
+public class TenantExistValidator implements IValidator<String> {
+
+    private final TenantDao tenantDao;
+
+    public TenantExistValidator(TenantDao tenantDao) {
+        this.tenantDao = tenantDao;
+    }
+
+    @Override
+    public void validate(String tenantCode) {
+        if (!tenantDao.queryByCode(tenantCode).isPresent()) {
+            throw new IllegalArgumentException(String.format("Tenant: [%s] not exists", tenantCode));
+        }
+    }
+}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/BackfillWorkflowDTOValidator.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/BackfillWorkflowDTOValidator.java
@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.api.validator.workflow;
 
 import org.apache.dolphinscheduler.api.validator.IValidator;
+import org.apache.dolphinscheduler.api.validator.TenantExistValidator;
 import org.apache.dolphinscheduler.common.enums.CommandType;
 import org.apache.dolphinscheduler.common.enums.ReleaseState;
 
@@ -31,6 +32,12 @@
 @Component
 public class BackfillWorkflowDTOValidator implements IValidator<BackfillWorkflowDTO> {
 
+    private final TenantExistValidator tenantExistValidator;
+
+    public BackfillWorkflowDTOValidator(TenantExistValidator tenantExistValidator) {
+        this.tenantExistValidator = tenantExistValidator;
+    }
+
     @Override
     public void validate(final BackfillWorkflowDTO backfillWorkflowDTO) {
         final BackfillWorkflowDTO.BackfillParamsDTO backfillParams = backfillWorkflowDTO.getBackfillParams();
@@ -52,5 +59,6 @@ public void validate(final BackfillWorkflowDTO backfillWorkflowDTO) {
         if (backfillWorkflowDTO.getWorkflowDefinition().getReleaseState() != ReleaseState.ONLINE) {
             throw new IllegalStateException("The workflowDefinition should be online");
         }
+        tenantExistValidator.validate(backfillWorkflowDTO.getTenantCode());
     }
 }
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTOValidator.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/validator/workflow/TriggerWorkflowDTOValidator.java
@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.api.validator.workflow;
 
 import org.apache.dolphinscheduler.api.validator.IValidator;
+import org.apache.dolphinscheduler.api.validator.TenantExistValidator;
 import org.apache.dolphinscheduler.common.enums.CommandType;
 import org.apache.dolphinscheduler.common.enums.ReleaseState;
 
@@ -29,6 +30,12 @@
 @Component
 public class TriggerWorkflowDTOValidator implements IValidator<TriggerWorkflowDTO> {
 
+    private final TenantExistValidator tenantExistValidator;
+
+    public TriggerWorkflowDTOValidator(TenantExistValidator tenantExistValidator) {
+        this.tenantExistValidator = tenantExistValidator;
+    }
+
     @Override
     public void validate(final TriggerWorkflowDTO triggerWorkflowDTO) {
         if (triggerWorkflowDTO.getExecType() != CommandType.START_PROCESS) {
@@ -40,5 +47,6 @@ public void validate(final TriggerWorkflowDTO triggerWorkflowDTO) {
         if (triggerWorkflowDTO.getWorkflowDefinition().getReleaseState() != ReleaseState.ONLINE) {
             throw new IllegalStateException("The workflowDefinition should be online");
         }
+        tenantExistValidator.validate(triggerWorkflowDTO.getTenantCode());
     }
 }
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/SchedulerServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/SchedulerServiceTest.java
@@ -26,6 +26,7 @@
 import org.apache.dolphinscheduler.api.enums.Status;
 import org.apache.dolphinscheduler.api.exceptions.ServiceException;
 import org.apache.dolphinscheduler.api.service.impl.SchedulerServiceImpl;
+import org.apache.dolphinscheduler.api.validator.TenantExistValidator;
 import org.apache.dolphinscheduler.common.constants.Constants;
 import org.apache.dolphinscheduler.common.enums.ReleaseState;
 import org.apache.dolphinscheduler.dao.entity.Environment;
@@ -36,11 +37,8 @@
 import org.apache.dolphinscheduler.dao.mapper.EnvironmentMapper;
 import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
 import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
-import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
 import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionMapper;
-import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationMapper;
 import org.apache.dolphinscheduler.scheduler.api.SchedulerApi;
-import org.apache.dolphinscheduler.service.process.ProcessService;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
@@ -63,15 +61,6 @@ public class SchedulerServiceTest extends BaseServiceTestTool {
     @InjectMocks
     private SchedulerServiceImpl schedulerService;
 
-    @Mock
-    private WorkflowTaskRelationMapper workflowTaskRelationMapper;
-
-    @Mock
-    private MonitorService monitorService;
-
-    @Mock
-    private ProcessService processService;
-
     @Mock
     private ScheduleMapper scheduleMapper;
 
@@ -94,7 +83,7 @@ public class SchedulerServiceTest extends BaseServiceTestTool {
     private EnvironmentMapper environmentMapper;
 
     @Mock
-    private TenantMapper tenantMapper;
+    private TenantExistValidator tenantExistValidator;
 
     protected static User user;
     protected Exception exception;
@@ -128,6 +117,7 @@ public void testCreateSchedulesV2() {
         scheduleCreateRequest.setWorkflowDefinitionCode(processDefinitionCode);
         scheduleCreateRequest.setEnvironmentCode(environmentCode);
         scheduleCreateRequest.setTenantCode(Constants.DEFAULT);
+        scheduleCreateRequest.setStartTime(startTime);
 
         // error process definition not exists
         exception = Assertions.assertThrows(ServiceException.class,
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/TenantDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/TenantDao.java
@@ -19,6 +19,10 @@
 
 import org.apache.dolphinscheduler.dao.entity.Tenant;
 
+import java.util.Optional;
+
 public interface TenantDao extends IDao<Tenant> {
 
+    Optional<Tenant> queryByCode(String tenantCode);
+
 }
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/TenantDaoImpl.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/TenantDaoImpl.java
@@ -22,6 +22,8 @@
 import org.apache.dolphinscheduler.dao.repository.BaseDao;
 import org.apache.dolphinscheduler.dao.repository.TenantDao;
 
+import java.util.Optional;
+
 import lombok.NonNull;
 
 import org.springframework.stereotype.Repository;
@@ -33,4 +35,8 @@ public TenantDaoImpl(@NonNull TenantMapper tenantMapper) {
         super(tenantMapper);
     }
 
+    @Override
+    public Optional<Tenant> queryByCode(String tenantCode) {
+        return Optional.ofNullable(mybatisMapper.queryByTenantCode(tenantCode));
+    }
 }
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/K8sTaskExecutionContext.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/K8sTaskExecutionContext.java
@@ -19,7 +19,10 @@
 
 import java.io.Serializable;
 
+import lombok.AllArgsConstructor;
+import lombok.Builder;
 import lombok.Data;
+import lombok.NoArgsConstructor;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -28,6 +31,9 @@
  *  k8s Task ExecutionContext
  */
 @Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
 public class K8sTaskExecutionContext implements Serializable {
 
     private String configYaml;
@@ -36,23 +42,11 @@ public class K8sTaskExecutionContext implements Serializable {
 
     private String connectionParams;
 
-    public K8sTaskExecutionContext() {
-    }
-
     @JsonCreator(mode = JsonCreator.Mode.PROPERTIES)
     public K8sTaskExecutionContext(
                                    @JsonProperty("configYaml") String configYaml,
                                    @JsonProperty("namespace") String namespace) {
         this.configYaml = configYaml;
         this.namespace = namespace;
     }
-
-    @Override
-    public String toString() {
-        return "K8sTaskExecutionContext{"
-                + "namespace=" + namespace
-                + ", configYaml='" + configYaml + '\''
-                + ", connectionParams='" + connectionParams + '\''
-                + '}';
-    }
 }
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/AbstractParameters.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/AbstractParameters.java
@@ -18,12 +18,9 @@
 package org.apache.dolphinscheduler.plugin.task.api.parameters;
 
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
-import org.apache.dolphinscheduler.plugin.task.api.K8sTaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
-import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
 import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo;
-import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.DataSourceParameters;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
 import org.apache.dolphinscheduler.plugin.task.api.utils.VarPoolUtils;
 
@@ -72,16 +69,6 @@ public Map<String, Property> getLocalParametersMap() {
         return localParametersMaps;
     }
 
-    public K8sTaskExecutionContext generateK8sTaskExecutionContext(ResourceParametersHelper parametersHelper,
-                                                                   int datasource) {
-        DataSourceParameters dataSourceParameters =
-                (DataSourceParameters) parametersHelper.getResourceParameters(ResourceType.DATASOURCE, datasource);
-        K8sTaskExecutionContext k8sTaskExecutionContext = new K8sTaskExecutionContext();
-        k8sTaskExecutionContext.setConnectionParams(
-                Objects.nonNull(dataSourceParameters) ? dataSourceParameters.getConnectionParams() : null);
-        return k8sTaskExecutionContext;
-    }
-
     /**
      * get input local parameters map if the param direct is IN
      *
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/K8sTaskParameters.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/K8sTaskParameters.java
@@ -17,10 +17,12 @@
 
 package org.apache.dolphinscheduler.plugin.task.api.parameters;
 
+import org.apache.dolphinscheduler.plugin.task.api.K8sTaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType;
 import org.apache.dolphinscheduler.plugin.task.api.model.Label;
 import org.apache.dolphinscheduler.plugin.task.api.model.NodeSelectorExpression;
 import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.DataSourceParameters;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
 
 import org.apache.commons.lang3.StringUtils;
@@ -31,9 +33,6 @@
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 
-/**
- * k8s task parameters
- */
 @Data
 @Slf4j
 public class K8sTaskParameters extends AbstractParameters {
@@ -67,4 +66,21 @@ public ResourceParametersHelper getResources() {
         resources.put(ResourceType.DATASOURCE, datasource);
         return resources;
     }
+
+    public K8sTaskExecutionContext generateK8sTaskExecutionContext(
+                                                                   ResourceParametersHelper parametersHelper,
+                                                                   int datasource) {
+        DataSourceParameters dataSourceParameters =
+                (DataSourceParameters) parametersHelper
+                        .getResourceParameters(ResourceType.DATASOURCE, datasource);
+
+        String connectionParams = null;
+        if (dataSourceParameters != null) {
+            connectionParams = dataSourceParameters.getConnectionParams();
+        }
+
+        return K8sTaskExecutionContext.builder()
+                .connectionParams(connectionParams)
+                .build();
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -163,6 +163,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ`
`163`	`163`	`@Transactional`
`164`	`164`	`public Integer triggerWorkflowDefinition(final WorkflowTriggerRequest triggerRequest) {`
`165`	`165`	`final TriggerWorkflowDTO triggerWorkflowDTO = triggerWorkflowRequestTransformer.transform(triggerRequest);`
	`166`	`+ // todo: use validator chain`
`166`	`167`	`triggerWorkflowDTOValidator.validate(triggerWorkflowDTO);`
`167`	`168`	`return executorClient.triggerWorkflowDefinition().execute(triggerWorkflowDTO);`
`168`	`169`	`}`
Original file line number	Diff line number	Diff line change
`@@ -19,6 +19,10 @@`
`19`	`19`
`20`	`20`	`import org.apache.dolphinscheduler.dao.entity.Tenant;`
`21`	`21`
	`22`	`+import java.util.Optional;`
	`23`	`+`
`22`	`24`	`public interface TenantDao extends IDao<Tenant> {`
`23`	`25`
	`26`	`+ Optional<Tenant> queryByCode(String tenantCode);`
	`27`	`+`
`24`	`28`	`}`
Original file line number	Diff line number	Diff line change
`@@ -22,6 +22,8 @@`
`22`	`22`	`import org.apache.dolphinscheduler.dao.repository.BaseDao;`
`23`	`23`	`import org.apache.dolphinscheduler.dao.repository.TenantDao;`
`24`	`24`
	`25`	`+import java.util.Optional;`
	`26`	`+`
`25`	`27`	`import lombok.NonNull;`
`26`	`28`
`27`	`29`	`import org.springframework.stereotype.Repository;`
`@@ -33,4 +35,8 @@ public TenantDaoImpl(@NonNull TenantMapper tenantMapper) {`
`33`	`35`	`super(tenantMapper);`
`34`	`36`	`}`
`35`	`37`
	`38`	`+ @Override`
	`39`	`+ public Optional<Tenant> queryByCode(String tenantCode) {`
	`40`	`+ return Optional.ofNullable(mybatisMapper.queryByTenantCode(tenantCode));`
	`41`	`+ }`
`36`	`42`	`}`