Merge branch 'apache:dev' into dev
This commit is contained in:
FalconSL
GitHub
commit
00fbaca7d8
|
|
@ -28,7 +28,6 @@ export SPRING_DATASOURCE_PASSWORD=123456
|
|||
# DolphinScheduler server related configuration
|
||||
export SPRING_CACHE_TYPE=${SPRING_CACHE_TYPE:-none}
|
||||
export SPRING_JACKSON_TIME_ZONE=${SPRING_JACKSON_TIME_ZONE:-UTC}
|
||||
export MASTER_FETCH_COMMAND_NUM=${MASTER_FETCH_COMMAND_NUM:-10}
|
||||
|
||||
# Registry center configuration, determines the type and link of the registry center
|
||||
export REGISTRY_TYPE=${REGISTRY_TYPE:-jdbc}
|
||||
|
|
|
|||
|
|
@ -28,7 +28,6 @@ export SPRING_DATASOURCE_PASSWORD=123456
|
|||
# DolphinScheduler server related configuration
|
||||
export SPRING_CACHE_TYPE=${SPRING_CACHE_TYPE:-none}
|
||||
export SPRING_JACKSON_TIME_ZONE=${SPRING_JACKSON_TIME_ZONE:-UTC}
|
||||
export MASTER_FETCH_COMMAND_NUM=${MASTER_FETCH_COMMAND_NUM:-10}
|
||||
|
||||
# Registry center configuration, determines the type and link of the registry center
|
||||
export REGISTRY_TYPE=${REGISTRY_TYPE:-zookeeper}
|
||||
|
|
|
|||
|
|
@ -28,7 +28,6 @@ export SPRING_DATASOURCE_PASSWORD=postgres
|
|||
# DolphinScheduler server related configuration
|
||||
export SPRING_CACHE_TYPE=${SPRING_CACHE_TYPE:-none}
|
||||
export SPRING_JACKSON_TIME_ZONE=${SPRING_JACKSON_TIME_ZONE:-UTC}
|
||||
export MASTER_FETCH_COMMAND_NUM=${MASTER_FETCH_COMMAND_NUM:-10}
|
||||
|
||||
# Registry center configuration, determines the type and link of the registry center
|
||||
export REGISTRY_TYPE=jdbc
|
||||
|
|
|
|||
|
|
@ -28,7 +28,6 @@ export SPRING_DATASOURCE_PASSWORD=postgres
|
|||
# DolphinScheduler server related configuration
|
||||
export SPRING_CACHE_TYPE=${SPRING_CACHE_TYPE:-none}
|
||||
export SPRING_JACKSON_TIME_ZONE=${SPRING_JACKSON_TIME_ZONE:-UTC}
|
||||
export MASTER_FETCH_COMMAND_NUM=${MASTER_FETCH_COMMAND_NUM:-10}
|
||||
|
||||
# Registry center configuration, determines the type and link of the registry center
|
||||
export REGISTRY_TYPE=${REGISTRY_TYPE:-zookeeper}
|
||||
|
|
|
|||
|
|
@ -286,7 +286,6 @@ Location: `master-server/conf/application.yaml`
|
|||
| Parameters | Default value | Description |
|
||||
|-----------------------------------------------------------------------------|---------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
|
||||
| master.listen-port | 5678 | master listen port |
|
||||
| master.fetch-command-num | 10 | the number of commands fetched by master |
|
||||
| master.pre-exec-threads | 10 | master prepare execute thread number to limit handle commands in parallel |
|
||||
| master.exec-threads | 100 | master execute thread number to limit process instances in parallel |
|
||||
| master.dispatch-task-number | 3 | master dispatch task number per batch |
|
||||
|
|
@ -305,6 +304,9 @@ Location: `master-server/conf/application.yaml`
|
|||
| master.registry-disconnect-strategy.strategy | stop | Used when the master disconnect from registry, default value: stop. Optional values include stop, waiting |
|
||||
| master.registry-disconnect-strategy.max-waiting-time | 100s | Used when the master disconnect from registry, and the disconnect strategy is waiting, this config means the master will waiting to reconnect to registry in given times, and after the waiting times, if the master still cannot connect to registry, will stop itself, if the value is 0s, the Master will wait infinitely |
|
||||
| master.worker-group-refresh-interval | 10s | The interval to refresh worker group from db to memory |
|
||||
| master.command-fetch-strategy.type | ID_SLOT_BASED | The command fetch strategy, only support `ID_SLOT_BASED` |
|
||||
| master.command-fetch-strategy.config.id-step | 1 | The id auto incremental step of t_ds_command in db |
|
||||
| master.command-fetch-strategy.config.fetch-size | 10 | The number of commands fetched by master |
|
||||
|
||||
### Worker Server related configuration
|
||||
|
||||
|
|
|
|||
|
|
@ -123,7 +123,6 @@ export SPRING_DATASOURCE_PASSWORD={password}
|
|||
# DolphinScheduler server related configuration
|
||||
export SPRING_CACHE_TYPE=${SPRING_CACHE_TYPE:-none}
|
||||
export SPRING_JACKSON_TIME_ZONE=${SPRING_JACKSON_TIME_ZONE:-UTC}
|
||||
export MASTER_FETCH_COMMAND_NUM=${MASTER_FETCH_COMMAND_NUM:-10}
|
||||
|
||||
# Registry center configuration, determines the type and link of the registry center
|
||||
export REGISTRY_TYPE=${REGISTRY_TYPE:-zookeeper}
|
||||
|
|
|
|||
|
|
@ -281,29 +281,30 @@ common.properties配置文件目前主要是配置hadoop/s3/yarn/applicationId
|
|||
|
||||
位置：`master-server/conf/application.yaml`
|
||||
|
||||
| 参数 | 默认值 | 描述 |
|
||||
|-----------------------------------------------------------------------------|--------------|-----------------------------------------------------------------------------------------|
|
||||
| master.listen-port | 5678 | master监听端口 |
|
||||
| master.fetch-command-num | 10 | master拉取command数量 |
|
||||
| master.pre-exec-threads | 10 | master准备执行任务的数量，用于限制并行的command |
|
||||
| master.exec-threads | 100 | master工作线程数量,用于限制并行的流程实例数量 |
|
||||
| master.dispatch-task-number | 3 | master每个批次的派发任务数量 |
|
||||
| master.host-selector | lower_weight | master host选择器,用于选择合适的worker执行任务,可选值: random, round_robin, lower_weight |
|
||||
| master.max-heartbeat-interval | 10s | master最大心跳间隔 |
|
||||
| master.task-commit-retry-times | 5 | 任务重试次数 |
|
||||
| master.task-commit-interval | 1000 | 任务提交间隔,单位为毫秒 |
|
||||
| master.state-wheel-interval | 5 | 轮询检查状态时间 |
|
||||
| master.server-load-protection.enabled | true | 是否开启系统保护策略 |
|
||||
| master.server-load-protection.max-system-cpu-usage-percentage-thresholds | 0.7 | master最大系统cpu使用值,只有当前系统cpu使用值低于最大系统cpu使用值,master服务才能调度任务. 默认值为0.7: 会使用70%的操作系统CPU |
|
||||
| master.server-load-protection.max-jvm-cpu-usage-percentage-thresholds | 0.7 | master最大JVM cpu使用值,只有当前JVM cpu使用值低于最大JVM cpu使用值,master服务才能调度任务. 默认值为0.7: 会使用70%的JVM CPU |
|
||||
| master.server-load-protection.max-system-memory-usage-percentage-thresholds | 0.7 | master最大系统 内存使用值,只有当前系统内存使用值低于最大系统内存使用值,master服务才能调度任务. 默认值为0.7: 会使用70%的操作系统内存 |
|
||||
| master.server-load-protection.max-disk-usage-percentage-thresholds | 0.7 | master最大系统磁盘使用值,只有当前系统磁盘使用值低于最大系统磁盘使用值,master服务才能调度任务. 默认值为0.7: 会使用70%的操作系统磁盘空间 |
|
||||
| master.failover-interval | 10 | failover间隔，单位为分钟 |
|
||||
| master.kill-application-when-task-failover | true | 当任务实例failover时，是否kill掉yarn或k8s application |
|
||||
| master.registry-disconnect-strategy.strategy | stop | 当Master与注册中心失联之后采取的策略, 默认值是: stop. 可选值包括： stop, waiting |
|
||||
| master.registry-disconnect-strategy.max-waiting-time | 100s | 当Master与注册中心失联之后重连时间, 之后当strategy为waiting时，该值生效。 该值表示当Master与注册中心失联时会在给定时间之内进行重连, |
|
||||
| 在给定时间之内重连失败将会停止自己，在重连时，Master会丢弃目前正在执行的工作流，值为0表示会无限期等待 |
|
||||
| master.master.worker-group-refresh-interval | 10s | 定期将workerGroup从数据库中同步到内存的时间间隔 |
|
||||
| 参数 | 默认值 | 描述 |
|
||||
|-----------------------------------------------------------------------------|---------------|------------------------------------------------------------------------------------------------------------------------------------------|
|
||||
| master.listen-port | 5678 | master监听端口 |
|
||||
| master.pre-exec-threads | 10 | master准备执行任务的数量，用于限制并行的command |
|
||||
| master.exec-threads | 100 | master工作线程数量,用于限制并行的流程实例数量 |
|
||||
| master.dispatch-task-number | 3 | master每个批次的派发任务数量 |
|
||||
| master.host-selector | lower_weight | master host选择器,用于选择合适的worker执行任务,可选值: random, round_robin, lower_weight |
|
||||
| master.max-heartbeat-interval | 10s | master最大心跳间隔 |
|
||||
| master.task-commit-retry-times | 5 | 任务重试次数 |
|
||||
| master.task-commit-interval | 1000 | 任务提交间隔,单位为毫秒 |
|
||||
| master.state-wheel-interval | 5 | 轮询检查状态时间 |
|
||||
| master.server-load-protection.enabled | true | 是否开启系统保护策略 |
|
||||
| master.server-load-protection.max-system-cpu-usage-percentage-thresholds | 0.7 | master最大系统cpu使用值,只有当前系统cpu使用值低于最大系统cpu使用值,master服务才能调度任务. 默认值为0.7: 会使用70%的操作系统CPU |
|
||||
| master.server-load-protection.max-jvm-cpu-usage-percentage-thresholds | 0.7 | master最大JVM cpu使用值,只有当前JVM cpu使用值低于最大JVM cpu使用值,master服务才能调度任务. 默认值为0.7: 会使用70%的JVM CPU |
|
||||
| master.server-load-protection.max-system-memory-usage-percentage-thresholds | 0.7 | master最大系统 内存使用值,只有当前系统内存使用值低于最大系统内存使用值,master服务才能调度任务. 默认值为0.7: 会使用70%的操作系统内存 |
|
||||
| master.server-load-protection.max-disk-usage-percentage-thresholds | 0.7 | master最大系统磁盘使用值,只有当前系统磁盘使用值低于最大系统磁盘使用值,master服务才能调度任务. 默认值为0.7: 会使用70%的操作系统磁盘空间 |
|
||||
| master.failover-interval | 10 | failover间隔，单位为分钟 |
|
||||
| master.kill-application-when-task-failover | true | 当任务实例failover时，是否kill掉yarn或k8s application |
|
||||
| master.registry-disconnect-strategy.strategy | stop | 当Master与注册中心失联之后采取的策略, 默认值是: stop. 可选值包括： stop, waiting |
|
||||
| master.registry-disconnect-strategy.max-waiting-time | 100s | 当Master与注册中心失联之后重连时间, 之后当strategy为waiting时，该值生效。 该值表示当Master与注册中心失联时会在给定时间之内进行重连, 在给定时间之内重连失败将会停止自己，在重连时，Master会丢弃目前正在执行的工作流，值为0表示会无限期等待 |
|
||||
| master.master.worker-group-refresh-interval | 10s | 定期将workerGroup从数据库中同步到内存的时间间隔 |
|
||||
| master.command-fetch-strategy.type | ID_SLOT_BASED | Command拉取策略, 目前仅支持 `ID_SLOT_BASED` |
|
||||
| master.command-fetch-strategy.config.id-step | 1 | 数据库中t_ds_command的id自增步长 |
|
||||
| master.command-fetch-strategy.config.fetch-size | 10 | master拉取command数量 |
|
||||
|
||||
## Worker Server相关配置
|
||||
|
||||
|
|
|
|||
|
|
@ -118,7 +118,6 @@ export SPRING_DATASOURCE_PASSWORD={password}
|
|||
# DolphinScheduler server related configuration
|
||||
export SPRING_CACHE_TYPE=${SPRING_CACHE_TYPE:-none}
|
||||
export SPRING_JACKSON_TIME_ZONE=${SPRING_JACKSON_TIME_ZONE:-UTC}
|
||||
export MASTER_FETCH_COMMAND_NUM=${MASTER_FETCH_COMMAND_NUM:-10}
|
||||
|
||||
# Registry center configuration, determines the type and link of the registry center
|
||||
export REGISTRY_TYPE=${REGISTRY_TYPE:-zookeeper}
|
||||
|
|
|
|||
|
|
@ -52,6 +52,7 @@
|
|||
<directory>${basedir}/../../dolphinscheduler-common/src/main/resources</directory>
|
||||
<includes>
|
||||
<include>**/*.properties</include>
|
||||
<include>**/*.yaml</include>
|
||||
</includes>
|
||||
<outputDirectory>conf</outputDirectory>
|
||||
</fileSet>
|
||||
|
|
|
|||
|
|
@ -53,6 +53,7 @@
|
|||
<directory>${basedir}/../dolphinscheduler-common/src/main/resources</directory>
|
||||
<includes>
|
||||
<include>**/*.properties</include>
|
||||
<include>**/*.yaml</include>
|
||||
</includes>
|
||||
<outputDirectory>conf</outputDirectory>
|
||||
</fileSet>
|
||||
|
|
|
|||
|
|
@ -26,6 +26,7 @@ import org.apache.dolphinscheduler.common.constants.Constants;
|
|||
import org.apache.dolphinscheduler.dao.entity.Project;
|
||||
import org.apache.dolphinscheduler.dao.entity.ProjectWorkerGroup;
|
||||
import org.apache.dolphinscheduler.dao.entity.User;
|
||||
import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
|
||||
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
|
||||
import org.apache.dolphinscheduler.dao.mapper.ProjectWorkerGroupMapper;
|
||||
import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
|
||||
|
|
@ -38,6 +39,7 @@ import org.apache.commons.lang3.StringUtils;
|
|||
|
||||
import java.util.Date;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
|
|
@ -113,23 +115,25 @@ public class ProjectWorkerGroupRelationServiceImpl extends BaseServiceImpl
|
|||
}
|
||||
|
||||
Set<String> workerGroupNames =
|
||||
workerGroupMapper.queryAllWorkerGroup().stream().map(item -> item.getName()).collect(
|
||||
workerGroupMapper.queryAllWorkerGroup().stream().map(WorkerGroup::getName).collect(
|
||||
Collectors.toSet());
|
||||
|
||||
workerGroupNames.add(Constants.DEFAULT_WORKER_GROUP);
|
||||
|
||||
Set<String> assignedWorkerGroupNames = workerGroups.stream().collect(Collectors.toSet());
|
||||
Set<String> assignedWorkerGroupNames = new HashSet<>(workerGroups);
|
||||
|
||||
Set<String> difference = SetUtils.difference(assignedWorkerGroupNames, workerGroupNames);
|
||||
|
||||
if (difference.size() > 0) {
|
||||
if (!difference.isEmpty()) {
|
||||
putMsg(result, Status.WORKER_GROUP_NOT_EXIST, difference.toString());
|
||||
return result;
|
||||
}
|
||||
|
||||
Set<String> projectWorkerGroupNames = projectWorkerGroupMapper.selectList(new QueryWrapper<ProjectWorkerGroup>()
|
||||
.lambda()
|
||||
.eq(ProjectWorkerGroup::getProjectCode, projectCode)).stream().map(item -> item.getWorkerGroup())
|
||||
.eq(ProjectWorkerGroup::getProjectCode, projectCode))
|
||||
.stream()
|
||||
.map(ProjectWorkerGroup::getWorkerGroup)
|
||||
.collect(Collectors.toSet());
|
||||
|
||||
difference = SetUtils.difference(projectWorkerGroupNames, assignedWorkerGroupNames);
|
||||
|
|
|
|||
|
|
@ -60,28 +60,65 @@ public class ProjectPreferenceServiceTest {
|
|||
public void testUpdateProjectPreference() {
|
||||
User loginUser = getGeneralUser();
|
||||
|
||||
// no permission
|
||||
Mockito.when(projectService.hasProjectAndWritePerm(Mockito.any(), Mockito.any(), Mockito.any(Result.class)))
|
||||
.thenReturn(false);
|
||||
Result result = projectPreferenceService.updateProjectPreference(loginUser, projectCode, "value");
|
||||
Assertions.assertNull(result.getCode());
|
||||
Assertions.assertNull(result.getData());
|
||||
Assertions.assertNull(result.getMsg());
|
||||
|
||||
// when preference exists in project
|
||||
Mockito.when(projectPreferenceMapper.selectOne(Mockito.any())).thenReturn(null);
|
||||
Mockito.when(projectMapper.queryByCode(projectCode)).thenReturn(getProject(projectCode));
|
||||
|
||||
// success
|
||||
Mockito.when(projectService.hasProjectAndWritePerm(Mockito.any(), Mockito.any(), Mockito.any(Result.class)))
|
||||
.thenReturn(true);
|
||||
|
||||
Mockito.when(projectPreferenceMapper.selectOne(Mockito.any())).thenReturn(null);
|
||||
Mockito.when(projectPreferenceMapper.insert(Mockito.any())).thenReturn(1);
|
||||
|
||||
Result result = projectPreferenceService.updateProjectPreference(loginUser, projectCode, "value");
|
||||
result = projectPreferenceService.updateProjectPreference(loginUser, projectCode, "value");
|
||||
Assertions.assertEquals(Status.SUCCESS.getCode(), result.getCode());
|
||||
|
||||
// database operatation fail
|
||||
Mockito.when(projectPreferenceMapper.insert(Mockito.any())).thenReturn(-1);
|
||||
result = projectPreferenceService.updateProjectPreference(loginUser, projectCode, "value");
|
||||
Assertions.assertEquals(Status.CREATE_PROJECT_PREFERENCE_ERROR.getCode(), result.getCode());
|
||||
|
||||
// when preference exists in project
|
||||
Mockito.when(projectPreferenceMapper.selectOne(Mockito.any())).thenReturn(getProjectPreference());
|
||||
|
||||
// success
|
||||
Mockito.when(projectPreferenceMapper.updateById(Mockito.any())).thenReturn(1);
|
||||
result = projectPreferenceService.updateProjectPreference(loginUser, projectCode, "value");
|
||||
Assertions.assertEquals(Status.SUCCESS.getCode(), result.getCode());
|
||||
|
||||
// database operation fail
|
||||
Mockito.when(projectPreferenceMapper.updateById(Mockito.any())).thenReturn(-1);
|
||||
result = projectPreferenceService.updateProjectPreference(loginUser, projectCode, "value");
|
||||
Assertions.assertEquals(Status.UPDATE_PROJECT_PREFERENCE_ERROR.getCode(), result.getCode());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testQueryProjectPreferenceByProjectCode() {
|
||||
User loginUser = getGeneralUser();
|
||||
|
||||
// no permission
|
||||
Mockito.when(projectService.hasProjectAndWritePerm(Mockito.any(), Mockito.any(), Mockito.any(Result.class)))
|
||||
.thenReturn(false);
|
||||
Result result = projectPreferenceService.queryProjectPreferenceByProjectCode(loginUser, projectCode);
|
||||
Assertions.assertNull(result.getCode());
|
||||
Assertions.assertNull(result.getData());
|
||||
Assertions.assertNull(result.getMsg());
|
||||
|
||||
// PROJECT_PARAMETER_NOT_EXISTS
|
||||
Mockito.when(projectMapper.queryByCode(projectCode)).thenReturn(getProject(projectCode));
|
||||
Mockito.when(projectService.hasProjectAndPerm(Mockito.any(), Mockito.any(), Mockito.any(Result.class),
|
||||
Mockito.any())).thenReturn(true);
|
||||
|
||||
Mockito.when(projectPreferenceMapper.selectOne(Mockito.any())).thenReturn(null);
|
||||
Result result = projectPreferenceService.queryProjectPreferenceByProjectCode(loginUser, projectCode);
|
||||
result = projectPreferenceService.queryProjectPreferenceByProjectCode(loginUser, projectCode);
|
||||
Assertions.assertEquals(Status.SUCCESS.getCode(), result.getCode());
|
||||
|
||||
// SUCCESS
|
||||
|
|
@ -94,14 +131,29 @@ public class ProjectPreferenceServiceTest {
|
|||
public void testEnableProjectPreference() {
|
||||
User loginUser = getGeneralUser();
|
||||
|
||||
// no permission
|
||||
Mockito.when(projectService.hasProjectAndWritePerm(Mockito.any(), Mockito.any(), Mockito.any(Result.class)))
|
||||
.thenReturn(false);
|
||||
Result result = projectPreferenceService.enableProjectPreference(loginUser, projectCode, 1);
|
||||
Assertions.assertNull(result.getCode());
|
||||
Assertions.assertNull(result.getData());
|
||||
Assertions.assertNull(result.getMsg());
|
||||
|
||||
Mockito.when(projectMapper.queryByCode(projectCode)).thenReturn(getProject(projectCode));
|
||||
Mockito.when(projectService.hasProjectAndWritePerm(Mockito.any(), Mockito.any(), Mockito.any(Result.class)))
|
||||
.thenReturn(true);
|
||||
|
||||
// success
|
||||
Mockito.when(projectPreferenceMapper.selectOne(Mockito.any())).thenReturn(getProjectPreference());
|
||||
Result result = projectPreferenceService.enableProjectPreference(loginUser, projectCode, 1);
|
||||
Mockito.when(projectPreferenceMapper.updateById(Mockito.any())).thenReturn(1);
|
||||
result = projectPreferenceService.enableProjectPreference(loginUser, projectCode, 2);
|
||||
Assertions.assertEquals(Status.SUCCESS.getCode(), result.getCode());
|
||||
|
||||
// db operation fail
|
||||
Mockito.when(projectPreferenceMapper.selectOne(Mockito.any())).thenReturn(getProjectPreference());
|
||||
Mockito.when(projectPreferenceMapper.updateById(Mockito.any())).thenReturn(-1);
|
||||
result = projectPreferenceService.enableProjectPreference(loginUser, projectCode, 2);
|
||||
Assertions.assertEquals(Status.UPDATE_PROJECT_PREFERENCE_STATE_ERROR.getCode(), result.getCode());
|
||||
}
|
||||
|
||||
private User getGeneralUser() {
|
||||
|
|
|
|||
|
|
@ -17,13 +17,17 @@
|
|||
|
||||
package org.apache.dolphinscheduler.api.service;
|
||||
|
||||
import static org.apache.dolphinscheduler.api.utils.ServiceTestUtil.getAdminUser;
|
||||
import static org.apache.dolphinscheduler.api.utils.ServiceTestUtil.getGeneralUser;
|
||||
|
||||
import org.apache.dolphinscheduler.api.AssertionsHelper;
|
||||
import org.apache.dolphinscheduler.api.enums.Status;
|
||||
import org.apache.dolphinscheduler.api.service.impl.ProjectWorkerGroupRelationServiceImpl;
|
||||
import org.apache.dolphinscheduler.api.utils.Result;
|
||||
import org.apache.dolphinscheduler.common.constants.Constants;
|
||||
import org.apache.dolphinscheduler.common.enums.UserType;
|
||||
import org.apache.dolphinscheduler.dao.entity.Project;
|
||||
import org.apache.dolphinscheduler.dao.entity.ProjectWorkerGroup;
|
||||
import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
|
||||
import org.apache.dolphinscheduler.dao.entity.User;
|
||||
import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
|
||||
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
|
||||
|
|
@ -33,6 +37,7 @@ import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
|
|||
import org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
|
|
@ -77,27 +82,87 @@ public class ProjectWorkerGroupRelationServiceTest {
|
|||
|
||||
@Test
|
||||
public void testAssignWorkerGroupsToProject() {
|
||||
User generalUser = getGeneralUser();
|
||||
User loginUser = getAdminUser();
|
||||
|
||||
Mockito.when(projectMapper.queryByCode(projectCode)).thenReturn(null);
|
||||
Result result = projectWorkerGroupRelationService.assignWorkerGroupsToProject(loginUser, projectCode,
|
||||
// no permission
|
||||
Result result = projectWorkerGroupRelationService.assignWorkerGroupsToProject(generalUser, projectCode,
|
||||
getWorkerGroups());
|
||||
Assertions.assertEquals(Status.USER_NO_OPERATION_PERM.getCode(), result.getCode());
|
||||
|
||||
// project code is null
|
||||
result = projectWorkerGroupRelationService.assignWorkerGroupsToProject(loginUser, null,
|
||||
getWorkerGroups());
|
||||
Assertions.assertEquals(Status.PROJECT_NOT_EXIST.getCode(), result.getCode());
|
||||
|
||||
// worker group is empty
|
||||
result = projectWorkerGroupRelationService.assignWorkerGroupsToProject(loginUser, projectCode,
|
||||
Collections.emptyList());
|
||||
Assertions.assertEquals(Status.WORKER_GROUP_TO_PROJECT_IS_EMPTY.getCode(), result.getCode());
|
||||
|
||||
// project not exists
|
||||
Mockito.when(projectMapper.queryByCode(projectCode)).thenReturn(null);
|
||||
result = projectWorkerGroupRelationService.assignWorkerGroupsToProject(loginUser, projectCode,
|
||||
getWorkerGroups());
|
||||
Assertions.assertEquals(Status.PROJECT_NOT_EXIST.getCode(), result.getCode());
|
||||
|
||||
// worker group not exists
|
||||
WorkerGroup workerGroup = new WorkerGroup();
|
||||
workerGroup.setName("test");
|
||||
Mockito.when(projectMapper.queryByCode(Mockito.anyLong())).thenReturn(getProject());
|
||||
Mockito.when(workerGroupMapper.queryAllWorkerGroup()).thenReturn(Lists.newArrayList(workerGroup));
|
||||
Mockito.when(workerGroupMapper.queryAllWorkerGroup()).thenReturn(Collections.singletonList(workerGroup));
|
||||
result = projectWorkerGroupRelationService.assignWorkerGroupsToProject(loginUser, projectCode,
|
||||
getDiffWorkerGroups());
|
||||
Assertions.assertEquals(Status.WORKER_GROUP_NOT_EXIST.getCode(), result.getCode());
|
||||
|
||||
// db insertion fail
|
||||
Mockito.when(workerGroupMapper.queryAllWorkerGroup()).thenReturn(Collections.singletonList(workerGroup));
|
||||
Mockito.when(projectWorkerGroupMapper.insert(Mockito.any())).thenReturn(-1);
|
||||
AssertionsHelper.assertThrowsServiceException(Status.ASSIGN_WORKER_GROUP_TO_PROJECT_ERROR,
|
||||
() -> projectWorkerGroupRelationService.assignWorkerGroupsToProject(loginUser, projectCode,
|
||||
getWorkerGroups()));
|
||||
|
||||
// success
|
||||
Mockito.when(projectWorkerGroupMapper.insert(Mockito.any())).thenReturn(1);
|
||||
|
||||
result = projectWorkerGroupRelationService.assignWorkerGroupsToProject(loginUser, projectCode,
|
||||
getWorkerGroups());
|
||||
Assertions.assertEquals(Status.SUCCESS.getCode(), result.getCode());
|
||||
|
||||
// success when there is diff between current wg and assigned wg
|
||||
Mockito.when(projectWorkerGroupMapper.selectList(Mockito.any()))
|
||||
.thenReturn(Collections.singletonList(getDiffProjectWorkerGroup()));
|
||||
Mockito.when(projectWorkerGroupMapper.delete(Mockito.any())).thenReturn(1);
|
||||
result = projectWorkerGroupRelationService.assignWorkerGroupsToProject(loginUser, projectCode,
|
||||
getWorkerGroups());
|
||||
Assertions.assertEquals(Status.SUCCESS.getCode(), result.getCode());
|
||||
|
||||
// db deletion fail
|
||||
Mockito.when(projectWorkerGroupMapper.delete(Mockito.any())).thenReturn(-1);
|
||||
AssertionsHelper.assertThrowsServiceException(Status.ASSIGN_WORKER_GROUP_TO_PROJECT_ERROR,
|
||||
() -> projectWorkerGroupRelationService.assignWorkerGroupsToProject(loginUser, projectCode,
|
||||
getWorkerGroups()));
|
||||
|
||||
// fail when wg is referenced by task definition
|
||||
Mockito.when(taskDefinitionMapper.queryAllDefinitionList(Mockito.anyLong()))
|
||||
.thenReturn(Collections.singletonList(getTaskDefinitionWithDiffWorkerGroup()));
|
||||
AssertionsHelper.assertThrowsServiceException(Status.USED_WORKER_GROUP_EXISTS,
|
||||
() -> projectWorkerGroupRelationService.assignWorkerGroupsToProject(loginUser, projectCode,
|
||||
getWorkerGroups()));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testQueryWorkerGroupsByProject() {
|
||||
// no permission
|
||||
Mockito.when(projectService.hasProjectAndPerm(Mockito.any(), Mockito.any(), Mockito.anyMap(), Mockito.any()))
|
||||
.thenReturn(false);
|
||||
|
||||
Map<String, Object> result =
|
||||
projectWorkerGroupRelationService.queryWorkerGroupsByProject(getGeneralUser(), projectCode);
|
||||
|
||||
Assertions.assertTrue(result.isEmpty());
|
||||
|
||||
// success
|
||||
Mockito.when(projectService.hasProjectAndPerm(Mockito.any(), Mockito.any(), Mockito.anyMap(), Mockito.any()))
|
||||
.thenReturn(true);
|
||||
|
||||
|
|
@ -113,8 +178,7 @@ public class ProjectWorkerGroupRelationServiceTest {
|
|||
Mockito.when(scheduleMapper.querySchedulerListByProjectName(Mockito.any()))
|
||||
.thenReturn(Lists.newArrayList());
|
||||
|
||||
Map<String, Object> result =
|
||||
projectWorkerGroupRelationService.queryWorkerGroupsByProject(getGeneralUser(), projectCode);
|
||||
result = projectWorkerGroupRelationService.queryWorkerGroupsByProject(getGeneralUser(), projectCode);
|
||||
|
||||
ProjectWorkerGroup[] actualValue =
|
||||
((List<ProjectWorkerGroup>) result.get(Constants.DATA_LIST)).toArray(new ProjectWorkerGroup[0]);
|
||||
|
|
@ -126,20 +190,8 @@ public class ProjectWorkerGroupRelationServiceTest {
|
|||
return Lists.newArrayList("default");
|
||||
}
|
||||
|
||||
private User getGeneralUser() {
|
||||
User loginUser = new User();
|
||||
loginUser.setUserType(UserType.GENERAL_USER);
|
||||
loginUser.setUserName("userName");
|
||||
loginUser.setId(1);
|
||||
return loginUser;
|
||||
}
|
||||
|
||||
private User getAdminUser() {
|
||||
User loginUser = new User();
|
||||
loginUser.setUserType(UserType.ADMIN_USER);
|
||||
loginUser.setUserName("userName");
|
||||
loginUser.setId(1);
|
||||
return loginUser;
|
||||
private List<String> getDiffWorkerGroups() {
|
||||
return Lists.newArrayList("default", "new");
|
||||
}
|
||||
|
||||
private Project getProject() {
|
||||
|
|
@ -158,4 +210,20 @@ public class ProjectWorkerGroupRelationServiceTest {
|
|||
projectWorkerGroup.setWorkerGroup("default");
|
||||
return projectWorkerGroup;
|
||||
}
|
||||
|
||||
private ProjectWorkerGroup getDiffProjectWorkerGroup() {
|
||||
ProjectWorkerGroup projectWorkerGroup = new ProjectWorkerGroup();
|
||||
projectWorkerGroup.setId(2);
|
||||
projectWorkerGroup.setProjectCode(projectCode);
|
||||
projectWorkerGroup.setWorkerGroup("new");
|
||||
return projectWorkerGroup;
|
||||
}
|
||||
|
||||
private TaskDefinition getTaskDefinitionWithDiffWorkerGroup() {
|
||||
TaskDefinition taskDefinition = new TaskDefinition();
|
||||
taskDefinition.setProjectCode(projectCode);
|
||||
taskDefinition.setId(1);
|
||||
taskDefinition.setWorkerGroup("new");
|
||||
return taskDefinition;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -31,12 +31,18 @@ import lombok.extern.slf4j.Slf4j;
|
|||
* This class will get the property by the priority of the following: env > jvm > properties.
|
||||
*/
|
||||
@Slf4j
|
||||
public class ImmutablePriorityPropertyDelegate extends ImmutablePropertyDelegate {
|
||||
public class ImmutablePriorityPropertyDelegate implements IPropertyDelegate {
|
||||
|
||||
private static final Map<String, Optional<ConfigValue<String>>> configValueMap = new ConcurrentHashMap<>();
|
||||
|
||||
public ImmutablePriorityPropertyDelegate(String propertyAbsolutePath) {
|
||||
super(propertyAbsolutePath);
|
||||
private ImmutablePropertyDelegate immutablePropertyDelegate;
|
||||
|
||||
private ImmutableYamlDelegate immutableYamlDelegate;
|
||||
|
||||
public ImmutablePriorityPropertyDelegate(ImmutablePropertyDelegate immutablePropertyDelegate,
|
||||
ImmutableYamlDelegate immutableYamlDelegate) {
|
||||
this.immutablePropertyDelegate = immutablePropertyDelegate;
|
||||
this.immutableYamlDelegate = immutableYamlDelegate;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -56,8 +62,14 @@ public class ImmutablePriorityPropertyDelegate extends ImmutablePropertyDelegate
|
|||
return value;
|
||||
}
|
||||
value = getConfigValueFromProperties(key);
|
||||
if (value.isPresent()) {
|
||||
log.debug("Get config value from properties, key: {} actualKey: {}, value: {}",
|
||||
k, value.get().getActualKey(), value.get().getValue());
|
||||
return value;
|
||||
}
|
||||
value = getConfigValueFromYaml(key);
|
||||
value.ifPresent(
|
||||
stringConfigValue -> log.debug("Get config value from properties, key: {} actualKey: {}, value: {}",
|
||||
stringConfigValue -> log.debug("Get config value from yaml, key: {} actualKey: {}, value: {}",
|
||||
k, stringConfigValue.getActualKey(), stringConfigValue.getValue()));
|
||||
return value;
|
||||
});
|
||||
|
|
@ -76,7 +88,8 @@ public class ImmutablePriorityPropertyDelegate extends ImmutablePropertyDelegate
|
|||
@Override
|
||||
public Set<String> getPropertyKeys() {
|
||||
Set<String> propertyKeys = new HashSet<>();
|
||||
propertyKeys.addAll(super.getPropertyKeys());
|
||||
propertyKeys.addAll(this.immutablePropertyDelegate.getPropertyKeys());
|
||||
propertyKeys.addAll(this.immutableYamlDelegate.getPropertyKeys());
|
||||
propertyKeys.addAll(System.getProperties().stringPropertyNames());
|
||||
propertyKeys.addAll(System.getenv().keySet());
|
||||
return propertyKeys;
|
||||
|
|
@ -104,7 +117,15 @@ public class ImmutablePriorityPropertyDelegate extends ImmutablePropertyDelegate
|
|||
}
|
||||
|
||||
private Optional<ConfigValue<String>> getConfigValueFromProperties(String key) {
|
||||
String value = super.get(key);
|
||||
String value = this.immutablePropertyDelegate.get(key);
|
||||
if (value != null) {
|
||||
return Optional.of(ConfigValue.fromProperties(key, value));
|
||||
}
|
||||
return Optional.empty();
|
||||
}
|
||||
|
||||
private Optional<ConfigValue<String>> getConfigValueFromYaml(String key) {
|
||||
String value = this.immutableYamlDelegate.get(key);
|
||||
if (value != null) {
|
||||
return Optional.of(ConfigValue.fromProperties(key, value));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -49,7 +49,7 @@ public class ImmutablePropertyDelegate implements IPropertyDelegate {
|
|||
} catch (IOException e) {
|
||||
log.error("Load property: {} error, please check if the file exist under classpath",
|
||||
propertyAbsolutePath, e);
|
||||
System.exit(1);
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
printProperties();
|
||||
|
|
|
|||
|
|
@ -0,0 +1,82 @@
|
|||
/*
|
||||
* 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.common.config;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.util.Properties;
|
||||
import java.util.Set;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
|
||||
import org.springframework.core.io.InputStreamResource;
|
||||
|
||||
@Slf4j
|
||||
public class ImmutableYamlDelegate implements IPropertyDelegate {
|
||||
|
||||
private static final String REMOTE_LOGGING_YAML_NAME = "/remote-logging.yaml";
|
||||
|
||||
private final Properties properties;
|
||||
|
||||
public ImmutableYamlDelegate() {
|
||||
this(REMOTE_LOGGING_YAML_NAME);
|
||||
}
|
||||
|
||||
public ImmutableYamlDelegate(String... yamlAbsolutePath) {
|
||||
properties = new Properties();
|
||||
// read from classpath
|
||||
for (String fileName : yamlAbsolutePath) {
|
||||
try (InputStream fis = ImmutableYamlDelegate.class.getResourceAsStream(fileName)) {
|
||||
YamlPropertiesFactoryBean factory = new YamlPropertiesFactoryBean();
|
||||
factory.setResources(new InputStreamResource(fis));
|
||||
factory.afterPropertiesSet();
|
||||
Properties subProperties = factory.getObject();
|
||||
properties.putAll(subProperties);
|
||||
} catch (IOException e) {
|
||||
log.error("Load property: {} error, please check if the file exist under classpath",
|
||||
yamlAbsolutePath, e);
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
printProperties();
|
||||
}
|
||||
|
||||
public ImmutableYamlDelegate(Properties properties) {
|
||||
this.properties = properties;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String get(String key) {
|
||||
return properties.getProperty(key);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String get(String key, String defaultValue) {
|
||||
return properties.getProperty(key, defaultValue);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Set<String> getPropertyKeys() {
|
||||
return properties.stringPropertyNames();
|
||||
}
|
||||
|
||||
private void printProperties() {
|
||||
properties.forEach((k, v) -> log.debug("Get property {} -> {}", k, v));
|
||||
}
|
||||
}
|
||||
|
|
@ -35,6 +35,8 @@ public final class Constants {
|
|||
*/
|
||||
public static final String COMMON_PROPERTIES_PATH = "/common.properties";
|
||||
|
||||
public static final String REMOTE_LOGGING_YAML_PATH = "/remote-logging.yaml";
|
||||
|
||||
public static final String FORMAT_SS = "%s%s";
|
||||
public static final String FORMAT_S_S = "%s/%s";
|
||||
public static final String FORMAT_S_S_COLON = "%s:%s";
|
||||
|
|
@ -683,9 +685,6 @@ public final class Constants {
|
|||
public static final Integer QUERY_ALL_ON_WORKFLOW = 2;
|
||||
public static final Integer QUERY_ALL_ON_TASK = 3;
|
||||
|
||||
/**
|
||||
* remote logging
|
||||
*/
|
||||
public static final String REMOTE_LOGGING_ENABLE = "remote.logging.enable";
|
||||
|
||||
public static final String REMOTE_LOGGING_TARGET = "remote.logging.target";
|
||||
|
|
|
|||
|
|
@ -18,9 +18,11 @@
|
|||
package org.apache.dolphinscheduler.common.utils;
|
||||
|
||||
import static org.apache.dolphinscheduler.common.constants.Constants.COMMON_PROPERTIES_PATH;
|
||||
import static org.apache.dolphinscheduler.common.constants.Constants.REMOTE_LOGGING_YAML_PATH;
|
||||
|
||||
import org.apache.dolphinscheduler.common.config.IPropertyDelegate;
|
||||
import org.apache.dolphinscheduler.common.config.ImmutablePriorityPropertyDelegate;
|
||||
import org.apache.dolphinscheduler.common.config.ImmutablePropertyDelegate;
|
||||
import org.apache.dolphinscheduler.common.config.ImmutableYamlDelegate;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
|
@ -37,8 +39,10 @@ import com.google.common.base.Strings;
|
|||
public class PropertyUtils {
|
||||
|
||||
// todo: add another implementation for zookeeper/etcd/consul/xx
|
||||
private static final IPropertyDelegate propertyDelegate =
|
||||
new ImmutablePriorityPropertyDelegate(COMMON_PROPERTIES_PATH);
|
||||
private final ImmutablePriorityPropertyDelegate propertyDelegate =
|
||||
new ImmutablePriorityPropertyDelegate(
|
||||
new ImmutablePropertyDelegate(COMMON_PROPERTIES_PATH),
|
||||
new ImmutableYamlDelegate(REMOTE_LOGGING_YAML_PATH));
|
||||
|
||||
public static String getString(String key) {
|
||||
return propertyDelegate.get(key.trim());
|
||||
|
|
|
|||
|
|
@ -0,0 +1,61 @@
|
|||
#
|
||||
# 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.
|
||||
#
|
||||
|
||||
remote-logging:
|
||||
# Whether to enable remote logging
|
||||
enable: false
|
||||
# if remote-logging.enable = true, set the target of remote logging
|
||||
target: OSS
|
||||
# if remote-logging.enable = true, set the log base directory
|
||||
base.dir: logs
|
||||
# if remote-logging.enable = true, set the number of threads to send logs to remote storage
|
||||
thread.pool.size: 10
|
||||
# required if you set remote-logging.target=OSS
|
||||
oss:
|
||||
# oss access key id, required if you set remote-logging.target=OSS
|
||||
access.key.id: <access.key.id>
|
||||
# oss access key secret, required if you set remote-logging.target=OSS
|
||||
access.key.secret: <access.key.secret>
|
||||
# oss bucket name, required if you set remote-logging.target=OSS
|
||||
bucket.name: <bucket.name>
|
||||
# oss endpoint, required if you set remote-logging.target=OSS
|
||||
endpoint: <endpoint>
|
||||
# required if you set remote-logging.target=S3
|
||||
s3:
|
||||
# s3 access key id, required if you set remote-logging.target=S3
|
||||
access.key.id: <access.key.id>
|
||||
# s3 access key secret, required if you set remote-logging.target=S3
|
||||
access.key.secret: <access.key.secret>
|
||||
# s3 bucket name, required if you set remote-logging.target=S3
|
||||
bucket.name: <bucket.name>
|
||||
# s3 endpoint, required if you set remote-logging.target=S3
|
||||
endpoint: <endpoint>
|
||||
# s3 region, required if you set remote-logging.target=S3
|
||||
region: <region>
|
||||
google.cloud.storage:
|
||||
# the location of the google cloud credential, required if you set remote-logging.target=GCS
|
||||
credential: /path/to/credential
|
||||
# gcs bucket name, required if you set remote-logging.target=GCS
|
||||
bucket.name: <your-bucket>
|
||||
abs:
|
||||
# abs account name, required if you set resource.storage.type=ABS
|
||||
account.name: <your-account-name>
|
||||
# abs account key, required if you set resource.storage.type=ABS
|
||||
account.key: <your-account-key>
|
||||
# abs container name, required if you set resource.storage.type=ABS
|
||||
container.name: <your-container-name>
|
||||
|
||||
|
|
@ -19,6 +19,7 @@ package org.apache.dolphinscheduler.common.config;
|
|||
|
||||
import static com.github.stefanbirkner.systemlambda.SystemLambda.withEnvironmentVariable;
|
||||
import static org.apache.dolphinscheduler.common.constants.Constants.COMMON_PROPERTIES_PATH;
|
||||
import static org.apache.dolphinscheduler.common.constants.Constants.REMOTE_LOGGING_YAML_PATH;
|
||||
|
||||
import org.junit.jupiter.api.Assertions;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
|
@ -26,7 +27,9 @@ import org.junit.jupiter.api.Test;
|
|||
class ImmutablePriorityPropertyDelegateTest {
|
||||
|
||||
private final ImmutablePriorityPropertyDelegate immutablePriorityPropertyDelegate =
|
||||
new ImmutablePriorityPropertyDelegate(COMMON_PROPERTIES_PATH);
|
||||
new ImmutablePriorityPropertyDelegate(
|
||||
new ImmutablePropertyDelegate(COMMON_PROPERTIES_PATH),
|
||||
new ImmutableYamlDelegate(REMOTE_LOGGING_YAML_PATH));
|
||||
|
||||
@Test
|
||||
void getOverrideFromEnv() throws Exception {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,61 @@
|
|||
#
|
||||
# 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.
|
||||
#
|
||||
|
||||
remote-logging:
|
||||
# Whether to enable remote logging
|
||||
enable: false
|
||||
# if remote-logging.enable = true, set the target of remote logging
|
||||
target: OSS
|
||||
# if remote-logging.enable = true, set the log base directory
|
||||
base.dir: logs
|
||||
# if remote-logging.enable = true, set the number of threads to send logs to remote storage
|
||||
thread.pool.size: 10
|
||||
# required if you set remote-logging.target=OSS
|
||||
oss:
|
||||
# oss access key id, required if you set remote-logging.target=OSS
|
||||
access.key.id: <access.key.id>
|
||||
# oss access key secret, required if you set remote-logging.target=OSS
|
||||
access.key.secret: <access.key.secret>
|
||||
# oss bucket name, required if you set remote-logging.target=OSS
|
||||
bucket.name: <bucket.name>
|
||||
# oss endpoint, required if you set remote-logging.target=OSS
|
||||
endpoint: <endpoint>
|
||||
# required if you set remote-logging.target=S3
|
||||
s3:
|
||||
# s3 access key id, required if you set remote-logging.target=S3
|
||||
access.key.id: <access.key.id>
|
||||
# s3 access key secret, required if you set remote-logging.target=S3
|
||||
access.key.secret: <access.key.secret>
|
||||
# s3 bucket name, required if you set remote-logging.target=S3
|
||||
bucket.name: <bucket.name>
|
||||
# s3 endpoint, required if you set remote-logging.target=S3
|
||||
endpoint: <endpoint>
|
||||
# s3 region, required if you set remote-logging.target=S3
|
||||
region: <region>
|
||||
google.cloud.storage:
|
||||
# the location of the google cloud credential, required if you set remote-logging.target=GCS
|
||||
credential: /path/to/credential
|
||||
# gcs bucket name, required if you set remote-logging.target=GCS
|
||||
bucket.name: <your-bucket>
|
||||
abs:
|
||||
# abs account name, required if you set resource.storage.type=ABS
|
||||
account.name: <your-account-name>
|
||||
# abs account key, required if you set resource.storage.type=ABS
|
||||
account.key: <your-account-key>
|
||||
# abs container name, required if you set resource.storage.type=ABS
|
||||
container.name: <your-container-name>
|
||||
|
||||
|
|
@ -52,14 +52,10 @@ public interface CommandMapper extends BaseMapper<Command> {
|
|||
*/
|
||||
List<Command> queryCommandPage(@Param("limit") int limit, @Param("offset") int offset);
|
||||
|
||||
/**
|
||||
* query command page by slot
|
||||
*
|
||||
* @return command list
|
||||
*/
|
||||
List<Command> queryCommandPageBySlot(@Param("limit") int limit,
|
||||
@Param("masterCount") int masterCount,
|
||||
@Param("thisMasterSlot") int thisMasterSlot);
|
||||
List<Command> queryCommandByIdSlot(@Param("currentSlotIndex") int currentSlotIndex,
|
||||
@Param("totalSlot") int totalSlot,
|
||||
@Param("idStep") int idStep,
|
||||
@Param("fetchNumber") int fetchNum);
|
||||
|
||||
void deleteByWorkflowInstanceIds(@Param("workflowInstanceIds") List<Integer> workflowInstanceIds);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -56,6 +56,11 @@ public abstract class BaseDao<ENTITY, MYBATIS_MAPPER extends BaseMapper<ENTITY>>
|
|||
return mybatisMapper.selectBatchIds(ids);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<ENTITY> queryAll() {
|
||||
return mybatisMapper.selectList(null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<ENTITY> queryByCondition(ENTITY queryCondition) {
|
||||
if (queryCondition == null) {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,39 @@
|
|||
/*
|
||||
* 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.repository;
|
||||
|
||||
import org.apache.dolphinscheduler.dao.entity.Command;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public interface CommandDao extends IDao<Command> {
|
||||
|
||||
/**
|
||||
* Query command by command id and server slot, return the command which match (commandId / step) %s totalSlot = currentSlotIndex
|
||||
*
|
||||
* @param currentSlotIndex current slot index
|
||||
* @param totalSlot total slot number
|
||||
* @param idStep id step in db
|
||||
* @param fetchNum fetch number
|
||||
* @return command list
|
||||
*/
|
||||
List<Command> queryCommandByIdSlot(int currentSlotIndex,
|
||||
int totalSlot,
|
||||
int idStep,
|
||||
int fetchNum);
|
||||
}
|
||||
|
|
@ -41,6 +41,11 @@ public interface IDao<Entity> {
|
|||
*/
|
||||
List<Entity> queryByIds(Collection<? extends Serializable> ids);
|
||||
|
||||
/**
|
||||
* Query all entities.
|
||||
*/
|
||||
List<Entity> queryAll();
|
||||
|
||||
/**
|
||||
* Query the entity by condition.
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -0,0 +1,41 @@
|
|||
/*
|
||||
* 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.repository.impl;
|
||||
|
||||
import org.apache.dolphinscheduler.dao.entity.Command;
|
||||
import org.apache.dolphinscheduler.dao.mapper.CommandMapper;
|
||||
import org.apache.dolphinscheduler.dao.repository.BaseDao;
|
||||
import org.apache.dolphinscheduler.dao.repository.CommandDao;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import org.springframework.stereotype.Repository;
|
||||
|
||||
@Repository
|
||||
public class CommandDaoImpl extends BaseDao<Command, CommandMapper> implements CommandDao {
|
||||
|
||||
public CommandDaoImpl(CommandMapper commandMapper) {
|
||||
super(commandMapper);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<Command> queryCommandByIdSlot(int currentSlotIndex, int totalSlot, int idStep, int fetchNum) {
|
||||
return mybatisMapper.queryCommandByIdSlot(currentSlotIndex, totalSlot, idStep, fetchNum);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -40,12 +40,12 @@
|
|||
limit #{limit} offset #{offset}
|
||||
</select>
|
||||
|
||||
<select id="queryCommandPageBySlot" resultType="org.apache.dolphinscheduler.dao.entity.Command">
|
||||
<select id="queryCommandByIdSlot" resultType="org.apache.dolphinscheduler.dao.entity.Command">
|
||||
select *
|
||||
from t_ds_command
|
||||
where id % #{masterCount} = #{thisMasterSlot}
|
||||
where (id / #{idStep}) % #{totalSlot} = #{currentSlotIndex}
|
||||
order by process_instance_priority, id asc
|
||||
limit #{limit}
|
||||
limit #{fetchNumber}
|
||||
</select>
|
||||
<delete id="deleteByWorkflowInstanceIds" >
|
||||
delete from t_ds_command
|
||||
|
|
|
|||
|
|
@ -187,7 +187,7 @@ public class CommandMapperTest extends BaseDaoTest {
|
|||
Command command = createCommand();
|
||||
Integer id = command.getId();
|
||||
boolean hit = id % masterCount == thisMasterSlot;
|
||||
List<Command> commandList = commandMapper.queryCommandPageBySlot(1, masterCount, thisMasterSlot);
|
||||
List<Command> commandList = commandMapper.queryCommandByIdSlot(thisMasterSlot, masterCount, 1, 1);
|
||||
if (hit) {
|
||||
Assertions.assertEquals(id, commandList.get(0).getId());
|
||||
} else {
|
||||
|
|
@ -201,8 +201,9 @@ public class CommandMapperTest extends BaseDaoTest {
|
|||
|
||||
/**
|
||||
* create command map
|
||||
* @param count map count
|
||||
* @param commandType comman type
|
||||
*
|
||||
* @param count map count
|
||||
* @param commandType comman type
|
||||
* @param processDefinitionCode process definition code
|
||||
* @return command map
|
||||
*/
|
||||
|
|
@ -223,7 +224,8 @@ public class CommandMapperTest extends BaseDaoTest {
|
|||
}
|
||||
|
||||
/**
|
||||
* create process definition
|
||||
* create process definition
|
||||
*
|
||||
* @return process definition
|
||||
*/
|
||||
private ProcessDefinition createProcessDefinition() {
|
||||
|
|
@ -243,6 +245,7 @@ public class CommandMapperTest extends BaseDaoTest {
|
|||
|
||||
/**
|
||||
* create command map
|
||||
*
|
||||
* @param count map count
|
||||
* @return command map
|
||||
*/
|
||||
|
|
@ -258,6 +261,7 @@ public class CommandMapperTest extends BaseDaoTest {
|
|||
|
||||
/**
|
||||
* create command
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
private Command createCommand() {
|
||||
|
|
@ -266,6 +270,7 @@ public class CommandMapperTest extends BaseDaoTest {
|
|||
|
||||
/**
|
||||
* create command
|
||||
*
|
||||
* @return Command
|
||||
*/
|
||||
private Command createCommand(CommandType commandType, long processDefinitionCode) {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,88 @@
|
|||
/*
|
||||
* 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.repository.impl;
|
||||
|
||||
import static com.google.common.truth.Truth.assertThat;
|
||||
|
||||
import org.apache.dolphinscheduler.common.constants.Constants;
|
||||
import org.apache.dolphinscheduler.common.enums.CommandType;
|
||||
import org.apache.dolphinscheduler.common.enums.FailureStrategy;
|
||||
import org.apache.dolphinscheduler.common.enums.Priority;
|
||||
import org.apache.dolphinscheduler.common.enums.TaskDependType;
|
||||
import org.apache.dolphinscheduler.common.enums.WarningType;
|
||||
import org.apache.dolphinscheduler.common.utils.DateUtils;
|
||||
import org.apache.dolphinscheduler.dao.BaseDaoTest;
|
||||
import org.apache.dolphinscheduler.dao.entity.Command;
|
||||
import org.apache.dolphinscheduler.dao.repository.CommandDao;
|
||||
|
||||
import org.apache.commons.lang3.RandomUtils;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
|
||||
class CommandDaoImplTest extends BaseDaoTest {
|
||||
|
||||
@Autowired
|
||||
private CommandDao commandDao;
|
||||
|
||||
@Test
|
||||
void fetchCommandByIdSlot() {
|
||||
int commandSize = RandomUtils.nextInt(1, 1000);
|
||||
for (int i = 0; i < commandSize; i++) {
|
||||
createCommand(CommandType.START_PROCESS, 0);
|
||||
}
|
||||
int totalSlot = RandomUtils.nextInt(1, 10);
|
||||
int currentSlotIndex = RandomUtils.nextInt(0, totalSlot);
|
||||
int fetchSize = RandomUtils.nextInt(10, 100);
|
||||
for (int i = 1; i < 5; i++) {
|
||||
int idStep = i;
|
||||
List<Command> commands = commandDao.queryCommandByIdSlot(currentSlotIndex, totalSlot, idStep, fetchSize);
|
||||
assertThat(commands.size()).isGreaterThan(0);
|
||||
assertThat(commands.size())
|
||||
.isEqualTo(commandDao.queryAll()
|
||||
.stream()
|
||||
.filter(command -> (command.getId() / idStep) % totalSlot == currentSlotIndex)
|
||||
.limit(fetchSize)
|
||||
.count());
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private void createCommand(CommandType commandType, int processDefinitionCode) {
|
||||
Command command = new Command();
|
||||
command.setCommandType(commandType);
|
||||
command.setProcessDefinitionCode(processDefinitionCode);
|
||||
command.setExecutorId(4);
|
||||
command.setCommandParam("test command param");
|
||||
command.setTaskDependType(TaskDependType.TASK_ONLY);
|
||||
command.setFailureStrategy(FailureStrategy.CONTINUE);
|
||||
command.setWarningType(WarningType.ALL);
|
||||
command.setWarningGroupId(1);
|
||||
command.setScheduleTime(DateUtils.stringToDate("2019-12-29 12:10:00"));
|
||||
command.setProcessInstancePriority(Priority.MEDIUM);
|
||||
command.setStartTime(DateUtils.stringToDate("2019-12-29 10:10:00"));
|
||||
command.setUpdateTime(DateUtils.stringToDate("2019-12-29 10:10:00"));
|
||||
command.setWorkerGroup(Constants.DEFAULT_WORKER_GROUP);
|
||||
command.setProcessInstanceId(0);
|
||||
command.setProcessDefinitionVersion(0);
|
||||
commandDao.insert(command);
|
||||
}
|
||||
}
|
||||
|
|
@ -52,6 +52,7 @@
|
|||
<directory>${basedir}/../dolphinscheduler-common/src/main/resources</directory>
|
||||
<includes>
|
||||
<include>**/*.properties</include>
|
||||
<include>**/*.yaml</include>
|
||||
</includes>
|
||||
<outputDirectory>conf</outputDirectory>
|
||||
</fileSet>
|
||||
|
|
|
|||
|
|
@ -0,0 +1,49 @@
|
|||
/*
|
||||
* 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.command;
|
||||
|
||||
import static com.google.common.base.Preconditions.checkNotNull;
|
||||
|
||||
import org.apache.dolphinscheduler.dao.repository.CommandDao;
|
||||
import org.apache.dolphinscheduler.server.master.config.CommandFetchStrategy;
|
||||
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
|
||||
import org.apache.dolphinscheduler.server.master.registry.MasterSlotManager;
|
||||
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
@Configuration
|
||||
public class CommandFetcherConfiguration {
|
||||
|
||||
@Bean
|
||||
public ICommandFetcher commandFetcher(MasterConfig masterConfig,
|
||||
MasterSlotManager masterSlotManager,
|
||||
CommandDao commandDao) {
|
||||
CommandFetchStrategy commandFetchStrategy =
|
||||
checkNotNull(masterConfig.getCommandFetchStrategy(), "command fetch strategy is null");
|
||||
switch (commandFetchStrategy.getType()) {
|
||||
case ID_SLOT_BASED:
|
||||
CommandFetchStrategy.IdSlotBasedFetchConfig idSlotBasedFetchConfig =
|
||||
(CommandFetchStrategy.IdSlotBasedFetchConfig) commandFetchStrategy.getConfig();
|
||||
return new IdSlotBasedCommandFetcher(idSlotBasedFetchConfig, masterSlotManager, commandDao);
|
||||
default:
|
||||
throw new IllegalArgumentException(
|
||||
"unsupported command fetch strategy type: " + commandFetchStrategy.getType());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
/*
|
||||
* 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.command;
|
||||
|
||||
import org.apache.dolphinscheduler.dao.entity.Command;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* The command fetcher used to fetch commands
|
||||
*/
|
||||
public interface ICommandFetcher {
|
||||
|
||||
/**
|
||||
* Fetch commands
|
||||
*
|
||||
* @return command list which need to be handled
|
||||
*/
|
||||
List<Command> fetchCommands();
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,73 @@
|
|||
/*
|
||||
* 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.command;
|
||||
|
||||
import org.apache.dolphinscheduler.dao.entity.Command;
|
||||
import org.apache.dolphinscheduler.dao.repository.CommandDao;
|
||||
import org.apache.dolphinscheduler.server.master.config.CommandFetchStrategy;
|
||||
import org.apache.dolphinscheduler.server.master.metrics.ProcessInstanceMetrics;
|
||||
import org.apache.dolphinscheduler.server.master.registry.MasterSlotManager;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
/**
|
||||
* The command fetcher which is fetch commands by command id and slot.
|
||||
*/
|
||||
@Slf4j
|
||||
public class IdSlotBasedCommandFetcher implements ICommandFetcher {
|
||||
|
||||
private final CommandFetchStrategy.IdSlotBasedFetchConfig idSlotBasedFetchConfig;
|
||||
|
||||
private final CommandDao commandDao;
|
||||
|
||||
private final MasterSlotManager masterSlotManager;
|
||||
|
||||
public IdSlotBasedCommandFetcher(CommandFetchStrategy.IdSlotBasedFetchConfig idSlotBasedFetchConfig,
|
||||
MasterSlotManager masterSlotManager,
|
||||
CommandDao commandDao) {
|
||||
this.idSlotBasedFetchConfig = idSlotBasedFetchConfig;
|
||||
this.masterSlotManager = masterSlotManager;
|
||||
this.commandDao = commandDao;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<Command> fetchCommands() {
|
||||
long scheduleStartTime = System.currentTimeMillis();
|
||||
int currentSlotIndex = masterSlotManager.getSlot();
|
||||
int totalSlot = masterSlotManager.getMasterSize();
|
||||
if (totalSlot <= 0 || currentSlotIndex < 0) {
|
||||
log.warn("Slot is validated, current master slots: {}, the current slot index is {}", totalSlot,
|
||||
currentSlotIndex);
|
||||
return Collections.emptyList();
|
||||
}
|
||||
List<Command> commands = commandDao.queryCommandByIdSlot(
|
||||
currentSlotIndex,
|
||||
totalSlot,
|
||||
idSlotBasedFetchConfig.getIdStep(),
|
||||
idSlotBasedFetchConfig.getFetchSize());
|
||||
long cost = System.currentTimeMillis() - scheduleStartTime;
|
||||
log.info("Fetch commands: {} success, cost: {}ms, totalSlot: {}, currentSlotIndex: {}", commands.size(), cost,
|
||||
totalSlot, currentSlotIndex);
|
||||
ProcessInstanceMetrics.recordCommandQueryTime(cost);
|
||||
return commands;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,63 @@
|
|||
/*
|
||||
* 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.config;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
import org.springframework.validation.Errors;
|
||||
|
||||
@Data
|
||||
public class CommandFetchStrategy {
|
||||
|
||||
private CommandFetchStrategyType type = CommandFetchStrategyType.ID_SLOT_BASED;
|
||||
|
||||
private CommandFetchConfig config = new IdSlotBasedFetchConfig();
|
||||
|
||||
public void validate(Errors errors) {
|
||||
config.validate(errors);
|
||||
}
|
||||
|
||||
public enum CommandFetchStrategyType {
|
||||
ID_SLOT_BASED,
|
||||
;
|
||||
}
|
||||
|
||||
public interface CommandFetchConfig {
|
||||
|
||||
void validate(Errors errors);
|
||||
|
||||
}
|
||||
|
||||
@Data
|
||||
public static class IdSlotBasedFetchConfig implements CommandFetchConfig {
|
||||
|
||||
private int idStep = 1;
|
||||
private int fetchSize = 10;
|
||||
|
||||
@Override
|
||||
public void validate(Errors errors) {
|
||||
if (idStep <= 0) {
|
||||
errors.rejectValue("step", null, "step must be greater than 0");
|
||||
}
|
||||
if (fetchSize <= 0) {
|
||||
errors.rejectValue("fetchSize", null, "fetchSize must be greater than 0");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -48,10 +48,6 @@ public class MasterConfig implements Validator {
|
|||
* The master RPC server listen port.
|
||||
*/
|
||||
private int listenPort = 5678;
|
||||
/**
|
||||
* The max batch size used to fetch command from database.
|
||||
*/
|
||||
private int fetchCommandNum = 10;
|
||||
/**
|
||||
* The thread number used to prepare processInstance. This number shouldn't bigger than fetchCommandNum.
|
||||
*/
|
||||
|
|
@ -98,6 +94,8 @@ public class MasterConfig implements Validator {
|
|||
|
||||
private Duration workerGroupRefreshInterval = Duration.ofSeconds(10L);
|
||||
|
||||
private CommandFetchStrategy commandFetchStrategy = new CommandFetchStrategy();
|
||||
|
||||
// ip:listenPort
|
||||
private String masterAddress;
|
||||
|
||||
|
|
@ -115,9 +113,6 @@ public class MasterConfig implements Validator {
|
|||
if (masterConfig.getListenPort() <= 0) {
|
||||
errors.rejectValue("listen-port", null, "is invalidated");
|
||||
}
|
||||
if (masterConfig.getFetchCommandNum() <= 0) {
|
||||
errors.rejectValue("fetch-command-num", null, "should be a positive value");
|
||||
}
|
||||
if (masterConfig.getPreExecThreads() <= 0) {
|
||||
errors.rejectValue("per-exec-threads", null, "should be a positive value");
|
||||
}
|
||||
|
|
@ -149,6 +144,7 @@ public class MasterConfig implements Validator {
|
|||
if (StringUtils.isEmpty(masterConfig.getMasterAddress())) {
|
||||
masterConfig.setMasterAddress(NetUtils.getAddr(masterConfig.getListenPort()));
|
||||
}
|
||||
commandFetchStrategy.validate(errors);
|
||||
|
||||
masterConfig.setMasterRegistryPath(
|
||||
RegistryNodeType.MASTER.getRegistryPath() + "/" + masterConfig.getMasterAddress());
|
||||
|
|
@ -159,7 +155,6 @@ public class MasterConfig implements Validator {
|
|||
String config =
|
||||
"\n****************************Master Configuration**************************************" +
|
||||
"\n listen-port -> " + listenPort +
|
||||
"\n fetch-command-num -> " + fetchCommandNum +
|
||||
"\n pre-exec-threads -> " + preExecThreads +
|
||||
"\n exec-threads -> " + execThreads +
|
||||
"\n dispatch-task-number -> " + dispatchTaskNumber +
|
||||
|
|
@ -175,6 +170,7 @@ public class MasterConfig implements Validator {
|
|||
"\n master-address -> " + masterAddress +
|
||||
"\n master-registry-path: " + masterRegistryPath +
|
||||
"\n worker-group-refresh-interval: " + workerGroupRefreshInterval +
|
||||
"\n command-fetch-strategy: " + commandFetchStrategy +
|
||||
"\n****************************Master Configuration**************************************";
|
||||
log.info(config);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -26,21 +26,18 @@ import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
|
|||
import org.apache.dolphinscheduler.meter.metrics.MetricsProvider;
|
||||
import org.apache.dolphinscheduler.meter.metrics.SystemMetrics;
|
||||
import org.apache.dolphinscheduler.server.master.cache.ProcessInstanceExecCacheManager;
|
||||
import org.apache.dolphinscheduler.server.master.command.ICommandFetcher;
|
||||
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
|
||||
import org.apache.dolphinscheduler.server.master.config.MasterServerLoadProtection;
|
||||
import org.apache.dolphinscheduler.server.master.event.WorkflowEvent;
|
||||
import org.apache.dolphinscheduler.server.master.event.WorkflowEventQueue;
|
||||
import org.apache.dolphinscheduler.server.master.event.WorkflowEventType;
|
||||
import org.apache.dolphinscheduler.server.master.exception.MasterException;
|
||||
import org.apache.dolphinscheduler.server.master.exception.WorkflowCreateException;
|
||||
import org.apache.dolphinscheduler.server.master.metrics.MasterServerMetrics;
|
||||
import org.apache.dolphinscheduler.server.master.metrics.ProcessInstanceMetrics;
|
||||
import org.apache.dolphinscheduler.server.master.registry.MasterSlotManager;
|
||||
import org.apache.dolphinscheduler.service.command.CommandService;
|
||||
|
||||
import org.apache.commons.collections4.CollectionUtils;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
|
|
@ -56,6 +53,9 @@ import org.springframework.stereotype.Service;
|
|||
@Slf4j
|
||||
public class MasterSchedulerBootstrap extends BaseDaemonThread implements AutoCloseable {
|
||||
|
||||
@Autowired
|
||||
private ICommandFetcher commandFetcher;
|
||||
|
||||
@Autowired
|
||||
private CommandService commandService;
|
||||
|
||||
|
|
@ -74,9 +74,6 @@ public class MasterSchedulerBootstrap extends BaseDaemonThread implements AutoCl
|
|||
@Autowired
|
||||
private WorkflowEventLooper workflowEventLooper;
|
||||
|
||||
@Autowired
|
||||
private MasterSlotManager masterSlotManager;
|
||||
|
||||
@Autowired
|
||||
private MasterTaskExecutorBootstrap masterTaskExecutorBootstrap;
|
||||
|
||||
|
|
@ -125,7 +122,7 @@ public class MasterSchedulerBootstrap extends BaseDaemonThread implements AutoCl
|
|||
Thread.sleep(Constants.SLEEP_TIME_MILLIS);
|
||||
continue;
|
||||
}
|
||||
List<Command> commands = findCommands();
|
||||
List<Command> commands = commandFetcher.fetchCommands();
|
||||
if (CollectionUtils.isEmpty(commands)) {
|
||||
// indicate that no command ,sleep for 1s
|
||||
Thread.sleep(Constants.SLEEP_TIME_MILLIS);
|
||||
|
|
@ -170,29 +167,4 @@ public class MasterSchedulerBootstrap extends BaseDaemonThread implements AutoCl
|
|||
}
|
||||
}
|
||||
|
||||
private List<Command> findCommands() throws MasterException {
|
||||
try {
|
||||
long scheduleStartTime = System.currentTimeMillis();
|
||||
int thisMasterSlot = masterSlotManager.getSlot();
|
||||
int masterCount = masterSlotManager.getMasterSize();
|
||||
if (masterCount <= 0) {
|
||||
log.warn("Master count: {} is invalid, the current slot: {}", masterCount, thisMasterSlot);
|
||||
return Collections.emptyList();
|
||||
}
|
||||
int pageSize = masterConfig.getFetchCommandNum();
|
||||
final List<Command> result =
|
||||
commandService.findCommandPageBySlot(pageSize, masterCount, thisMasterSlot);
|
||||
if (CollectionUtils.isNotEmpty(result)) {
|
||||
long cost = System.currentTimeMillis() - scheduleStartTime;
|
||||
log.info(
|
||||
"Master schedule bootstrap loop command success, fetch command size: {}, cost: {}ms, current slot: {}, total slot size: {}",
|
||||
result.size(), cost, thisMasterSlot, masterCount);
|
||||
ProcessInstanceMetrics.recordCommandQueryTime(cost);
|
||||
}
|
||||
return result;
|
||||
} catch (Exception ex) {
|
||||
throw new MasterException("Master loop command from database error", ex);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -83,8 +83,6 @@ registry:
|
|||
|
||||
master:
|
||||
listen-port: 5678
|
||||
# master fetch command num
|
||||
fetch-command-num: 10
|
||||
# master prepare execute thread number to limit handle commands in parallel
|
||||
pre-exec-threads: 10
|
||||
# master execute thread number to limit process instances in parallel
|
||||
|
|
@ -121,6 +119,13 @@ master:
|
|||
# The max waiting time to reconnect to registry if you set the strategy to waiting
|
||||
max-waiting-time: 100s
|
||||
worker-group-refresh-interval: 10s
|
||||
command-fetch-strategy:
|
||||
type: ID_SLOT_BASED
|
||||
config:
|
||||
# The incremental id step
|
||||
id-step: 1
|
||||
# master fetch command num
|
||||
fetch-size: 10
|
||||
|
||||
server:
|
||||
port: 5679
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@
|
|||
|
||||
package org.apache.dolphinscheduler.server.master.config;
|
||||
|
||||
import static com.google.common.truth.Truth.assertThat;
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
|
|
@ -47,6 +48,17 @@ public class MasterConfigTest {
|
|||
assertEquals(0.77, serverLoadProtection.getMaxJvmCpuUsagePercentageThresholds());
|
||||
assertEquals(0.77, serverLoadProtection.getMaxSystemMemoryUsagePercentageThresholds());
|
||||
assertEquals(0.77, serverLoadProtection.getMaxDiskUsagePercentageThresholds());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void getCommandFetchStrategy() {
|
||||
CommandFetchStrategy commandFetchStrategy = masterConfig.getCommandFetchStrategy();
|
||||
assertThat(commandFetchStrategy.getType())
|
||||
.isEqualTo(CommandFetchStrategy.CommandFetchStrategyType.ID_SLOT_BASED);
|
||||
|
||||
CommandFetchStrategy.IdSlotBasedFetchConfig idSlotBasedFetchConfig =
|
||||
(CommandFetchStrategy.IdSlotBasedFetchConfig) commandFetchStrategy.getConfig();
|
||||
assertThat(idSlotBasedFetchConfig.getIdStep()).isEqualTo(3);
|
||||
assertThat(idSlotBasedFetchConfig.getFetchSize()).isEqualTo(11);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -89,8 +89,6 @@ registry:
|
|||
|
||||
master:
|
||||
listen-port: 5678
|
||||
# master fetch command num
|
||||
fetch-command-num: 10
|
||||
# master prepare execute thread number to limit handle commands in parallel
|
||||
pre-exec-threads: 10
|
||||
# master execute thread number to limit process instances in parallel
|
||||
|
|
@ -127,6 +125,13 @@ master:
|
|||
# The max waiting time to reconnect to registry if you set the strategy to waiting
|
||||
max-waiting-time: 100s
|
||||
worker-group-refresh-interval: 10s
|
||||
command-fetch-strategy:
|
||||
type: ID_SLOT_BASED
|
||||
config:
|
||||
# The incremental id step
|
||||
id-step: 3
|
||||
# master fetch command num
|
||||
fetch-size: 11
|
||||
|
||||
server:
|
||||
port: 5679
|
||||
|
|
|
|||
|
|
@ -22,8 +22,6 @@ import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
|
|||
import org.apache.dolphinscheduler.dao.entity.ProcessInstanceMap;
|
||||
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Command Service
|
||||
*/
|
||||
|
|
@ -44,15 +42,6 @@ public interface CommandService {
|
|||
*/
|
||||
int createCommand(Command command);
|
||||
|
||||
/**
|
||||
* Get command page
|
||||
* @param pageSize page size
|
||||
* @param masterCount master count
|
||||
* @param thisMasterSlot master slot
|
||||
* @return command page
|
||||
*/
|
||||
List<Command> findCommandPageBySlot(int pageSize, int masterCount, int thisMasterSlot);
|
||||
|
||||
/**
|
||||
* check the input command exists in queue list
|
||||
*
|
||||
|
|
|
|||
|
|
@ -57,7 +57,6 @@ import org.springframework.beans.factory.annotation.Autowired;
|
|||
import org.springframework.stereotype.Component;
|
||||
|
||||
import com.fasterxml.jackson.databind.node.ObjectNode;
|
||||
import com.google.common.collect.Lists;
|
||||
import io.micrometer.core.annotation.Counted;
|
||||
|
||||
/**
|
||||
|
|
@ -107,14 +106,6 @@ public class CommandServiceImpl implements CommandService {
|
|||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<Command> findCommandPageBySlot(int pageSize, int masterCount, int thisMasterSlot) {
|
||||
if (masterCount <= 0) {
|
||||
return Lists.newArrayList();
|
||||
}
|
||||
return commandMapper.queryCommandPageBySlot(pageSize, masterCount, thisMasterSlot);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean verifyIsNeedCreateCommand(Command command) {
|
||||
boolean isNeedCreate = true;
|
||||
|
|
|
|||
|
|
@ -214,14 +214,4 @@ class MessageServiceImplTest {
|
|||
Mockito.verify(commandMapper, Mockito.times(1)).insert(command);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFindCommandPageBySlot() {
|
||||
int pageSize = 1;
|
||||
int masterCount = 0;
|
||||
int thisMasterSlot = 2;
|
||||
List<Command> commandList =
|
||||
commandService.findCommandPageBySlot(pageSize, masterCount, thisMasterSlot);
|
||||
Assertions.assertEquals(0, commandList.size());
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -160,8 +160,6 @@ casdoor:
|
|||
|
||||
master:
|
||||
listen-port: 5678
|
||||
# master fetch command num
|
||||
fetch-command-num: 10
|
||||
# master prepare execute thread number to limit handle commands in parallel
|
||||
pre-exec-threads: 10
|
||||
# master execute thread number to limit process instances in parallel
|
||||
|
|
@ -192,6 +190,13 @@ master:
|
|||
# kill yarn/k8s application when failover taskInstance, default true
|
||||
kill-application-when-task-failover: true
|
||||
worker-group-refresh-interval: 10s
|
||||
command-fetch-strategy:
|
||||
type: ID_SLOT_BASED
|
||||
config:
|
||||
# The incremental id step
|
||||
id-step: 1
|
||||
# master fetch command num
|
||||
fetch-size: 10
|
||||
|
||||
worker:
|
||||
# worker listener port
|
||||
|
|
|
|||
|
|
@ -60,7 +60,6 @@ public class ResourceContext {
|
|||
public static class ResourceItem {
|
||||
|
||||
private String resourceAbsolutePathInStorage;
|
||||
private String resourceRelativePath;
|
||||
private String resourceAbsolutePathInLocal;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -65,7 +65,7 @@ public class HiveCliTaskTest {
|
|||
}
|
||||
|
||||
@Test
|
||||
public void hiveCliTaskExecuteSqlFromScript() throws Exception {
|
||||
public void hiveCliTaskExecuteSqlFromScript() {
|
||||
String hiveCliTaskParameters = buildHiveCliTaskExecuteSqlFromScriptParameters();
|
||||
HiveCliTask hiveCliTask = prepareHiveCliTaskForTest(hiveCliTaskParameters);
|
||||
hiveCliTask.init();
|
||||
|
|
@ -78,7 +78,7 @@ public class HiveCliTaskTest {
|
|||
TaskExecutionContext taskExecutionContext = new TaskExecutionContext();
|
||||
taskExecutionContext.setTaskParams(hiveCliTaskParameters);
|
||||
ResourceContext resourceContext = new ResourceContext();
|
||||
resourceContext.addResourceItem(new ResourceContext.ResourceItem("/sql_tasks/hive_task.sql", "123_node.sql",
|
||||
resourceContext.addResourceItem(new ResourceContext.ResourceItem("/sql_tasks/hive_task.sql",
|
||||
"/sql_tasks/hive_task.sql"));
|
||||
taskExecutionContext.setResourceContext(resourceContext);
|
||||
|
||||
|
|
|
|||
|
|
@ -88,6 +88,7 @@ public class JavaTask extends AbstractTask {
|
|||
|
||||
/**
|
||||
* Initializes a Java task
|
||||
*
|
||||
* @return void
|
||||
**/
|
||||
@Override
|
||||
|
|
@ -178,7 +179,8 @@ public class JavaTask extends AbstractTask {
|
|||
**/
|
||||
protected String buildJarCommand() {
|
||||
ResourceContext resourceContext = taskRequest.getResourceContext();
|
||||
String mainJarName = resourceContext.getResourceItem(javaParameters.getMainJar().getResourceName())
|
||||
String mainJarAbsolutePathInLocal = resourceContext
|
||||
.getResourceItem(javaParameters.getMainJar().getResourceName())
|
||||
.getResourceAbsolutePathInLocal();
|
||||
StringBuilder builder = new StringBuilder();
|
||||
builder.append(getJavaCommandPath())
|
||||
|
|
@ -186,7 +188,7 @@ public class JavaTask extends AbstractTask {
|
|||
.append(buildResourcePath()).append(" ")
|
||||
.append("-jar").append(" ")
|
||||
.append(taskRequest.getExecutePath()).append(FOLDER_SEPARATOR)
|
||||
.append(mainJarName).append(" ")
|
||||
.append(mainJarAbsolutePathInLocal).append(" ")
|
||||
.append(javaParameters.getMainArgs().trim()).append(" ")
|
||||
.append(javaParameters.getJvmArgs().trim());
|
||||
return builder.toString();
|
||||
|
|
@ -207,39 +209,6 @@ public class JavaTask extends AbstractTask {
|
|||
return javaParameters;
|
||||
}
|
||||
|
||||
/**
|
||||
* Replaces placeholders such as local variables in source files
|
||||
*
|
||||
* @param rawScript
|
||||
* @return String
|
||||
* @throws StringIndexOutOfBoundsException
|
||||
*/
|
||||
protected static String convertJavaSourceCodePlaceholders(String rawScript) throws StringIndexOutOfBoundsException {
|
||||
int len = "${setShareVar(${".length();
|
||||
|
||||
int scriptStart = 0;
|
||||
while ((scriptStart = rawScript.indexOf("${setShareVar(${", scriptStart)) != -1) {
|
||||
int start = -1;
|
||||
int end = rawScript.indexOf('}', scriptStart + len);
|
||||
String prop = rawScript.substring(scriptStart + len, end);
|
||||
|
||||
start = rawScript.indexOf(',', end);
|
||||
end = rawScript.indexOf(')', start);
|
||||
|
||||
String value = rawScript.substring(start + 1, end);
|
||||
|
||||
start = rawScript.indexOf('}', start) + 1;
|
||||
end = rawScript.length();
|
||||
|
||||
String replaceScript = String.format("print(\"${{setValue({},{})}}\".format(\"%s\",%s))", prop, value);
|
||||
|
||||
rawScript = rawScript.substring(0, scriptStart) + replaceScript + rawScript.substring(start, end);
|
||||
|
||||
scriptStart += replaceScript.length();
|
||||
}
|
||||
return rawScript;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a Java source file when it does not exist
|
||||
*
|
||||
|
|
@ -290,8 +259,6 @@ public class JavaTask extends AbstractTask {
|
|||
for (ResourceInfo info : javaParameters.getResourceFilesList()) {
|
||||
builder.append(JavaConstants.PATH_SEPARATOR);
|
||||
builder
|
||||
.append(taskRequest.getExecutePath())
|
||||
.append(FOLDER_SEPARATOR)
|
||||
.append(resourceContext.getResourceItem(info.getResourceName()).getResourceAbsolutePathInLocal());
|
||||
}
|
||||
return builder.toString();
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@
|
|||
|
||||
package org.apache.dolphinscheduler.plugin.task.java;
|
||||
|
||||
import static com.google.common.truth.Truth.assertThat;
|
||||
import static org.apache.dolphinscheduler.plugin.task.api.enums.DataType.VARCHAR;
|
||||
import static org.apache.dolphinscheduler.plugin.task.api.enums.Direct.IN;
|
||||
import static org.apache.dolphinscheduler.plugin.task.java.JavaConstants.RUN_TYPE_JAR;
|
||||
|
|
@ -34,7 +35,6 @@ import org.apache.dolphinscheduler.plugin.task.java.exception.JavaSourceFileExis
|
|||
import org.apache.dolphinscheduler.plugin.task.java.exception.PublicClassNotFoundException;
|
||||
import org.apache.dolphinscheduler.plugin.task.java.exception.RunTypeNotFoundException;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.lang.reflect.Field;
|
||||
import java.nio.file.Files;
|
||||
|
|
@ -82,10 +82,10 @@ public class JavaTaskTest {
|
|||
**/
|
||||
@Test
|
||||
public void buildJarCommand() {
|
||||
String homeBinPath = JavaConstants.JAVA_HOME_VAR + File.separator + "bin" + File.separator;
|
||||
JavaTask javaTask = runJarType();
|
||||
Assertions.assertEquals(javaTask.buildJarCommand(), homeBinPath
|
||||
+ "java -classpath .:/tmp/dolphinscheduler/test/executepath:/tmp/dolphinscheduler/test/executepath/opt/share/jar/resource2.jar -jar /tmp/dolphinscheduler/test/executepath/opt/share/jar/main.jar -host 127.0.0.1 -port 8080 -xms:50m");
|
||||
assertThat(javaTask.buildJarCommand())
|
||||
.isEqualTo(
|
||||
"${JAVA_HOME}/bin/java -classpath .:/tmp/dolphinscheduler/test/executepath:opt/share/jar/resource2.jar -jar /tmp/dolphinscheduler/test/executepath/opt/share/jar/main.jar -host 127.0.0.1 -port 8080 -xms:50m");
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -101,14 +101,13 @@ public class JavaTaskTest {
|
|||
Assertions.assertEquals("JavaTaskTest", publicClassName);
|
||||
String fileName = javaTask.buildJavaSourceCodeFileFullName(publicClassName);
|
||||
try {
|
||||
String homeBinPath = JavaConstants.JAVA_HOME_VAR + File.separator + "bin" + File.separator;
|
||||
Path path = Paths.get(fileName);
|
||||
if (Files.exists(path)) {
|
||||
Files.delete(path);
|
||||
}
|
||||
Assertions.assertEquals(homeBinPath
|
||||
+ "javac -classpath .:/tmp/dolphinscheduler/test/executepath:/tmp/dolphinscheduler/test/executepath/opt/share/jar/resource2.jar /tmp/dolphinscheduler/test/executepath/JavaTaskTest.java",
|
||||
javaTask.buildJavaCompileCommand(sourceCode));
|
||||
assertThat(javaTask.buildJavaCompileCommand(sourceCode))
|
||||
.isEqualTo(
|
||||
"${JAVA_HOME}/bin/javac -classpath .:/tmp/dolphinscheduler/test/executepath:opt/share/jar/resource2.jar /tmp/dolphinscheduler/test/executepath/JavaTaskTest.java");
|
||||
} finally {
|
||||
Path path = Paths.get(fileName);
|
||||
if (Files.exists(path)) {
|
||||
|
|
@ -121,26 +120,29 @@ public class JavaTaskTest {
|
|||
/**
|
||||
* Construct java to run the command
|
||||
*
|
||||
* @return void
|
||||
* @return void
|
||||
**/
|
||||
@Test
|
||||
public void buildJavaCommand() throws Exception {
|
||||
String wantJavaCommand =
|
||||
"${JAVA_HOME}/bin/javac -classpath .:/tmp/dolphinscheduler/test/executepath:/tmp/dolphinscheduler/test/executepath/opt/share/jar/resource2.jar /tmp/dolphinscheduler/test/executepath/JavaTaskTest.java;${JAVA_HOME}/bin/java -classpath .:/tmp/dolphinscheduler/test/executepath:/tmp/dolphinscheduler/test/executepath/opt/share/jar/resource2.jar JavaTaskTest -host 127.0.0.1 -port 8080 -xms:50m";
|
||||
JavaTask javaTask = runJavaType();
|
||||
String sourceCode = javaTask.buildJavaSourceContent();
|
||||
String publicClassName = javaTask.getPublicClassName(sourceCode);
|
||||
|
||||
Assertions.assertEquals("JavaTaskTest", publicClassName);
|
||||
|
||||
String fileName = javaTask.buildJavaSourceCodeFileFullName(publicClassName);
|
||||
Path path = Paths.get(fileName);
|
||||
if (Files.exists(path)) {
|
||||
Files.delete(path);
|
||||
}
|
||||
Assertions.assertEquals(wantJavaCommand, javaTask.buildJavaCommand());
|
||||
assertThat(javaTask.buildJavaCommand())
|
||||
.isEqualTo(
|
||||
"${JAVA_HOME}/bin/javac -classpath .:/tmp/dolphinscheduler/test/executepath:opt/share/jar/resource2.jar /tmp/dolphinscheduler/test/executepath/JavaTaskTest.java;${JAVA_HOME}/bin/java -classpath .:/tmp/dolphinscheduler/test/executepath:opt/share/jar/resource2.jar JavaTaskTest -host 127.0.0.1 -port 8080 -xms:50m");
|
||||
}
|
||||
|
||||
/**
|
||||
* There is no exception to overwriting the Java source file
|
||||
*
|
||||
* @return void
|
||||
* @throws IOException
|
||||
**/
|
||||
|
|
@ -259,8 +261,8 @@ public class JavaTaskTest {
|
|||
resourceItem2.setResourceAbsolutePathInLocal("opt/share/jar/main.jar");
|
||||
|
||||
ResourceContext.ResourceItem resourceItem3 = new ResourceContext.ResourceItem();
|
||||
resourceItem2.setResourceAbsolutePathInStorage("/JavaTaskTest.java");
|
||||
resourceItem2.setResourceAbsolutePathInLocal("JavaTaskTest.java");
|
||||
resourceItem3.setResourceAbsolutePathInStorage("/JavaTaskTest.java");
|
||||
resourceItem3.setResourceAbsolutePathInLocal("JavaTaskTest.java");
|
||||
|
||||
ResourceContext resourceContext = new ResourceContext();
|
||||
resourceContext.addResourceItem(resourceItem1);
|
||||
|
|
@ -275,7 +277,7 @@ public class JavaTaskTest {
|
|||
/**
|
||||
* The Java task to construct the jar run mode
|
||||
*
|
||||
* @return JavaTask
|
||||
* @return JavaTask
|
||||
**/
|
||||
private JavaTask runJarType() {
|
||||
TaskExecutionContext taskExecutionContext = new TaskExecutionContext();
|
||||
|
|
|
|||
|
|
@ -53,6 +53,7 @@
|
|||
<directory>${basedir}/../dolphinscheduler-common/src/main/resources</directory>
|
||||
<includes>
|
||||
<include>**/*.properties</include>
|
||||
<include>**/*.yaml</include>
|
||||
</includes>
|
||||
<outputDirectory>conf</outputDirectory>
|
||||
</fileSet>
|
||||
|
|
|
|||
|
|
@ -151,7 +151,6 @@ public class TaskExecutionContextUtils {
|
|||
}
|
||||
ResourceContext.ResourceItem resourceItem = ResourceContext.ResourceItem.builder()
|
||||
.resourceAbsolutePathInStorage(resourceAbsolutePathInStorage)
|
||||
.resourceRelativePath(resourceRelativePath)
|
||||
.resourceAbsolutePathInLocal(resourceAbsolutePathInLocal)
|
||||
.build();
|
||||
resourceContext.addResourceItem(resourceItem);
|
||||
|
|
|
|||
Loading…
Reference in New Issue