@ -17,7 +17,10 @@
package org.apache.dolphinscheduler.service.process ;
package org.apache.dolphinscheduler.service.process ;
import io.micrometer.core.annotation.Counted ;
import static java.util.stream.Collectors.toSet ;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE ;
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.CMDPARAM_COMPLEMENT_DATA_START_DATE ;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_EMPTY_SUB_PROCESS ;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_EMPTY_SUB_PROCESS ;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_FATHER_PARAMS ;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_FATHER_PARAMS ;
@ -30,8 +33,6 @@ 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.api.enums.Direct.IN ;
import static org.apache.dolphinscheduler.plugin.task.api.utils.DataQualityConstants.TASK_INSTANCE_ID ;
import static org.apache.dolphinscheduler.plugin.task.api.utils.DataQualityConstants.TASK_INSTANCE_ID ;
import static java.util.stream.Collectors.toSet ;
import org.apache.dolphinscheduler.common.Constants ;
import org.apache.dolphinscheduler.common.Constants ;
import org.apache.dolphinscheduler.common.enums.AuthorizationType ;
import org.apache.dolphinscheduler.common.enums.AuthorizationType ;
import org.apache.dolphinscheduler.common.enums.CommandType ;
import org.apache.dolphinscheduler.common.enums.CommandType ;
@ -42,7 +43,6 @@ import org.apache.dolphinscheduler.common.enums.TaskDependType;
import org.apache.dolphinscheduler.common.enums.TaskGroupQueueStatus ;
import org.apache.dolphinscheduler.common.enums.TaskGroupQueueStatus ;
import org.apache.dolphinscheduler.common.enums.TimeoutFlag ;
import org.apache.dolphinscheduler.common.enums.TimeoutFlag ;
import org.apache.dolphinscheduler.common.enums.WarningType ;
import org.apache.dolphinscheduler.common.enums.WarningType ;
import org.apache.dolphinscheduler.service.expand.CuringParamsService ;
import org.apache.dolphinscheduler.common.graph.DAG ;
import org.apache.dolphinscheduler.common.graph.DAG ;
import org.apache.dolphinscheduler.common.model.TaskNode ;
import org.apache.dolphinscheduler.common.model.TaskNode ;
import org.apache.dolphinscheduler.common.model.TaskNodeRelation ;
import org.apache.dolphinscheduler.common.model.TaskNodeRelation ;
@ -51,6 +51,7 @@ import org.apache.dolphinscheduler.common.utils.CodeGenerateUtils;
import org.apache.dolphinscheduler.common.utils.CodeGenerateUtils.CodeGenerateException ;
import org.apache.dolphinscheduler.common.utils.CodeGenerateUtils.CodeGenerateException ;
import org.apache.dolphinscheduler.common.utils.DateUtils ;
import org.apache.dolphinscheduler.common.utils.DateUtils ;
import org.apache.dolphinscheduler.common.utils.JSONUtils ;
import org.apache.dolphinscheduler.common.utils.JSONUtils ;
import org.apache.dolphinscheduler.common.utils.ParameterUtils ;
import org.apache.dolphinscheduler.dao.entity.Command ;
import org.apache.dolphinscheduler.dao.entity.Command ;
import org.apache.dolphinscheduler.dao.entity.DagData ;
import org.apache.dolphinscheduler.dao.entity.DagData ;
import org.apache.dolphinscheduler.dao.entity.DataSource ;
import org.apache.dolphinscheduler.dao.entity.DataSource ;
@ -131,13 +132,13 @@ import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext ;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext ;
import org.apache.dolphinscheduler.service.corn.CronUtils ;
import org.apache.dolphinscheduler.service.corn.CronUtils ;
import org.apache.dolphinscheduler.service.exceptions.ServiceException ;
import org.apache.dolphinscheduler.service.exceptions.ServiceException ;
import org.apache.dolphinscheduler.service.expand.CuringParamsService ;
import org.apache.dolphinscheduler.service.log.LogClientService ;
import org.apache.dolphinscheduler.service.log.LogClientService ;
import org.apache.dolphinscheduler.service.task.TaskPluginManager ;
import org.apache.dolphinscheduler.service.task.TaskPluginManager ;
import org.apache.dolphinscheduler.spi.enums.ResourceType ;
import org.apache.dolphinscheduler.spi.enums.ResourceType ;
import org.apache.commons.collections.CollectionUtils ;
import org.apache.commons.collections.CollectionUtils ;
import java.sql.SQLIntegrityConstraintViolationException ;
import java.util.ArrayList ;
import java.util.ArrayList ;
import java.util.Arrays ;
import java.util.Arrays ;
import java.util.Date ;
import java.util.Date ;
@ -151,6 +152,7 @@ import java.util.Objects;
import java.util.Set ;
import java.util.Set ;
import java.util.stream.Collectors ;
import java.util.stream.Collectors ;
import org.apache.dolphinscheduler.spi.utils.StringUtils ;
import org.slf4j.Logger ;
import org.slf4j.Logger ;
import org.slf4j.LoggerFactory ;
import org.slf4j.LoggerFactory ;
import org.springframework.beans.factory.annotation.Autowired ;
import org.springframework.beans.factory.annotation.Autowired ;
@ -164,8 +166,6 @@ import com.google.common.base.Joiner;
import com.google.common.base.Strings ;
import com.google.common.base.Strings ;
import com.google.common.collect.Lists ;
import com.google.common.collect.Lists ;
import io.micrometer.core.annotation.Counted ;
/ * *
/ * *
* process relative dao that some mappers in this .
* process relative dao that some mappers in this .
* /
* /
@ -1129,8 +1129,14 @@ public class ProcessServiceImpl implements ProcessService {
Date start = DateUtils . stringToDate ( cmdParam . get ( CMDPARAM_COMPLEMENT_DATA_START_DATE ) ) ;
Date start = DateUtils . stringToDate ( cmdParam . get ( CMDPARAM_COMPLEMENT_DATA_START_DATE ) ) ;
Date end = DateUtils . stringToDate ( cmdParam . get ( CMDPARAM_COMPLEMENT_DATA_END_DATE ) ) ;
Date end = DateUtils . stringToDate ( cmdParam . get ( CMDPARAM_COMPLEMENT_DATA_END_DATE ) ) ;
List < Date > complementDate = Lists . newLinkedList ( ) ;
if ( start ! = null & & end ! = null ) {
List < Schedule > listSchedules = queryReleaseSchedulerListByProcessDefinitionCode ( processInstance . getProcessDefinitionCode ( ) ) ;
List < Schedule > listSchedules = queryReleaseSchedulerListByProcessDefinitionCode ( processInstance . getProcessDefinitionCode ( ) ) ;
List < Date > complementDate = CronUtils . getSelfFireDateList ( start , end , listSchedules ) ;
complementDate = CronUtils . getSelfFireDateList ( start , end , listSchedules ) ;
}
if ( cmdParam . containsKey ( CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST ) ) {
complementDate = CronUtils . getSelfScheduleDateList ( cmdParam ) ;
}
if ( complementDate . size ( ) > 0
if ( complementDate . size ( ) > 0
& & Flag . NO = = processInstance . getIsSubProcess ( ) ) {
& & Flag . NO = = processInstance . getIsSubProcess ( ) ) {
@ -1409,8 +1415,14 @@ public class ProcessServiceImpl implements ProcessService {
Map < String , String > parentParam = JSONUtils . toMap ( parentProcessInstance . getCommandParam ( ) ) ;
Map < String , String > parentParam = JSONUtils . toMap ( parentProcessInstance . getCommandParam ( ) ) ;
String endTime = parentParam . get ( CMDPARAM_COMPLEMENT_DATA_END_DATE ) ;
String endTime = parentParam . get ( CMDPARAM_COMPLEMENT_DATA_END_DATE ) ;
String startTime = parentParam . get ( CMDPARAM_COMPLEMENT_DATA_START_DATE ) ;
String startTime = parentParam . get ( CMDPARAM_COMPLEMENT_DATA_START_DATE ) ;
String scheduleTime = parentParam . get ( CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST ) ;
if ( StringUtils . isNotEmpty ( startTime ) & & StringUtils . isNotEmpty ( endTime ) ) {
cmdParam . put ( CMDPARAM_COMPLEMENT_DATA_END_DATE , endTime ) ;
cmdParam . put ( CMDPARAM_COMPLEMENT_DATA_END_DATE , endTime ) ;
cmdParam . put ( CMDPARAM_COMPLEMENT_DATA_START_DATE , startTime ) ;
cmdParam . put ( CMDPARAM_COMPLEMENT_DATA_START_DATE , startTime ) ;
}
if ( StringUtils . isNotEmpty ( scheduleTime ) ) {
cmdParam . put ( CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST , scheduleTime ) ;
}
processMapStr = JSONUtils . toJsonString ( cmdParam ) ;
processMapStr = JSONUtils . toJsonString ( cmdParam ) ;
}
}
if ( fatherParams . size ( ) ! = 0 ) {
if ( fatherParams . size ( ) ! = 0 ) {