natas_hw
/
dolphinscheduler
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[Bug][Timezone][3.0.1] fix timezone when complement data and preview (#12093) 

* fix timezone in complement data and schedule preview

* fix timezone in complement data and schedule preview

* code improvement
This commit is contained in:
caishunfeng 2022-09-22 14:32:24 +08:00 committed by GitHub
parent b3c27efac3
commit 54a19b3cb7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 40 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
View File

@ -18,6 +18,7 @@
package org.apache.dolphinscheduler.api.service.impl;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_NODES;
@ -77,6 +78,7 @@ import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
@ -84,6 +86,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TimeZone;
import java.util.stream.Collectors;
import org.slf4j.Logger;
@ -704,14 +707,16 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
logger.warn("The startDate {} is later than the endDate {}", start, end);
break;
}
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(start));
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(end));
command.setCommandParam(JSONUtils.toJsonString(cmdParam));
createCount = processService.createCommand(command);
// dependent process definition
List<Schedule> schedules = processService.queryReleaseSchedulerListByProcessDefinitionCode(command.getProcessDefinitionCode());
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(start, ZoneId.systemDefault().getId()));
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(end, ZoneId.systemDefault().getId()));
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST, JSONUtils.toJsonString(CronUtils.getSelfFireDateList(start, end, schedules)));
command.setCommandParam(JSONUtils.toJsonString(cmdParam));
createCount = processService.createCommand(command);
if (schedules.isEmpty() || complementDependentMode == ComplementDependentMode.OFF_MODE) {
logger.info("process code: {} complement dependent in off mode or schedule's size is 0, skip "
+ "dependent complement data", command.getProcessDefinitionCode());
@ -727,6 +732,8 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
break;
}
logger.info("before createComplementCommandList {} {}", start, end);
List<Date> listDate = new ArrayList<>();
List<Schedule> schedules = processService.queryReleaseSchedulerListByProcessDefinitionCode(command.getProcessDefinitionCode());
listDate.addAll(CronUtils.getSelfFireDateList(start, end, schedules));
@ -759,9 +766,21 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
endDateIndex += singleCommandItems;
}
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(listDate.get(startDateIndex)));
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(listDate.get(endDateIndex)));
Date startDate = listDate.get(startDateIndex);
Date endDate = listDate.get(endDateIndex);
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(startDate, ZoneId.systemDefault().getId()));
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(endDate, ZoneId.systemDefault().getId()));
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST, JSONUtils.toJsonString(CronUtils.getSelfFireDateList(startDate, endDate, schedules)));
command.setCommandParam(JSONUtils.toJsonString(cmdParam));
logger.info("before create complement command: startDate:{}, startStr:{}, endDate:{}, endStr:{}, complement date list: {}",
listDate.get(startDateIndex),
cmdParam.get(CMDPARAM_COMPLEMENT_DATA_START_DATE),
listDate.get(endDateIndex),
cmdParam.get(CMDPARAM_COMPLEMENT_DATA_END_DATE),
cmdParam.get(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST));
processService.createCommand(command);
if (schedules.isEmpty() || complementDependentMode == ComplementDependentMode.OFF_MODE) {

5
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java
View File

@ -34,6 +34,7 @@ import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.model.Server;
import org.apache.dolphinscheduler.common.thread.ThreadLocalContext;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
@ -53,11 +54,13 @@ import org.apache.dolphinscheduler.service.quartz.cron.CronUtils;
import org.apache.commons.lang.StringUtils;
import java.text.ParseException;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;
import org.quartz.CronExpression;
import org.quartz.JobKey;
@ -578,11 +581,13 @@ public class SchedulerServiceImpl extends BaseServiceImpl implements SchedulerSe
try {
cronExpression = CronUtils.parse2CronExpression(scheduleParam.getCrontab());
cronExpression.setTimeZone(TimeZone.getTimeZone(scheduleParam.getTimezoneId()));
} catch (ParseException e) {
logger.error(e.getMessage(), e);
putMsg(result, Status.PARSE_TO_CRON_EXPRESSION_ERROR);
return result;
}
List<Date> selfFireDateList = CronUtils.getSelfFireDateList(startTime, endTime, cronExpression, Constants.PREVIEW_SCHEDULE_EXECUTE_COUNT);
List<String> previewDateList = new ArrayList<>();
selfFireDateList.forEach(date -> previewDateList.add(DateUtils.dateToString(date, scheduleParam.getTimezoneId())));

17
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java
View File

@ -622,9 +622,7 @@ public class WorkflowExecuteRunnable implements Callable<WorkflowSubmitStatue> {
}
if (cmdParam.containsKey(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST)) {
cmdParam.replace(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST,
cmdParam.get(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST)
.substring(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST).indexOf(COMMA) + 1));
cmdParam.replace(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST, JSONUtils.toJsonString(complementListDate.subList(1, complementListDate.size())));
}
if (cmdParam.containsKey(CMDPARAM_COMPLEMENT_DATA_START_DATE)) {
@ -848,18 +846,9 @@ public class WorkflowExecuteRunnable implements Callable<WorkflowSubmitStatue> {
// reset global params while there are start parameters
setGlobalParamIfCommanded(processDefinition, cmdParam);
Date start = null;
Date end = null;
if(cmdParam.containsKey(CMDPARAM_COMPLEMENT_DATA_START_DATE) && cmdParam.containsKey(CMDPARAM_COMPLEMENT_DATA_END_DATE)){
start = DateUtils.stringToDate(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_START_DATE));
end = DateUtils.stringToDate(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_END_DATE));
}
List<Schedule> schedules = processService.queryReleaseSchedulerListByProcessDefinitionCode(processInstance.getProcessDefinitionCode());
if (complementListDate.isEmpty() && needComplementProcess()) {
complementListDate = CronUtils.getSelfFireDateList(start, end, schedules);
logger.info(" process definition code:{} complement data: {}",
processInstance.getProcessDefinitionCode(),
complementListDate.toString());
complementListDate = JSONUtils.toList(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST), Date.class);
logger.info("process definition code:{}, complement data: {}", processInstance.getProcessDefinitionCode(), complementListDate.toString());
if (!complementListDate.isEmpty() && Flag.NO == processInstance.getIsSubProcess()) {
processInstance.setScheduleTime(complementListDate.get(0));

7
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/cron/CronUtils.java
View File

@ -29,18 +29,22 @@ import static com.cronutils.model.CronType.QUARTZ;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.CycleEnum;
import org.apache.dolphinscheduler.common.thread.Stopper;
import org.apache.dolphinscheduler.common.thread.ThreadLocalContext;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.dao.entity.Schedule;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.text.ParseException;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.GregorianCalendar;
import java.util.List;
import java.util.TimeZone;
import org.quartz.CronExpression;
import org.slf4j.Logger;
@ -225,6 +229,9 @@ public class CronUtils {
logger.error(e.getMessage(), e);
return Collections.emptyList();
}
String timezone = ThreadLocalContext.getTimezoneThreadLocal().get();
ZoneId zoneId = StringUtils.isNotEmpty(timezone) ? ZoneId.of(timezone) : ZoneId.systemDefault();
cronExpression.setTimeZone(TimeZone.getTimeZone(zoneId));
return getSelfFireDateList(startTime, endTime, cronExpression);
}