This is an automated email from the ASF dual-hosted git repository.
wenhemin pushed a commit to branch json_split_two
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/json_split_two by this push:
new d1a61f1 [Fix-5778]: The t_ds_schedules table, process_definition_id
change to process_definition_code (#5928)
d1a61f1 is described below
commit d1a61f16934c0c9423a507cc129d55ff5aace84a
Author: wen-hemin <[email protected]>
AuthorDate: Mon Aug 2 10:32:32 2021 +0800
[Fix-5778]: The t_ds_schedules table, process_definition_id change to
process_definition_code (#5928)
* fix: the t_ds_schedules table, process_definition_id change to
process_definition_code
* fix checkstyle
* fix: recovery code
* fix UT
Co-authored-by: wen-hemin <[email protected]>
---
.../api/service/impl/ExecutorServiceImpl.java | 2 +-
.../service/impl/ProcessDefinitionServiceImpl.java | 8 +-
.../api/service/impl/SchedulerServiceImpl.java | 14 +--
.../api/service/ExecutorServiceTest.java | 12 +--
.../api/service/ProcessDefinitionServiceTest.java | 18 +---
.../api/service/SchedulerServiceTest.java | 12 ++-
.../dolphinscheduler/dao/entity/Schedule.java | 15 +--
.../dao/mapper/ScheduleMapper.java | 32 +++---
.../dao/mapper/ProcessDefinitionMapper.xml | 6 +-
.../dolphinscheduler/dao/mapper/ScheduleMapper.xml | 32 +++---
.../dao/mapper/WorkFlowLineageMapper.xml | 4 +-
.../dao/mapper/ScheduleMapperTest.java | 44 ++++----
.../dao/mapper/WorkFlowLineageMapperTest.java | 7 +-
.../server/master/runner/MasterExecThread.java | 4 +-
.../server/master/MasterExecThreadTest.java | 13 +--
.../service/process/ProcessService.java | 120 ++++-----------------
.../service/quartz/ProcessScheduleJob.java | 6 +-
sql/dolphinscheduler_mysql.sql | 2 +-
sql/dolphinscheduler_postgre.sql | 2 +-
19 files changed, 134 insertions(+), 219 deletions(-)
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
index 1d944d0..8978257 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
@@ -549,7 +549,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl
implements ExecutorServ
command.setCommandParam(JSONUtils.toJsonString(cmdParam));
return processService.createCommand(command);
} else if (runMode == RunMode.RUN_MODE_PARALLEL) {
- List<Schedule> schedules =
processService.queryReleaseSchedulerListByProcessDefinitionId(processDefineId);
+ List<Schedule> schedules =
processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefineId);
// TODO: next pr change to code
List<Date> listDate = new LinkedList<>();
if (!CollectionUtils.isEmpty(schedules)) {
for (Schedule item : schedules) {
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
index b28e899..65e2b10 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
@@ -555,7 +555,7 @@ public class ProcessDefinitionServiceImpl extends
BaseServiceImpl implements Pro
}
// get the timing according to the process definition
- List<Schedule> schedules =
scheduleMapper.queryByProcessDefinitionId(processDefinitionId);
+ List<Schedule> schedules =
scheduleMapper.queryByProcessDefinitionCode(processDefinitionId);
if (!schedules.isEmpty() && schedules.size() > 1) {
logger.warn("scheduler num is {},Greater than 1",
schedules.size());
putMsg(result, Status.DELETE_PROCESS_DEFINE_BY_ID_ERROR);
@@ -630,7 +630,7 @@ public class ProcessDefinitionServiceImpl extends
BaseServiceImpl implements Pro
processDefinition.setReleaseState(releaseState);
processDefinitionMapper.updateById(processDefinition);
List<Schedule> scheduleList =
scheduleMapper.selectAllByProcessDefineArray(
- new int[]{processDefinition.getId()}
+ new long[]{processDefinition.getCode()}
);
for (Schedule schedule : scheduleList) {
@@ -712,7 +712,7 @@ public class ProcessDefinitionServiceImpl extends
BaseServiceImpl implements Pro
* @return DagDataSchedule
*/
public DagDataSchedule exportProcessDagData(ProcessDefinition
processDefinition) {
- List<Schedule> schedules =
scheduleMapper.queryByProcessDefinitionId(processDefinition.getId());
+ List<Schedule> schedules =
scheduleMapper.queryByProcessDefinitionCode(processDefinition.getCode());
DagDataSchedule dagDataSchedule = new
DagDataSchedule(processService.genDagData(processDefinition));
if (!schedules.isEmpty()) {
Schedule schedule = schedules.get(0);
@@ -821,7 +821,7 @@ public class ProcessDefinitionServiceImpl extends
BaseServiceImpl implements Pro
Schedule schedule = dagDataSchedule.getSchedule();
if (null != schedule) {
ProcessDefinition newProcessDefinition =
processDefinitionMapper.queryByCode(processDefinition.getCode());
- schedule.setProcessDefinitionId(newProcessDefinition.getId());
+ schedule.setProcessDefinitionCode(newProcessDefinition.getCode());
schedule.setUserId(loginUser.getId());
schedule.setCreateTime(now);
schedule.setUpdateTime(now);
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
index 350ab9a..edc7074 100644
---
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
@@ -140,7 +140,7 @@ public class SchedulerServiceImpl extends BaseServiceImpl
implements SchedulerSe
Date now = new Date();
scheduleObj.setProjectName(project.getName());
- scheduleObj.setProcessDefinitionId(processDefinition.getId());
+ scheduleObj.setProcessDefinitionCode(processDefineCode);
scheduleObj.setProcessDefinitionName(processDefinition.getName());
ScheduleParam scheduleParam = JSONUtils.parseObject(schedule,
ScheduleParam.class);
@@ -228,9 +228,9 @@ public class SchedulerServiceImpl extends BaseServiceImpl
implements SchedulerSe
return result;
}
- ProcessDefinition processDefinition =
processService.findProcessDefineById(schedule.getProcessDefinitionId());
+ ProcessDefinition processDefinition =
processDefinitionMapper.queryByCode(schedule.getProcessDefinitionCode());
if (processDefinition == null) {
- putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST,
schedule.getProcessDefinitionId());
+ putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST,
schedule.getProcessDefinitionCode());
return result;
}
@@ -326,9 +326,9 @@ public class SchedulerServiceImpl extends BaseServiceImpl
implements SchedulerSe
putMsg(result, Status.SCHEDULE_CRON_REALEASE_NEED_NOT_CHANGE,
scheduleStatus);
return result;
}
- ProcessDefinition processDefinition =
processService.findProcessDefineById(scheduleObj.getProcessDefinitionId());
+ ProcessDefinition processDefinition =
processDefinitionMapper.queryByCode(scheduleObj.getProcessDefinitionCode());
if (processDefinition == null) {
- putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST,
scheduleObj.getProcessDefinitionId());
+ putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST,
scheduleObj.getProcessDefinitionCode());
return result;
}
@@ -342,7 +342,7 @@ public class SchedulerServiceImpl extends BaseServiceImpl
implements SchedulerSe
}
// check sub process definition release state
List<Integer> subProcessDefineIds = new ArrayList<>();
-
processService.recurseFindSubProcessId(scheduleObj.getProcessDefinitionId(),
subProcessDefineIds);
+ processService.recurseFindSubProcessId(processDefinition.getId(),
subProcessDefineIds);
Integer[] idArray = subProcessDefineIds.toArray(new
Integer[subProcessDefineIds.size()]);
if (!subProcessDefineIds.isEmpty()) {
List<ProcessDefinition> subProcessDefinitionList =
@@ -430,7 +430,7 @@ public class SchedulerServiceImpl extends BaseServiceImpl
implements SchedulerSe
}
Page<Schedule> page = new Page<>(pageNo, pageSize);
- IPage<Schedule> scheduleIPage =
scheduleMapper.queryByProcessDefineIdPaging(page, processDefinition.getId(),
+ IPage<Schedule> scheduleIPage =
scheduleMapper.queryByProcessDefineCodePaging(page, processDefineCode,
searchVal);
PageInfo<Schedule> pageInfo = new PageInfo<>(pageNo, pageSize);
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorServiceTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorServiceTest.java
index 624fa6e..2cae051 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorServiceTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorServiceTest.java
@@ -152,7 +152,7 @@ public class ExecutorServiceTest {
@Test
public void testNoComplement() {
-
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(zeroSchedulerList());
Map<String, Object> result =
executorService.execProcessInstance(loginUser, projectCode,
processDefinitionCode, cronTime, CommandType.START_PROCESS,
null, null,
@@ -170,7 +170,7 @@ public class ExecutorServiceTest {
@Test
public void testComplementWithStartNodeList() {
-
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(zeroSchedulerList());
Map<String, Object> result =
executorService.execProcessInstance(loginUser, projectCode,
processDefinitionCode, cronTime, CommandType.START_PROCESS,
null, "n1,n2",
@@ -188,7 +188,7 @@ public class ExecutorServiceTest {
@Test
public void testDateError() {
-
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(zeroSchedulerList());
Map<String, Object> result =
executorService.execProcessInstance(loginUser, projectCode,
processDefinitionCode, "2020-01-31 23:00:00,2020-01-01
00:00:00", CommandType.COMPLEMENT_DATA,
null, null,
@@ -205,7 +205,7 @@ public class ExecutorServiceTest {
@Test
public void testSerial() {
-
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(zeroSchedulerList());
Map<String, Object> result =
executorService.execProcessInstance(loginUser, projectCode,
processDefinitionCode, cronTime, CommandType.COMPLEMENT_DATA,
null, null,
@@ -222,7 +222,7 @@ public class ExecutorServiceTest {
@Test
public void testParallelWithOutSchedule() {
-
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(zeroSchedulerList());
Map<String, Object> result =
executorService.execProcessInstance(loginUser, projectCode,
processDefinitionCode, cronTime, CommandType.COMPLEMENT_DATA,
null, null,
@@ -240,7 +240,7 @@ public class ExecutorServiceTest {
@Test
public void testParallelWithSchedule() {
-
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(oneSchedulerList());
+
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(oneSchedulerList());
Map<String, Object> result =
executorService.execProcessInstance(loginUser, projectCode,
processDefinitionCode, cronTime, CommandType.COMPLEMENT_DATA,
null, null,
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionServiceTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionServiceTest.java
index c25fa59..f6a6a7f 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionServiceTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionServiceTest.java
@@ -18,7 +18,6 @@
package org.apache.dolphinscheduler.api.service;
import static org.powermock.api.mockito.PowerMockito.mock;
-import static org.powermock.api.mockito.PowerMockito.when;
import org.apache.dolphinscheduler.api.enums.Status;
import
org.apache.dolphinscheduler.api.service.impl.ProcessDefinitionServiceImpl;
@@ -34,8 +33,6 @@ import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.graph.DAG;
import org.apache.dolphinscheduler.common.model.TaskNode;
import org.apache.dolphinscheduler.common.process.Property;
-import org.apache.dolphinscheduler.common.utils.DateUtils;
-import org.apache.dolphinscheduler.common.utils.FileUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.dao.entity.DagData;
@@ -55,10 +52,6 @@ import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
import org.apache.dolphinscheduler.service.process.ProcessService;
-import org.apache.http.entity.ContentType;
-
-import java.io.File;
-import java.io.FileInputStream;
import java.io.IOException;
import java.text.MessageFormat;
import java.util.ArrayList;
@@ -70,7 +63,6 @@ import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
-import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServletResponse;
import org.junit.Assert;
@@ -80,8 +72,6 @@ import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
-import org.springframework.mock.web.MockMultipartFile;
-import org.springframework.web.multipart.MultipartFile;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@@ -516,7 +506,7 @@ public class ProcessDefinitionServiceTest {
List<Schedule> schedules = new ArrayList<>();
schedules.add(getSchedule());
schedules.add(getSchedule());
-
Mockito.when(scheduleMapper.queryByProcessDefinitionId(46)).thenReturn(schedules);
+
Mockito.when(scheduleMapper.queryByProcessDefinitionCode(46)).thenReturn(schedules);
Map<String, Object> schedulerGreaterThanOneRes =
processDefinitionService.deleteProcessDefinitionById(loginUser, projectCode,
46);
Assert.assertEquals(Status.DELETE_PROCESS_DEFINE_BY_ID_ERROR,
schedulerGreaterThanOneRes.get(Constants.STATUS));
@@ -527,7 +517,7 @@ public class ProcessDefinitionServiceTest {
schedules.add(schedule);
putMsg(result, Status.SUCCESS, projectCode);
Mockito.when(projectService.checkProjectAndAuth(loginUser, project,
project.getName())).thenReturn(result);
-
Mockito.when(scheduleMapper.queryByProcessDefinitionId(46)).thenReturn(schedules);
+
Mockito.when(scheduleMapper.queryByProcessDefinitionCode(46)).thenReturn(schedules);
Map<String, Object> schedulerOnlineRes =
processDefinitionService.deleteProcessDefinitionById(loginUser, projectCode,
46);
Assert.assertEquals(Status.SCHEDULE_CRON_STATE_ONLINE,
schedulerOnlineRes.get(Constants.STATUS));
@@ -537,7 +527,7 @@ public class ProcessDefinitionServiceTest {
schedules.add(schedule);
putMsg(result, Status.SUCCESS, projectCode);
Mockito.when(projectService.checkProjectAndAuth(loginUser, project,
project.getName())).thenReturn(result);
-
Mockito.when(scheduleMapper.queryByProcessDefinitionId(46)).thenReturn(schedules);
+
Mockito.when(scheduleMapper.queryByProcessDefinitionCode(46)).thenReturn(schedules);
Mockito.when(processDefineMapper.deleteById(46)).thenReturn(0);
Map<String, Object> deleteFail =
processDefinitionService.deleteProcessDefinitionById(loginUser, projectCode,
46);
Assert.assertEquals(Status.DELETE_PROCESS_DEFINE_BY_ID_ERROR,
deleteFail.get(Constants.STATUS));
@@ -950,7 +940,7 @@ public class ProcessDefinitionServiceTest {
Date date = new Date();
Schedule schedule = new Schedule();
schedule.setId(46);
- schedule.setProcessDefinitionId(1);
+ schedule.setProcessDefinitionCode(1);
schedule.setStartTime(date);
schedule.setEndTime(date);
schedule.setCrontab("0 0 5 * * ? *");
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
index 7cd383a..238d1d0 100644
---
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
@@ -27,6 +27,7 @@ import
org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.Schedule;
import org.apache.dolphinscheduler.dao.entity.User;
+import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
import org.apache.dolphinscheduler.service.process.ProcessService;
@@ -71,6 +72,9 @@ public class SchedulerServiceTest {
private ProjectMapper projectMapper;
@Mock
+ private ProcessDefinitionMapper processDefinitionMapper;
+
+ @Mock
private ProjectServiceImpl projectService;
@Mock
@@ -102,7 +106,7 @@ public class SchedulerServiceTest {
Schedule schedule = new Schedule();
schedule.setId(1);
- schedule.setProcessDefinitionId(1);
+ schedule.setProcessDefinitionCode(1);
schedule.setReleaseState(ReleaseState.OFFLINE);
List<Server> masterServers = new ArrayList<>();
@@ -113,7 +117,7 @@ public class SchedulerServiceTest {
Mockito.when(projectMapper.queryByCode(projectCode)).thenReturn(project);
Mockito.when(projectMapper.queryByName(projectName)).thenReturn(project);
-
Mockito.when(processService.findProcessDefineById(1)).thenReturn(processDefinition);
+
Mockito.when(processDefinitionMapper.queryByCode(1)).thenReturn(processDefinition);
//hash no auth
result = schedulerService.setScheduleState(loginUser,
project.getCode(), 1, ReleaseState.ONLINE);
@@ -128,10 +132,10 @@ public class SchedulerServiceTest {
Assert.assertEquals(Status.SCHEDULE_CRON_REALEASE_NEED_NOT_CHANGE,
result.get(Constants.STATUS));
//PROCESS_DEFINE_NOT_EXIST
- schedule.setProcessDefinitionId(2);
+ schedule.setProcessDefinitionCode(2);
result = schedulerService.setScheduleState(loginUser,
project.getCode(), 1, ReleaseState.ONLINE);
Assert.assertEquals(Status.PROCESS_DEFINE_NOT_EXIST,
result.get(Constants.STATUS));
- schedule.setProcessDefinitionId(1);
+ schedule.setProcessDefinitionCode(1);
// PROCESS_DEFINE_NOT_RELEASE
result = schedulerService.setScheduleState(loginUser,
project.getCode(), 1, ReleaseState.ONLINE);
diff --git
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Schedule.java
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Schedule.java
index 74ed5c1..e1b4c90 100644
---
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Schedule.java
+++
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Schedule.java
@@ -39,10 +39,11 @@ public class Schedule {
@TableId(value = "id", type = IdType.AUTO)
private int id;
+
/**
- * process definition id
+ * process definition code
*/
- private int processDefinitionId;
+ private long processDefinitionCode;
/**
* process definition name
@@ -222,12 +223,12 @@ public class Schedule {
this.releaseState = releaseState;
}
- public int getProcessDefinitionId() {
- return processDefinitionId;
+ public long getProcessDefinitionCode() {
+ return processDefinitionCode;
}
- public void setProcessDefinitionId(int processDefinitionId) {
- this.processDefinitionId = processDefinitionId;
+ public void setProcessDefinitionCode(long processDefinitionCode) {
+ this.processDefinitionCode = processDefinitionCode;
}
public String getProcessDefinitionName() {
@@ -290,7 +291,7 @@ public class Schedule {
public String toString() {
return "Schedule{"
+ "id=" + id
- + ", processDefinitionId=" + processDefinitionId
+ + ", processDefinitionCode=" + processDefinitionCode
+ ", processDefinitionName='" + processDefinitionName + '\''
+ ", projectName='" + projectName + '\''
+ ", description='" + definitionDescription + '\''
diff --git
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.java
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.java
index 225677d..37fae5d 100644
---
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.java
+++
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.java
@@ -17,12 +17,14 @@
package org.apache.dolphinscheduler.dao.mapper;
import org.apache.dolphinscheduler.dao.entity.Schedule;
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-import com.baomidou.mybatisplus.core.metadata.IPage;
+
import org.apache.ibatis.annotations.Param;
import java.util.List;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+
/**
* scheduler mapper interface
*/
@@ -31,13 +33,13 @@ public interface ScheduleMapper extends
BaseMapper<Schedule> {
/**
* scheduler page
* @param page page
- * @param processDefinitionId processDefinitionId
+ * @param processDefinitionCode processDefinitionCode
* @param searchVal searchVal
* @return scheduler IPage
*/
- IPage<Schedule> queryByProcessDefineIdPaging(IPage<Schedule> page,
- @Param("processDefinitionId")
int processDefinitionId,
- @Param("searchVal") String
searchVal);
+ IPage<Schedule> queryByProcessDefineCodePaging(IPage<Schedule> page,
+
@Param("processDefinitionCode") long processDefinitionCode,
+ @Param("searchVal") String
searchVal);
/**
* query schedule list by project name
@@ -47,24 +49,24 @@ public interface ScheduleMapper extends
BaseMapper<Schedule> {
List<Schedule> querySchedulerListByProjectName(@Param("projectName")
String projectName);
/**
- * query schedule list by process definition ids
- * @param processDefineIds processDefineIds
+ * query schedule list by process definition codes
+ * @param processDefineCodes processDefineCodes
* @return schedule list
*/
- List<Schedule> selectAllByProcessDefineArray(@Param("processDefineIds")
int[] processDefineIds);
+ List<Schedule> selectAllByProcessDefineArray(@Param("processDefineCodes")
long[] processDefineCodes);
/**
- * query schedule list by process definition id
- * @param processDefinitionId processDefinitionId
+ * query schedule list by process definition code
+ * @param processDefinitionCode processDefinitionCode
* @return schedule list
*/
- List<Schedule> queryByProcessDefinitionId(@Param("processDefinitionId")
int processDefinitionId);
+ List<Schedule>
queryByProcessDefinitionCode(@Param("processDefinitionCode") long
processDefinitionCode);
/**
- * query schedule list by process definition id
- * @param processDefinitionId processDefinitionId
+ * query schedule list by process definition code
+ * @param processDefinitionCode processDefinitionCode
* @return schedule list
*/
- List<Schedule>
queryReleaseSchedulerListByProcessDefinitionId(@Param("processDefinitionId")
int processDefinitionId);
+ List<Schedule>
queryReleaseSchedulerListByProcessDefinitionCode(@Param("processDefinitionCode")
long processDefinitionCode);
}
diff --git
a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapper.xml
b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapper.xml
index 0ede680..590e447 100644
---
a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapper.xml
+++
b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapper.xml
@@ -72,8 +72,8 @@
td.global_params, td.flag, td.warning_group_id, td.timeout,
td.tenant_id, td.update_time, td.create_time,
sc.schedule_release_state, tu.user_name
FROM t_ds_process_definition td
- left join (select process_definition_id,release_state as
schedule_release_state from t_ds_schedules group by
- process_definition_id,release_state) sc on sc.process_definition_id =
td.id
+ left join (select process_definition_code,release_state as
schedule_release_state from t_ds_schedules group by
+ process_definition_code,release_state) sc on
sc.process_definition_code = td.code
left join t_ds_user tu on td.user_id = tu.id
where td.project_code = #{projectCode}
<if test=" searchVal != null and searchVal != ''">
@@ -120,7 +120,7 @@
</if>
group by td.user_id,tu.user_name
</select>
-
+
<select id="queryByDefineId"
resultType="org.apache.dolphinscheduler.dao.entity.ProcessDefinition">
SELECT
pd.id, pd.code, pd.name, pd.version, pd.release_state,
pd.project_code, pd.user_id, pd.description,
diff --git
a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.xml
b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.xml
index 587680f..482d30b 100644
---
a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.xml
+++
b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.xml
@@ -19,32 +19,34 @@
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-mapper.dtd"; >
<mapper namespace="org.apache.dolphinscheduler.dao.mapper.ScheduleMapper">
<sql id="baseSql">
- id, process_definition_id, start_time, end_time, timezone_id, crontab,
failure_strategy, user_id, release_state,
+ id, process_definition_code, start_time, end_time, timezone_id,
crontab, failure_strategy, user_id, release_state,
warning_type, warning_group_id, process_instance_priority,
worker_group, create_time, update_time
</sql>
<sql id="baseSqlV2">
- ${alias}.id, ${alias}.process_definition_id, ${alias}.start_time,
${alias}.end_time, ${alias}.timezone_id, ${alias}.crontab,
${alias}.failure_strategy, ${alias}.user_id, ${alias}.release_state,
- ${alias}.warning_type, ${alias}.warning_group_id,
${alias}.process_instance_priority, ${alias}.worker_group,
${alias}.create_time, ${alias}.update_time
+ ${alias}.id, ${alias}.process_definition_code, ${alias}.start_time,
${alias}.end_time, ${alias}.timezone_id,
+ ${alias}.crontab, ${alias}.failure_strategy, ${alias}.user_id,
${alias}.release_state, ${alias}.warning_type,
+ ${alias}.warning_group_id, ${alias}.process_instance_priority,
${alias}.worker_group, ${alias}.create_time,
+ ${alias}.update_time
</sql>
- <select id="queryByProcessDefineIdPaging"
resultType="org.apache.dolphinscheduler.dao.entity.Schedule">
+ <select id="queryByProcessDefineCodePaging"
resultType="org.apache.dolphinscheduler.dao.entity.Schedule">
select p_f.name as process_definition_name, p.name as
project_name,u.user_name,
<include refid="baseSqlV2">
<property name="alias" value="s"/>
</include>
from t_ds_schedules s
- join t_ds_process_definition p_f on s.process_definition_id = p_f.id
+ join t_ds_process_definition p_f on s.process_definition_code =
p_f.code
join t_ds_project as p on p_f.project_code = p.code
join t_ds_user as u on s.user_id = u.id
where 1=1
- <if test="processDefinitionId!= 0">
- and s.process_definition_id = #{processDefinitionId}
+ <if test="processDefinitionCode != 0">
+ and s.process_definition_code = #{processDefinitionCode}
</if>
order by s.update_time desc
</select>
<select id="querySchedulerListByProjectName"
resultType="org.apache.dolphinscheduler.dao.entity.Schedule">
select p_f.name as process_definition_name, p_f.description as
definition_description, p.name as project_name,u.user_name,s.*
from t_ds_schedules s
- join t_ds_process_definition p_f on s.process_definition_id = p_f.id
+ join t_ds_process_definition p_f on s.process_definition_code =
p_f.code
join t_ds_project as p on p_f.project_code = p.code
join t_ds_user as u on s.user_id = u.id
where p.name = #{projectName}
@@ -54,25 +56,25 @@
<include refid="baseSql"/>
from t_ds_schedules
where 1= 1
- <if test="processDefineIds != null and processDefineIds.length != 0 ">
- and process_definition_id in
- <foreach collection="processDefineIds" index="index" item="i"
open="(" separator="," close=")">
+ <if test="processDefineCodes != null and processDefineCodes.length !=
0 ">
+ and process_definition_code in
+ <foreach collection="processDefineCodes" index="index" item="i"
open="(" separator="," close=")">
#{i}
</foreach>
</if>
and release_state = 1
</select>
- <select id="queryByProcessDefinitionId"
resultType="org.apache.dolphinscheduler.dao.entity.Schedule">
+ <select id="queryByProcessDefinitionCode"
resultType="org.apache.dolphinscheduler.dao.entity.Schedule">
select
<include refid="baseSql"/>
from t_ds_schedules
- where process_definition_id =#{processDefinitionId}
+ where process_definition_code = #{processDefinitionCode}
</select>
- <select id="queryReleaseSchedulerListByProcessDefinitionId"
+ <select id="queryReleaseSchedulerListByProcessDefinitionCode"
resultType="org.apache.dolphinscheduler.dao.entity.Schedule">
select
<include refid="baseSql"/>
from t_ds_schedules
- where process_definition_id =#{processDefinitionId} and release_state
= 1
+ where process_definition_code = #{processDefinitionCode} and
release_state = 1
</select>
</mapper>
diff --git
a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkFlowLineageMapper.xml
b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkFlowLineageMapper.xml
index b78113b..eeddaf5 100644
---
a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkFlowLineageMapper.xml
+++
b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkFlowLineageMapper.xml
@@ -22,7 +22,7 @@
<select id="queryByName"
resultType="org.apache.dolphinscheduler.dao.entity.WorkFlowLineage">
select tepd.id as work_flow_id,tepd.name as work_flow_name
from t_ds_process_definition tepd
- left join t_ds_schedules tes on tepd.id = tes.process_definition_id
+ left join t_ds_schedules tes on tepd.code = tes.process_definition_code
where tepd.project_code = #{projectCode}
<if test="searchVal != null and searchVal != ''">
and tepd.name like concat('%', #{searchVal}, '%')
@@ -74,7 +74,7 @@
tes.crontab as crontab,
tes.release_state as schedule_publish_status
from t_ds_process_definition tepd
- left join t_ds_schedules tes on tepd.id = tes.process_definition_id
+ left join t_ds_schedules tes on tepd.code = tes.process_definition_code
where tepd.project_code = #{projectCode} and tepd.code =
#{processDefinitionCode}
</select>
diff --git
a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ScheduleMapperTest.java
b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ScheduleMapperTest.java
index e55aaca..752eca4 100644
---
a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ScheduleMapperTest.java
+++
b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ScheduleMapperTest.java
@@ -14,8 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.dolphinscheduler.dao.mapper;
+package org.apache.dolphinscheduler.dao.mapper;
import org.apache.dolphinscheduler.common.enums.FailureStrategy;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
@@ -24,8 +24,10 @@ import
org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.Schedule;
import org.apache.dolphinscheduler.dao.entity.User;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+
+import java.util.Date;
+import java.util.List;
+
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -35,8 +37,8 @@ import org.springframework.test.annotation.Rollback;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.transaction.annotation.Transactional;
-import java.util.Date;
-import java.util.List;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@RunWith(SpringRunner.class)
@SpringBootTest
@@ -61,7 +63,7 @@ public class ScheduleMapperTest {
* insert
* @return Schedule
*/
- private Schedule insertOne(){
+ private Schedule insertOne() {
//insertOne
Schedule schedule = new Schedule();
schedule.setStartTime(new Date());
@@ -80,7 +82,7 @@ public class ScheduleMapperTest {
* test update
*/
@Test
- public void testUpdate(){
+ public void testUpdate() {
//insertOne
Schedule schedule = insertOne();
schedule.setCreateTime(new Date());
@@ -93,7 +95,7 @@ public class ScheduleMapperTest {
* test delete
*/
@Test
- public void testDelete(){
+ public void testDelete() {
Schedule schedule = insertOne();
int delete = scheduleMapper.deleteById(schedule.getId());
Assert.assertEquals(delete, 1);
@@ -138,18 +140,15 @@ public class ScheduleMapperTest {
processDefinition.setUpdateTime(new Date());
processDefinitionMapper.insert(processDefinition);
- Schedule schedule= insertOne();
+ Schedule schedule = insertOne();
schedule.setUserId(user.getId());
- schedule.setProcessDefinitionId(processDefinition.getId());
+ schedule.setProcessDefinitionCode(processDefinition.getCode());
scheduleMapper.insert(schedule);
Page<Schedule> page = new Page(1,3);
- IPage<Schedule> scheduleIPage =
scheduleMapper.queryByProcessDefineIdPaging(page,
- processDefinition.getId(), ""
- );
+ IPage<Schedule> scheduleIPage =
scheduleMapper.queryByProcessDefineCodePaging(page,
+ processDefinition.getCode(), "");
Assert.assertNotEquals(scheduleIPage.getSize(), 0);
-
-
}
/**
@@ -158,7 +157,6 @@ public class ScheduleMapperTest {
@Test
public void testQuerySchedulerListByProjectName() {
-
User user = new User();
user.setUserName("ut name");
userMapper.insert(user);
@@ -181,9 +179,9 @@ public class ScheduleMapperTest {
processDefinition.setUpdateTime(new Date());
processDefinitionMapper.insert(processDefinition);
- Schedule schedule= insertOne();
+ Schedule schedule = insertOne();
schedule.setUserId(user.getId());
- schedule.setProcessDefinitionId(processDefinition.getId());
+ schedule.setProcessDefinitionCode(processDefinition.getCode());
scheduleMapper.insert(schedule);
Page<Schedule> page = new Page(1,3);
@@ -201,11 +199,11 @@ public class ScheduleMapperTest {
public void testSelectAllByProcessDefineArray() {
Schedule schedule = insertOne();
- schedule.setProcessDefinitionId(12345);
+ schedule.setProcessDefinitionCode(12345);
schedule.setReleaseState(ReleaseState.ONLINE);
scheduleMapper.updateById(schedule);
- List<Schedule> schedules=
scheduleMapper.selectAllByProcessDefineArray(new int[]
{schedule.getProcessDefinitionId()});
+ List<Schedule> schedules =
scheduleMapper.selectAllByProcessDefineArray(new long[]
{schedule.getProcessDefinitionCode()});
Assert.assertNotEquals(schedules.size(), 0);
}
@@ -215,10 +213,10 @@ public class ScheduleMapperTest {
@Test
public void queryByProcessDefinitionId() {
Schedule schedule = insertOne();
- schedule.setProcessDefinitionId(12345);
+ schedule.setProcessDefinitionCode(12345);
scheduleMapper.updateById(schedule);
- List<Schedule> schedules=
scheduleMapper.queryByProcessDefinitionId(schedule.getProcessDefinitionId());
+ List<Schedule> schedules =
scheduleMapper.queryByProcessDefinitionCode(schedule.getProcessDefinitionCode());
Assert.assertNotEquals(schedules.size(), 0);
}
-}
\ No newline at end of file
+}
diff --git
a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/WorkFlowLineageMapperTest.java
b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/WorkFlowLineageMapperTest.java
index 28e9dc2..cc60ce4 100644
---
a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/WorkFlowLineageMapperTest.java
+++
b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/WorkFlowLineageMapperTest.java
@@ -112,7 +112,7 @@ public class WorkFlowLineageMapperTest {
schedule.setWarningType(WarningType.NONE);
schedule.setCreateTime(new Date());
schedule.setUpdateTime(new Date());
- schedule.setProcessDefinitionId(id);
+ schedule.setProcessDefinitionCode(id);
scheduleMapper.insert(schedule);
return schedule;
}
@@ -131,8 +131,9 @@ public class WorkFlowLineageMapperTest {
public void testQueryCodeRelation() {
ProcessTaskRelation processTaskRelation =
insertOneProcessTaskRelation();
- List<ProcessLineage> workFlowLineages =
workFlowLineageMapper.queryCodeRelation(processTaskRelation.getPreTaskCode()
- , processTaskRelation.getPreTaskVersion(), 11L, 1L);
+ List<ProcessLineage> workFlowLineages =
workFlowLineageMapper.queryCodeRelation(
+ processTaskRelation.getPostTaskCode(),
processTaskRelation.getPostTaskVersion(),
+ processTaskRelation.getProcessDefinitionCode(),
processTaskRelation.getProjectCode());
Assert.assertNotEquals(workFlowLineages.size(), 0);
}
diff --git
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
index 1863087..72f7b47 100644
---
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
+++
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
@@ -255,8 +255,8 @@ public class MasterExecThread implements Runnable {
processService.saveProcessInstance(processInstance);
// get schedules
- int processDefinitionId =
processInstance.getProcessDefinition().getId();
- List<Schedule> schedules =
processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId);
+ List<Schedule> schedules =
processService.queryReleaseSchedulerListByProcessDefinitionCode(
+ processInstance.getProcessDefinitionCode());
List<Date> listDate = Lists.newLinkedList();
if (!CollectionUtils.isEmpty(schedules)) {
for (Schedule schedule : schedules) {
diff --git
a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/MasterExecThreadTest.java
b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/MasterExecThreadTest.java
index fbc4ed8..196fb54 100644
---
a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/MasterExecThreadTest.java
+++
b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/MasterExecThreadTest.java
@@ -76,7 +76,7 @@ public class MasterExecThreadTest {
private ProcessService processService;
- private int processDefinitionId = 1;
+ private long processDefinitionCode = 1L;
private MasterConfig config;
@@ -104,6 +104,7 @@ public class MasterExecThreadTest {
processDefinition.setGlobalParamMap(Collections.EMPTY_MAP);
processDefinition.setGlobalParamList(Collections.EMPTY_LIST);
Mockito.when(processInstance.getProcessDefinition()).thenReturn(processDefinition);
+
Mockito.when(processInstance.getProcessDefinitionCode()).thenReturn(processDefinitionCode);
masterExecThread = PowerMockito.spy(new
MasterExecThread(processInstance, processService, null, null, config));
// prepareProcess init dag
@@ -120,9 +121,9 @@ public class MasterExecThreadTest {
* without schedule
*/
@Test
- public void testParallelWithOutSchedule() throws ParseException {
+ public void testParallelWithOutSchedule() {
try {
-
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(zeroSchedulerList());
Method method =
MasterExecThread.class.getDeclaredMethod("executeComplementProcess");
method.setAccessible(true);
method.invoke(masterExecThread);
@@ -140,12 +141,12 @@ public class MasterExecThreadTest {
@Test
public void testParallelWithSchedule() {
try {
-
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(oneSchedulerList());
+
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(oneSchedulerList());
Method method =
MasterExecThread.class.getDeclaredMethod("executeComplementProcess");
method.setAccessible(true);
method.invoke(masterExecThread);
// one create save, and 9(1 to 20 step 2) for next save, and last
day 31 no save
- verify(processService,
times(20)).saveProcessInstance(processInstance);
+ verify(processService,
times(9)).saveProcessInstance(processInstance);
} catch (Exception e) {
Assert.fail();
}
@@ -267,4 +268,4 @@ public class MasterExecThreadTest {
return schedulerList;
}
-}
\ No newline at end of file
+}
diff --git
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
index 83b5e8c..01e4678 100644
---
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
+++
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
@@ -34,7 +34,6 @@ import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.ConditionType;
-import org.apache.dolphinscheduler.common.enums.CycleEnum;
import org.apache.dolphinscheduler.common.enums.Direct;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.enums.FailureStrategy;
@@ -64,7 +63,6 @@ import
org.apache.dolphinscheduler.common.utils.SnowFlakeUtils.SnowFlakeExceptio
import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.common.utils.TaskParametersUtils;
import org.apache.dolphinscheduler.dao.entity.Command;
-import org.apache.dolphinscheduler.dao.entity.CycleDependency;
import org.apache.dolphinscheduler.dao.entity.DagData;
import org.apache.dolphinscheduler.dao.entity.DataSource;
import org.apache.dolphinscheduler.dao.entity.ErrorCommand;
@@ -108,11 +106,9 @@ import org.apache.dolphinscheduler.dao.utils.DagHelper;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.service.exceptions.ServiceException;
import org.apache.dolphinscheduler.service.log.LogClientService;
-import org.apache.dolphinscheduler.service.quartz.cron.CronUtils;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Calendar;
import java.util.Date;
import java.util.EnumMap;
import java.util.HashMap;
@@ -125,14 +121,12 @@ import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
-import org.quartz.CronExpression;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
-import com.cronutils.model.Cron;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -404,6 +398,16 @@ public class ProcessService {
}
/**
+ * find process define by code.
+ *
+ * @param processDefinitionCode processDefinitionCode
+ * @return process definition
+ */
+ public ProcessDefinition findProcessDefinitionByCode(Long
processDefinitionCode) {
+ return processDefineMapper.queryByCode(processDefinitionCode);
+ }
+
+ /**
* delete work process instance by id
*
* @param processInstanceId processInstanceId
@@ -483,7 +487,6 @@ public class ProcessService {
public void recurseFindSubProcessId(int parentId, List<Integer> ids) {
List<TaskDefinition> taskNodeList =
this.getTaskNodeListByDefinitionId(parentId);
-
if (taskNodeList != null && !taskNodeList.isEmpty()) {
for (TaskDefinition taskNode : taskNodeList) {
@@ -1628,7 +1631,6 @@ public class ProcessService {
taskInstance.setTaskParams(JSONUtils.toJsonString(taskParams));
}
-
/**
* convert integer list to string list
*
@@ -1657,13 +1659,13 @@ public class ProcessService {
}
/**
- * query Schedule by processDefinitionId
+ * query Schedule by processDefinitionCode
*
- * @param processDefinitionId processDefinitionId
+ * @param processDefinitionCode processDefinitionCode
* @see Schedule
*/
- public List<Schedule> queryReleaseSchedulerListByProcessDefinitionId(int
processDefinitionId) {
- return
scheduleMapper.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId);
+ public List<Schedule>
queryReleaseSchedulerListByProcessDefinitionCode(long processDefinitionCode) {
+ return
scheduleMapper.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode);
}
/**
@@ -1730,7 +1732,6 @@ public class ProcessService {
ProcessInstance instance =
processInstanceMapper.selectById(processInstanceId);
instance.setState(executionStatus);
return processInstanceMapper.updateById(instance);
-
}
/**
@@ -1785,98 +1786,13 @@ public class ProcessService {
}
/**
- * find schedule list by process define id.
+ * find schedule list by process define codes.
*
- * @param ids ids
+ * @param codes codes
* @return schedule list
*/
- public List<Schedule> selectAllByProcessDefineId(int[] ids) {
- return scheduleMapper.selectAllByProcessDefineArray(
- ids);
- }
-
- /**
- * get dependency cycle by work process define id and scheduler fire time
- *
- * @param masterId masterId
- * @param processDefinitionId processDefinitionId
- * @param scheduledFireTime the time the task schedule is expected to
trigger
- * @return CycleDependency
- * @throws Exception if error throws Exception
- */
- public CycleDependency getCycleDependency(int masterId, int
processDefinitionId, Date scheduledFireTime) throws Exception {
- List<CycleDependency> list = getCycleDependencies(masterId, new
int[]{processDefinitionId}, scheduledFireTime);
- return !list.isEmpty() ? list.get(0) : null;
-
- }
-
- /**
- * get dependency cycle list by work process define id list and scheduler
fire time
- *
- * @param masterId masterId
- * @param ids ids
- * @param scheduledFireTime the time the task schedule is expected to
trigger
- * @return CycleDependency list
- * @throws Exception if error throws Exception
- */
- public List<CycleDependency> getCycleDependencies(int masterId, int[] ids,
Date scheduledFireTime) throws Exception {
- List<CycleDependency> cycleDependencyList = new ArrayList<>();
- if (null == ids || ids.length == 0) {
- logger.warn("ids[] is empty!is invalid!");
- return cycleDependencyList;
- }
- if (scheduledFireTime == null) {
- logger.warn("scheduledFireTime is null!is invalid!");
- return cycleDependencyList;
- }
-
- String strCrontab = "";
- CronExpression depCronExpression;
- Cron depCron;
- List<Date> list;
- List<Schedule> schedules = this.selectAllByProcessDefineId(ids);
- // for all scheduling information
- for (Schedule depSchedule : schedules) {
- strCrontab = depSchedule.getCrontab();
- depCronExpression = CronUtils.parse2CronExpression(strCrontab);
- depCron = CronUtils.parse2Cron(strCrontab);
- CycleEnum cycleEnum = CronUtils.getMiniCycle(depCron);
- if (cycleEnum == null) {
- logger.error("{} is not valid", strCrontab);
- continue;
- }
- Calendar calendar = Calendar.getInstance();
- switch (cycleEnum) {
- case HOUR:
- calendar.add(Calendar.HOUR, -25);
- break;
- case DAY:
- case WEEK:
- calendar.add(Calendar.DATE, -32);
- break;
- case MONTH:
- calendar.add(Calendar.MONTH, -13);
- break;
- default:
- String cycleName = cycleEnum.name();
- logger.warn("Dependent process definition's cycleEnum is
{},not support!!", cycleName);
- continue;
- }
- Date start = calendar.getTime();
-
- if (depSchedule.getProcessDefinitionId() == masterId) {
- list = CronUtils.getSelfFireDateList(start, scheduledFireTime,
depCronExpression);
- } else {
- list = CronUtils.getFireDateList(start, scheduledFireTime,
depCronExpression);
- }
- if (!list.isEmpty()) {
- start = list.get(list.size() - 1);
- CycleDependency dependency = new
CycleDependency(depSchedule.getProcessDefinitionId(), start,
CronUtils.getExpirationTime(start, cycleEnum), cycleEnum);
- cycleDependencyList.add(dependency);
- }
-
- }
- return cycleDependencyList;
+ public List<Schedule> selectAllByProcessDefineCode(long[] codes) {
+ return scheduleMapper.selectAllByProcessDefineArray(codes);
}
/**
diff --git
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/ProcessScheduleJob.java
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/ProcessScheduleJob.java
index 2921ce2..bda8ad8 100644
---
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/ProcessScheduleJob.java
+++
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/ProcessScheduleJob.java
@@ -81,11 +81,11 @@ public class ProcessScheduleJob implements Job {
return;
}
- ProcessDefinition processDefinition =
getProcessService().findProcessDefineById(schedule.getProcessDefinitionId());
+ ProcessDefinition processDefinition =
getProcessService().findProcessDefinitionByCode(schedule.getProcessDefinitionCode());
// release state : online/offline
ReleaseState releaseState = processDefinition.getReleaseState();
if (releaseState == ReleaseState.OFFLINE) {
- logger.warn("process definition does not exist in db or
offline，need not to create command, projectId:{}, processId:{}", projectId,
scheduleId);
+ logger.warn("process definition does not exist in db or
offline，need not to create command, projectId:{}, processId:{}", projectId,
processDefinition.getId());
return;
}
@@ -93,7 +93,7 @@ public class ProcessScheduleJob implements Job {
command.setCommandType(CommandType.SCHEDULER);
command.setExecutorId(schedule.getUserId());
command.setFailureStrategy(schedule.getFailureStrategy());
- command.setProcessDefinitionId(schedule.getProcessDefinitionId());
+ //command.setProcessDefinitionId(schedule.getProcessDefinitionCode());
TODO next pr
command.setScheduleTime(scheduledFireTime);
command.setStartTime(fireTime);
command.setWarningGroupId(schedule.getWarningGroupId());
diff --git a/sql/dolphinscheduler_mysql.sql b/sql/dolphinscheduler_mysql.sql
index a6e3f97..11bdc8d 100644
--- a/sql/dolphinscheduler_mysql.sql
+++ b/sql/dolphinscheduler_mysql.sql
@@ -747,7 +747,7 @@ CREATE TABLE `t_ds_resources` (
DROP TABLE IF EXISTS `t_ds_schedules`;
CREATE TABLE `t_ds_schedules` (
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'key',
- `process_definition_id` int(11) NOT NULL COMMENT 'process definition id',
+ `process_definition_code` bigint(20) NOT NULL COMMENT 'process definition
code',
`start_time` datetime NOT NULL COMMENT 'start time',
`end_time` datetime NOT NULL COMMENT 'end time',
`timezone_id` varchar(40) DEFAULT NULL COMMENT 'timezoneId',
diff --git a/sql/dolphinscheduler_postgre.sql b/sql/dolphinscheduler_postgre.sql
index 88cf143..eb97912 100644
--- a/sql/dolphinscheduler_postgre.sql
+++ b/sql/dolphinscheduler_postgre.sql
@@ -611,7 +611,7 @@ CREATE TABLE t_ds_resources (
DROP TABLE IF EXISTS t_ds_schedules;
CREATE TABLE t_ds_schedules (
id int NOT NULL ,
- process_definition_id int NOT NULL ,
+ process_definition_code bigint NOT NULL ,
start_time timestamp NOT NULL ,
end_time timestamp NOT NULL ,
timezone_id varchar(40) default NULL ,
