authorizedUdfs = udfFuncMapper.listAuthorizedUdfFunc(userId, needChecks).stream().map(UdfFunc::getId).collect(toSet());
originResSet.removeAll(authorizedUdfs);
break;
default:
@@ -2007,9 +1983,6 @@ public class ProcessService {
/**
* format task app id in task instance
- *
- * @param taskInstance
- * @return
*/
public String formatTaskAppId(TaskInstance taskInstance) {
ProcessDefinition definition = this.findProcessDefineById(taskInstance.getProcessDefinitionId());
@@ -2019,9 +1992,9 @@ public class ProcessService {
return "";
}
return String.format("%s_%s_%s",
- definition.getId(),
- processInstanceById.getId(),
- taskInstance.getId());
+ definition.getId(),
+ processInstanceById.getId(),
+ taskInstance.getId());
}
}
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 6ac847b8db3b0bbed0f92f8c10fc2dd0ea84a97f..2921ce2bbadd9fb1573c9f8e52bdc6fe11fe4fe2 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
@@ -14,8 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.dolphinscheduler.service.quartz;
+package org.apache.dolphinscheduler.service.quartz;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.CommandType;
@@ -25,6 +25,9 @@ import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.Schedule;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.process.ProcessService;
+
+import java.util.Date;
+
import org.quartz.Job;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
@@ -34,8 +37,6 @@ import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
-import java.util.Date;
-
/**
* process schedule job
*/
@@ -46,7 +47,7 @@ public class ProcessScheduleJob implements Job {
*/
private static final Logger logger = LoggerFactory.getLogger(ProcessScheduleJob.class);
- public ProcessService getProcessService(){
+ public ProcessService getProcessService() {
return SpringApplicationContext.getBean(ProcessService.class);
}
@@ -66,10 +67,8 @@ public class ProcessScheduleJob implements Job {
int projectId = dataMap.getInt(Constants.PROJECT_ID);
int scheduleId = dataMap.getInt(Constants.SCHEDULE_ID);
-
Date scheduledFireTime = context.getScheduledFireTime();
-
Date fireTime = context.getFireTime();
logger.info("scheduled fire time :{}, fire time :{}, process id :{}", scheduledFireTime, fireTime, scheduleId);
@@ -82,11 +81,10 @@ public class ProcessScheduleJob implements Job {
return;
}
-
ProcessDefinition processDefinition = getProcessService().findProcessDefineById(schedule.getProcessDefinitionId());
// release state : online/offline
ReleaseState releaseState = processDefinition.getReleaseState();
- if (processDefinition == null || releaseState == ReleaseState.OFFLINE) {
+ if (releaseState == ReleaseState.OFFLINE) {
logger.warn("process definition does not exist in db or offline，need not to create command, projectId:{}, processId:{}", projectId, scheduleId);
return;
}
@@ -107,7 +105,6 @@ public class ProcessScheduleJob implements Job {
getProcessService().createCommand(command);
}
-
/**
* delete job
*/
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/QuartzExecutors.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/QuartzExecutors.java
index 3b15810e057b6c592d90cf64fe5b00a4497d4ae5..fd91e4076dd04fafe93f6c8265ee157a7ff10a5f 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/QuartzExecutors.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/QuartzExecutors.java
@@ -14,15 +14,76 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.service.quartz;
+import static org.apache.dolphinscheduler.common.Constants.ORG_POSTGRESQL_DRIVER;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_DATASOURCE_MYDS_CONNECTIONPROVIDER_CLASS;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_JOBSTORE_ACQUIRETRIGGERSWITHINLOCK;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_JOBSTORE_CLASS;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_JOBSTORE_CLUSTERCHECKININTERVAL;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_JOBSTORE_DATASOURCE;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_JOBSTORE_DRIVERDELEGATECLASS;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_JOBSTORE_ISCLUSTERED;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_JOBSTORE_MISFIRETHRESHOLD;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_JOBSTORE_TABLEPREFIX;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_JOBSTORE_USEPROPERTIES;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_SCHEDULER_INSTANCEID;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_SCHEDULER_INSTANCENAME;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_SCHEDULER_MAKESCHEDULERTHREADDAEMON;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_THREADPOOL_CLASS;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_THREADPOOL_MAKETHREADSDAEMONS;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_THREADPOOL_THREADCOUNT;
+import static org.apache.dolphinscheduler.common.Constants.ORG_QUARTZ_THREADPOOL_THREADPRIORITY;
+import static org.apache.dolphinscheduler.common.Constants.PROJECT_ID;
+import static org.apache.dolphinscheduler.common.Constants.QUARTZ_ACQUIRETRIGGERSWITHINLOCK;
+import static org.apache.dolphinscheduler.common.Constants.QUARTZ_CLUSTERCHECKININTERVAL;
+import static org.apache.dolphinscheduler.common.Constants.QUARTZ_DATASOURCE;
+import static org.apache.dolphinscheduler.common.Constants.QUARTZ_INSTANCEID;
+import static org.apache.dolphinscheduler.common.Constants.QUARTZ_INSTANCENAME;
+import static org.apache.dolphinscheduler.common.Constants.QUARTZ_JOB_GROUP_PRIFIX;
+import static org.apache.dolphinscheduler.common.Constants.QUARTZ_JOB_PRIFIX;
+import static org.apache.dolphinscheduler.common.Constants.QUARTZ_MISFIRETHRESHOLD;
+import static org.apache.dolphinscheduler.common.Constants.QUARTZ_PROPERTIES_PATH;
+import static org.apache.dolphinscheduler.common.Constants.QUARTZ_TABLE_PREFIX;
+import static org.apache.dolphinscheduler.common.Constants.QUARTZ_THREADCOUNT;
+import static org.apache.dolphinscheduler.common.Constants.QUARTZ_THREADPRIORITY;
+import static org.apache.dolphinscheduler.common.Constants.SCHEDULE;
+import static org.apache.dolphinscheduler.common.Constants.SCHEDULE_ID;
+import static org.apache.dolphinscheduler.common.Constants.SPRING_DATASOURCE_DRIVER_CLASS_NAME;
+import static org.apache.dolphinscheduler.common.Constants.STRING_FALSE;
+import static org.apache.dolphinscheduler.common.Constants.STRING_TRUE;
+import static org.apache.dolphinscheduler.common.Constants.UNDERLINE;
+
+import static org.quartz.CronScheduleBuilder.cronSchedule;
+import static org.quartz.JobBuilder.newJob;
+import static org.quartz.TriggerBuilder.newTrigger;
+
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.common.utils.StringUtils;
+import org.apache.dolphinscheduler.dao.entity.Schedule;
+import org.apache.dolphinscheduler.service.exceptions.ServiceException;
+
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
-import org.apache.commons.lang.StringUtils;
-import org.apache.dolphinscheduler.common.utils.*;
-import org.apache.dolphinscheduler.dao.entity.Schedule;
-import org.quartz.*;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.quartz.CronTrigger;
+import org.quartz.Job;
+import org.quartz.JobDetail;
+import org.quartz.JobKey;
+import org.quartz.Scheduler;
+import org.quartz.SchedulerException;
+import org.quartz.TriggerKey;
import org.quartz.impl.StdSchedulerFactory;
import org.quartz.impl.jdbcjobstore.JobStoreTX;
import org.quartz.impl.jdbcjobstore.PostgreSQLDelegate;
@@ -32,300 +93,289 @@ import org.quartz.simpl.SimpleThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.*;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-import static org.apache.dolphinscheduler.common.Constants.*;
-import static org.quartz.CronScheduleBuilder.cronSchedule;
-import static org.quartz.JobBuilder.newJob;
-import static org.quartz.TriggerBuilder.newTrigger;
-
/**
* single Quartz executors instance
*/
public class QuartzExecutors {
- /**
- * logger of QuartzExecutors
- */
- private static final Logger logger = LoggerFactory.getLogger(QuartzExecutors.class);
-
- /**
- * read write lock
- */
- private final ReadWriteLock lock = new ReentrantReadWriteLock();
-
- /**
- * A Scheduler maintains a registry of org.quartz.JobDetail and Trigger.
- */
- private static Scheduler scheduler;
-
- /**
- * load conf
- */
- private static Configuration conf;
-
- private static final class Holder {
- private static final QuartzExecutors instance = new QuartzExecutors();
- }
-
-
- private QuartzExecutors() {
- try {
- conf = new PropertiesConfiguration(QUARTZ_PROPERTIES_PATH);
- init();
- }catch (ConfigurationException e){
- logger.warn("not loaded quartz configuration file, will used default value",e);
+ /**
+ * logger of QuartzExecutors
+ */
+ private static final Logger logger = LoggerFactory.getLogger(QuartzExecutors.class);
+
+ /**
+ * read write lock
+ */
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ /**
+ * A Scheduler maintains a registry of org.quartz.JobDetail and Trigger.
+ */
+ private static Scheduler scheduler;
+
+ /**
+ * load conf
+ */
+ private static Configuration conf;
+
+ private static final class Holder {
+ private static final QuartzExecutors instance = new QuartzExecutors();
}
- }
-
- /**
- * thread safe and performance promote
- * @return instance of Quartz Executors
- */
- public static QuartzExecutors getInstance() {
- return Holder.instance;
- }
-
-
- /**
- * init
- *
- * Returns a client-usable handle to a Scheduler.
- */
- private void init() {
- try {
- StdSchedulerFactory schedulerFactory = new StdSchedulerFactory();
- Properties properties = new Properties();
-
- String dataSourceDriverClass = org.apache.dolphinscheduler.dao.utils.PropertyUtils.getString(SPRING_DATASOURCE_DRIVER_CLASS_NAME);
- if (dataSourceDriverClass.equals(ORG_POSTGRESQL_DRIVER)){
- properties.setProperty(ORG_QUARTZ_JOBSTORE_DRIVERDELEGATECLASS,conf.getString(ORG_QUARTZ_JOBSTORE_DRIVERDELEGATECLASS, PostgreSQLDelegate.class.getName()));
- } else {
- properties.setProperty(ORG_QUARTZ_JOBSTORE_DRIVERDELEGATECLASS,conf.getString(ORG_QUARTZ_JOBSTORE_DRIVERDELEGATECLASS, StdJDBCDelegate.class.getName()));
- }
- properties.setProperty(ORG_QUARTZ_SCHEDULER_INSTANCENAME, conf.getString(ORG_QUARTZ_SCHEDULER_INSTANCENAME, QUARTZ_INSTANCENAME));
- properties.setProperty(ORG_QUARTZ_SCHEDULER_INSTANCEID, conf.getString(ORG_QUARTZ_SCHEDULER_INSTANCEID, QUARTZ_INSTANCEID));
- properties.setProperty(ORG_QUARTZ_SCHEDULER_MAKESCHEDULERTHREADDAEMON,conf.getString(ORG_QUARTZ_SCHEDULER_MAKESCHEDULERTHREADDAEMON,STRING_TRUE));
- properties.setProperty(ORG_QUARTZ_JOBSTORE_USEPROPERTIES,conf.getString(ORG_QUARTZ_JOBSTORE_USEPROPERTIES,STRING_FALSE));
- properties.setProperty(ORG_QUARTZ_THREADPOOL_CLASS,conf.getString(ORG_QUARTZ_THREADPOOL_CLASS, SimpleThreadPool.class.getName()));
- properties.setProperty(ORG_QUARTZ_THREADPOOL_MAKETHREADSDAEMONS,conf.getString(ORG_QUARTZ_THREADPOOL_MAKETHREADSDAEMONS,STRING_TRUE));
- properties.setProperty(ORG_QUARTZ_THREADPOOL_THREADCOUNT,conf.getString(ORG_QUARTZ_THREADPOOL_THREADCOUNT, QUARTZ_THREADCOUNT));
- properties.setProperty(ORG_QUARTZ_THREADPOOL_THREADPRIORITY,conf.getString(ORG_QUARTZ_THREADPOOL_THREADPRIORITY, QUARTZ_THREADPRIORITY));
- properties.setProperty(ORG_QUARTZ_JOBSTORE_CLASS,conf.getString(ORG_QUARTZ_JOBSTORE_CLASS, JobStoreTX.class.getName()));
- properties.setProperty(ORG_QUARTZ_JOBSTORE_TABLEPREFIX,conf.getString(ORG_QUARTZ_JOBSTORE_TABLEPREFIX, QUARTZ_TABLE_PREFIX));
- properties.setProperty(ORG_QUARTZ_JOBSTORE_ISCLUSTERED,conf.getString(ORG_QUARTZ_JOBSTORE_ISCLUSTERED,STRING_TRUE));
- properties.setProperty(ORG_QUARTZ_JOBSTORE_MISFIRETHRESHOLD,conf.getString(ORG_QUARTZ_JOBSTORE_MISFIRETHRESHOLD, QUARTZ_MISFIRETHRESHOLD));
- properties.setProperty(ORG_QUARTZ_JOBSTORE_CLUSTERCHECKININTERVAL,conf.getString(ORG_QUARTZ_JOBSTORE_CLUSTERCHECKININTERVAL, QUARTZ_CLUSTERCHECKININTERVAL));
- properties.setProperty(ORG_QUARTZ_JOBSTORE_ACQUIRETRIGGERSWITHINLOCK,conf.getString(ORG_QUARTZ_JOBSTORE_ACQUIRETRIGGERSWITHINLOCK, QUARTZ_ACQUIRETRIGGERSWITHINLOCK));
- properties.setProperty(ORG_QUARTZ_JOBSTORE_DATASOURCE,conf.getString(ORG_QUARTZ_JOBSTORE_DATASOURCE, QUARTZ_DATASOURCE));
- properties.setProperty(ORG_QUARTZ_DATASOURCE_MYDS_CONNECTIONPROVIDER_CLASS,conf.getString(ORG_QUARTZ_DATASOURCE_MYDS_CONNECTIONPROVIDER_CLASS,DruidConnectionProvider.class.getName()));
-
- schedulerFactory.initialize(properties);
- scheduler = schedulerFactory.getScheduler();
-
- } catch (SchedulerException e) {
- logger.error(e.getMessage(),e);
- System.exit(1);
+
+ private QuartzExecutors() {
+ try {
+ conf = new PropertiesConfiguration(QUARTZ_PROPERTIES_PATH);
+ init();
+ } catch (ConfigurationException e) {
+ logger.warn("not loaded quartz configuration file, will used default value", e);
+ }
}
- }
-
- /**
- * Whether the scheduler has been started.
- *
- * @throws SchedulerException scheduler exception
- */
- public void start() throws SchedulerException {
- if (!scheduler.isStarted()){
- scheduler.start();
- logger.info("Quartz service started" );
+ /**
+ * thread safe and performance promote
+ *
+ * @return instance of Quartz Executors
+ */
+ public static QuartzExecutors getInstance() {
+ return Holder.instance;
}
- }
-
- /**
- * stop all scheduled tasks
- *
- * Halts the Scheduler's firing of Triggers,
- * and cleans up all resources associated with the Scheduler.
- *
- * The scheduler cannot be re-started.
- * @throws SchedulerException scheduler exception
- */
- public void shutdown() throws SchedulerException {
- if (!scheduler.isShutdown()) {
- // don't wait for the task to complete
- scheduler.shutdown();
- logger.info("Quartz service stopped, and halt all tasks");
+
+ /**
+ * init
+ *
+ * Returns a client-usable handle to a Scheduler.
+ */
+ private void init() {
+ try {
+ StdSchedulerFactory schedulerFactory = new StdSchedulerFactory();
+ Properties properties = new Properties();
+
+ String dataSourceDriverClass = org.apache.dolphinscheduler.dao.utils.PropertyUtils.getString(SPRING_DATASOURCE_DRIVER_CLASS_NAME);
+ if (dataSourceDriverClass.equals(ORG_POSTGRESQL_DRIVER)) {
+ properties.setProperty(ORG_QUARTZ_JOBSTORE_DRIVERDELEGATECLASS, conf.getString(ORG_QUARTZ_JOBSTORE_DRIVERDELEGATECLASS, PostgreSQLDelegate.class.getName()));
+ } else {
+ properties.setProperty(ORG_QUARTZ_JOBSTORE_DRIVERDELEGATECLASS, conf.getString(ORG_QUARTZ_JOBSTORE_DRIVERDELEGATECLASS, StdJDBCDelegate.class.getName()));
+ }
+ properties.setProperty(ORG_QUARTZ_SCHEDULER_INSTANCENAME, conf.getString(ORG_QUARTZ_SCHEDULER_INSTANCENAME, QUARTZ_INSTANCENAME));
+ properties.setProperty(ORG_QUARTZ_SCHEDULER_INSTANCEID, conf.getString(ORG_QUARTZ_SCHEDULER_INSTANCEID, QUARTZ_INSTANCEID));
+ properties.setProperty(ORG_QUARTZ_SCHEDULER_MAKESCHEDULERTHREADDAEMON, conf.getString(ORG_QUARTZ_SCHEDULER_MAKESCHEDULERTHREADDAEMON, STRING_TRUE));
+ properties.setProperty(ORG_QUARTZ_JOBSTORE_USEPROPERTIES, conf.getString(ORG_QUARTZ_JOBSTORE_USEPROPERTIES, STRING_FALSE));
+ properties.setProperty(ORG_QUARTZ_THREADPOOL_CLASS, conf.getString(ORG_QUARTZ_THREADPOOL_CLASS, SimpleThreadPool.class.getName()));
+ properties.setProperty(ORG_QUARTZ_THREADPOOL_MAKETHREADSDAEMONS, conf.getString(ORG_QUARTZ_THREADPOOL_MAKETHREADSDAEMONS, STRING_TRUE));
+ properties.setProperty(ORG_QUARTZ_THREADPOOL_THREADCOUNT, conf.getString(ORG_QUARTZ_THREADPOOL_THREADCOUNT, QUARTZ_THREADCOUNT));
+ properties.setProperty(ORG_QUARTZ_THREADPOOL_THREADPRIORITY, conf.getString(ORG_QUARTZ_THREADPOOL_THREADPRIORITY, QUARTZ_THREADPRIORITY));
+ properties.setProperty(ORG_QUARTZ_JOBSTORE_CLASS, conf.getString(ORG_QUARTZ_JOBSTORE_CLASS, JobStoreTX.class.getName()));
+ properties.setProperty(ORG_QUARTZ_JOBSTORE_TABLEPREFIX, conf.getString(ORG_QUARTZ_JOBSTORE_TABLEPREFIX, QUARTZ_TABLE_PREFIX));
+ properties.setProperty(ORG_QUARTZ_JOBSTORE_ISCLUSTERED, conf.getString(ORG_QUARTZ_JOBSTORE_ISCLUSTERED, STRING_TRUE));
+ properties.setProperty(ORG_QUARTZ_JOBSTORE_MISFIRETHRESHOLD, conf.getString(ORG_QUARTZ_JOBSTORE_MISFIRETHRESHOLD, QUARTZ_MISFIRETHRESHOLD));
+ properties.setProperty(ORG_QUARTZ_JOBSTORE_CLUSTERCHECKININTERVAL, conf.getString(ORG_QUARTZ_JOBSTORE_CLUSTERCHECKININTERVAL, QUARTZ_CLUSTERCHECKININTERVAL));
+ properties.setProperty(ORG_QUARTZ_JOBSTORE_ACQUIRETRIGGERSWITHINLOCK, conf.getString(ORG_QUARTZ_JOBSTORE_ACQUIRETRIGGERSWITHINLOCK, QUARTZ_ACQUIRETRIGGERSWITHINLOCK));
+ properties.setProperty(ORG_QUARTZ_JOBSTORE_DATASOURCE, conf.getString(ORG_QUARTZ_JOBSTORE_DATASOURCE, QUARTZ_DATASOURCE));
+ properties.setProperty(ORG_QUARTZ_DATASOURCE_MYDS_CONNECTIONPROVIDER_CLASS, conf.getString(ORG_QUARTZ_DATASOURCE_MYDS_CONNECTIONPROVIDER_CLASS, DruidConnectionProvider.class.getName()));
+
+ schedulerFactory.initialize(properties);
+ scheduler = schedulerFactory.getScheduler();
+
+ } catch (SchedulerException e) {
+ logger.error(e.getMessage(), e);
+ System.exit(1);
+ }
+
}
- }
-
-
- /**
- * add task trigger , if this task already exists, return this task with updated trigger
- *
- * @param clazz job class name
- * @param jobName job name
- * @param jobGroupName job group name
- * @param startDate job start date
- * @param endDate job end date
- * @param cronExpression cron expression
- * @param jobDataMap job parameters data map
- */
- public void addJob(Class clazz,String jobName,String jobGroupName,Date startDate, Date endDate,
- String cronExpression,
- Map jobDataMap) {
- lock.writeLock().lock();
- try {
-
- JobKey jobKey = new JobKey(jobName, jobGroupName);
- JobDetail jobDetail;
- //add a task (if this task already exists, return this task directly)
- if (scheduler.checkExists(jobKey)) {
-
- jobDetail = scheduler.getJobDetail(jobKey);
- if (jobDataMap != null) {
- jobDetail.getJobDataMap().putAll(jobDataMap);
+
+ /**
+ * Whether the scheduler has been started.
+ *
+ * @throws SchedulerException scheduler exception
+ */
+ public void start() throws SchedulerException {
+ if (!scheduler.isStarted()) {
+ scheduler.start();
+ logger.info("Quartz service started");
}
- } else {
- jobDetail = newJob(clazz).withIdentity(jobKey).build();
+ }
- if (jobDataMap != null) {
- jobDetail.getJobDataMap().putAll(jobDataMap);
+ /**
+ * stop all scheduled tasks
+ *
+ * Halts the Scheduler's firing of Triggers,
+ * and cleans up all resources associated with the Scheduler.
+ *
+ * The scheduler cannot be re-started.
+ *
+ * @throws SchedulerException scheduler exception
+ */
+ public void shutdown() throws SchedulerException {
+ if (!scheduler.isShutdown()) {
+ // don't wait for the task to complete
+ scheduler.shutdown();
+ logger.info("Quartz service stopped, and halt all tasks");
}
+ }
- scheduler.addJob(jobDetail, false, true);
-
- logger.info("Add job, job name: {}, group name: {}",
- jobName, jobGroupName);
- }
-
- TriggerKey triggerKey = new TriggerKey(jobName, jobGroupName);
- /**
- * Instructs the Scheduler that upon a mis-fire
- * situation, the CronTrigger wants to have it's
- * next-fire-time updated to the next time in the schedule after the
- * current time (taking into account any associated Calendar),
- * but it does not want to be fired now.
- */
- CronTrigger cronTrigger = newTrigger().withIdentity(triggerKey).startAt(startDate).endAt(endDate)
- .withSchedule(cronSchedule(cronExpression).withMisfireHandlingInstructionDoNothing())
- .forJob(jobDetail).build();
-
- if (scheduler.checkExists(triggerKey)) {
- // updateProcessInstance scheduler trigger when scheduler cycle changes
- CronTrigger oldCronTrigger = (CronTrigger) scheduler.getTrigger(triggerKey);
- String oldCronExpression = oldCronTrigger.getCronExpression();
-
- if (!StringUtils.equalsIgnoreCase(cronExpression,oldCronExpression)) {
- // reschedule job trigger
- scheduler.rescheduleJob(triggerKey, cronTrigger);
- logger.info("reschedule job trigger, triggerName: {}, triggerGroupName: {}, cronExpression: {}, startDate: {}, endDate: {}",
- jobName, jobGroupName, cronExpression, startDate, endDate);
- }
- } else {
- scheduler.scheduleJob(cronTrigger);
- logger.info("schedule job trigger, triggerName: {}, triggerGroupName: {}, cronExpression: {}, startDate: {}, endDate: {}",
- jobName, jobGroupName, cronExpression, startDate, endDate);
- }
-
- } catch (Exception e) {
- logger.error("add job failed", e);
- throw new RuntimeException("add job failed", e);
- } finally {
- lock.writeLock().unlock();
+ /**
+ * add task trigger , if this task already exists, return this task with updated trigger
+ *
+ * @param clazz job class name
+ * @param jobName job name
+ * @param jobGroupName job group name
+ * @param startDate job start date
+ * @param endDate job end date
+ * @param cronExpression cron expression
+ * @param jobDataMap job parameters data map
+ */
+ public void addJob(Class clazz, String jobName, String jobGroupName, Date startDate, Date endDate,
+ String cronExpression,
+ Map jobDataMap) {
+ lock.writeLock().lock();
+ try {
+
+ JobKey jobKey = new JobKey(jobName, jobGroupName);
+ JobDetail jobDetail;
+ //add a task (if this task already exists, return this task directly)
+ if (scheduler.checkExists(jobKey)) {
+
+ jobDetail = scheduler.getJobDetail(jobKey);
+ if (jobDataMap != null) {
+ jobDetail.getJobDataMap().putAll(jobDataMap);
+ }
+ } else {
+ jobDetail = newJob(clazz).withIdentity(jobKey).build();
+
+ if (jobDataMap != null) {
+ jobDetail.getJobDataMap().putAll(jobDataMap);
+ }
+
+ scheduler.addJob(jobDetail, false, true);
+
+ logger.info("Add job, job name: {}, group name: {}",
+ jobName, jobGroupName);
+ }
+
+ TriggerKey triggerKey = new TriggerKey(jobName, jobGroupName);
+ /**
+ * Instructs the Scheduler that upon a mis-fire
+ * situation, the CronTrigger wants to have it's
+ * next-fire-time updated to the next time in the schedule after the
+ * current time (taking into account any associated Calendar),
+ * but it does not want to be fired now.
+ */
+ CronTrigger cronTrigger = newTrigger().withIdentity(triggerKey).startAt(startDate).endAt(endDate)
+ .withSchedule(cronSchedule(cronExpression).withMisfireHandlingInstructionDoNothing())
+ .forJob(jobDetail).build();
+
+ if (scheduler.checkExists(triggerKey)) {
+ // updateProcessInstance scheduler trigger when scheduler cycle changes
+ CronTrigger oldCronTrigger = (CronTrigger) scheduler.getTrigger(triggerKey);
+ String oldCronExpression = oldCronTrigger.getCronExpression();
+
+ if (!StringUtils.equalsIgnoreCase(cronExpression, oldCronExpression)) {
+ // reschedule job trigger
+ scheduler.rescheduleJob(triggerKey, cronTrigger);
+ logger.info("reschedule job trigger, triggerName: {}, triggerGroupName: {}, cronExpression: {}, startDate: {}, endDate: {}",
+ jobName, jobGroupName, cronExpression, startDate, endDate);
+ }
+ } else {
+ scheduler.scheduleJob(cronTrigger);
+ logger.info("schedule job trigger, triggerName: {}, triggerGroupName: {}, cronExpression: {}, startDate: {}, endDate: {}",
+ jobName, jobGroupName, cronExpression, startDate, endDate);
+ }
+
+ } catch (Exception e) {
+ throw new ServiceException("add job failed", e);
+ } finally {
+ lock.writeLock().unlock();
+ }
}
- }
-
-
- /**
- * delete job
- *
- * @param jobName job name
- * @param jobGroupName job group name
- * @return true if the Job was found and deleted.
- */
- public boolean deleteJob(String jobName, String jobGroupName) {
- lock.writeLock().lock();
- try {
- JobKey jobKey = new JobKey(jobName,jobGroupName);
- if(scheduler.checkExists(jobKey)){
- logger.info("try to delete job, job name: {}, job group name: {},", jobName, jobGroupName);
- return scheduler.deleteJob(jobKey);
- }else {
- return true;
- }
-
- } catch (SchedulerException e) {
- logger.error("delete job : {} failed",jobName, e);
- } finally {
- lock.writeLock().unlock();
+
+ /**
+ * delete job
+ *
+ * @param jobName job name
+ * @param jobGroupName job group name
+ * @return true if the Job was found and deleted.
+ */
+ public boolean deleteJob(String jobName, String jobGroupName) {
+ lock.writeLock().lock();
+ try {
+ JobKey jobKey = new JobKey(jobName, jobGroupName);
+ if (scheduler.checkExists(jobKey)) {
+ logger.info("try to delete job, job name: {}, job group name: {},", jobName, jobGroupName);
+ return scheduler.deleteJob(jobKey);
+ } else {
+ return true;
+ }
+
+ } catch (SchedulerException e) {
+ logger.error("delete job : {} failed", jobName, e);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ return false;
+ }
+
+ /**
+ * delete all jobs in job group
+ *
+ * @param jobGroupName job group name
+ * @return true if all of the Jobs were found and deleted, false if
+ * one or more were not deleted.
+ */
+ public boolean deleteAllJobs(String jobGroupName) {
+ lock.writeLock().lock();
+ try {
+ logger.info("try to delete all jobs in job group: {}", jobGroupName);
+ List jobKeys = new ArrayList<>();
+ jobKeys.addAll(scheduler.getJobKeys(GroupMatcher.groupEndsWith(jobGroupName)));
+
+ return scheduler.deleteJobs(jobKeys);
+ } catch (SchedulerException e) {
+ logger.error("delete all jobs in job group: {} failed", jobGroupName, e);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ return false;
}
- return false;
- }
-
- /**
- * delete all jobs in job group
- *
- * @param jobGroupName job group name
- *
- * @return true if all of the Jobs were found and deleted, false if
- * one or more were not deleted.
- */
- public boolean deleteAllJobs(String jobGroupName) {
- lock.writeLock().lock();
- try {
- logger.info("try to delete all jobs in job group: {}", jobGroupName);
- List jobKeys = new ArrayList<>();
- jobKeys.addAll(scheduler.getJobKeys(GroupMatcher.groupEndsWith(jobGroupName)));
-
- return scheduler.deleteJobs(jobKeys);
- } catch (SchedulerException e) {
- logger.error("delete all jobs in job group: {} failed",jobGroupName, e);
- } finally {
- lock.writeLock().unlock();
+
+ /**
+ * build job name
+ *
+ * @param processId process id
+ * @return job name
+ */
+ public static String buildJobName(int processId) {
+ StringBuilder sb = new StringBuilder(30);
+ sb.append(QUARTZ_JOB_PRIFIX).append(UNDERLINE).append(processId);
+ return sb.toString();
+ }
+
+ /**
+ * build job group name
+ *
+ * @param projectId project id
+ * @return job group name
+ */
+ public static String buildJobGroupName(int projectId) {
+ StringBuilder sb = new StringBuilder(30);
+ sb.append(QUARTZ_JOB_GROUP_PRIFIX).append(UNDERLINE).append(projectId);
+ return sb.toString();
+ }
+
+ /**
+ * add params to map
+ *
+ * @param projectId project id
+ * @param scheduleId schedule id
+ * @param schedule schedule
+ * @return data map
+ */
+ public static Map buildDataMap(int projectId, int scheduleId, Schedule schedule) {
+ Map dataMap = new HashMap<>(3);
+ dataMap.put(PROJECT_ID, projectId);
+ dataMap.put(SCHEDULE_ID, scheduleId);
+ dataMap.put(SCHEDULE, JSONUtils.toJsonString(schedule));
+
+ return dataMap;
}
- return false;
- }
-
- /**
- * build job name
- * @param processId process id
- * @return job name
- */
- public static String buildJobName(int processId) {
- StringBuilder sb = new StringBuilder(30);
- sb.append(QUARTZ_JOB_PRIFIX).append(UNDERLINE).append(processId);
- return sb.toString();
- }
-
- /**
- * build job group name
- * @param projectId project id
- * @return job group name
- */
- public static String buildJobGroupName(int projectId) {
- StringBuilder sb = new StringBuilder(30);
- sb.append(QUARTZ_JOB_GROUP_PRIFIX).append(UNDERLINE).append(projectId);
- return sb.toString();
- }
-
- /**
- * add params to map
- *
- * @param projectId project id
- * @param scheduleId schedule id
- * @param schedule schedule
- * @return data map
- */
- public static Map buildDataMap(int projectId, int scheduleId, Schedule schedule) {
- Map dataMap = new HashMap<>(3);
- dataMap.put(PROJECT_ID, projectId);
- dataMap.put(SCHEDULE_ID, scheduleId);
- dataMap.put(SCHEDULE, JSONUtils.toJsonString(schedule));
-
- return dataMap;
- }
}
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/cron/AbstractCycle.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/cron/AbstractCycle.java
index 0a2e31b610f53eb68af858cf1ca229a914864b5a..60c862340bfe4ea6894e851ab12f6f1294b88253 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/cron/AbstractCycle.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/cron/AbstractCycle.java
@@ -14,159 +14,177 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.service.quartz.cron;
+import org.apache.dolphinscheduler.common.enums.CycleEnum;
+
import com.cronutils.model.Cron;
import com.cronutils.model.field.CronField;
import com.cronutils.model.field.CronFieldName;
-import com.cronutils.model.field.expression.*;
-import org.apache.dolphinscheduler.common.enums.CycleEnum;
+import com.cronutils.model.field.expression.Always;
+import com.cronutils.model.field.expression.And;
+import com.cronutils.model.field.expression.Between;
+import com.cronutils.model.field.expression.Every;
+import com.cronutils.model.field.expression.FieldExpression;
+import com.cronutils.model.field.expression.On;
/**
* Cycle
*/
public abstract class AbstractCycle {
- protected Cron cron;
-
- protected CronField minField;
- protected CronField hourField;
- protected CronField dayOfMonthField;
- protected CronField dayOfWeekField;
- protected CronField monthField;
- protected CronField yearField;
-
- public CycleLinks addCycle(AbstractCycle cycle) {
- return new CycleLinks(this.cron).addCycle(this).addCycle(cycle);
- }
-
- /**
- * cycle constructor
- * @param cron cron
- */
- public AbstractCycle(Cron cron) {
- if (cron == null) {
- throw new IllegalArgumentException("cron must not be null!");
+ protected Cron cron;
+
+ protected CronField minField;
+ protected CronField hourField;
+ protected CronField dayOfMonthField;
+ protected CronField dayOfWeekField;
+ protected CronField monthField;
+ protected CronField yearField;
+
+ public CycleLinks addCycle(AbstractCycle cycle) {
+ return new CycleLinks(this.cron).addCycle(this).addCycle(cycle);
+ }
+
+ /**
+ * cycle constructor
+ *
+ * @param cron cron
+ */
+ protected AbstractCycle(Cron cron) {
+ if (cron == null) {
+ throw new IllegalArgumentException("cron must not be null!");
+ }
+
+ this.cron = cron;
+ this.minField = cron.retrieve(CronFieldName.MINUTE);
+ this.hourField = cron.retrieve(CronFieldName.HOUR);
+ this.dayOfMonthField = cron.retrieve(CronFieldName.DAY_OF_MONTH);
+ this.dayOfWeekField = cron.retrieve(CronFieldName.DAY_OF_WEEK);
+ this.monthField = cron.retrieve(CronFieldName.MONTH);
+ this.yearField = cron.retrieve(CronFieldName.YEAR);
+ }
+
+ /**
+ * whether the minute field has a value
+ *
+ * @return if minute field has a value return true，else return false
+ */
+ protected boolean minFiledIsSetAll() {
+ FieldExpression minFieldExpression = minField.getExpression();
+ return (minFieldExpression instanceof Every || minFieldExpression instanceof Always
+ || minFieldExpression instanceof Between || minFieldExpression instanceof And
+ || minFieldExpression instanceof On);
+ }
+
+ /**
+ * whether the minute field has a value of every or always
+ *
+ * @return if minute field has a value of every or always return true，else return false
+ */
+ protected boolean minFiledIsEvery() {
+ FieldExpression minFieldExpression = minField.getExpression();
+ return (minFieldExpression instanceof Every || minFieldExpression instanceof Always);
+ }
+
+ /**
+ * whether the hour field has a value
+ *
+ * @return if hour field has a value return true，else return false
+ */
+ protected boolean hourFiledIsSetAll() {
+ FieldExpression hourFieldExpression = hourField.getExpression();
+ return (hourFieldExpression instanceof Every || hourFieldExpression instanceof Always
+ || hourFieldExpression instanceof Between || hourFieldExpression instanceof And
+ || hourFieldExpression instanceof On);
+ }
+
+ /**
+ * whether the hour field has a value of every or always
+ *
+ * @return if hour field has a value of every or always return true，else return false
+ */
+ protected boolean hourFiledIsEvery() {
+ FieldExpression hourFieldExpression = hourField.getExpression();
+ return (hourFieldExpression instanceof Every || hourFieldExpression instanceof Always);
+ }
+
+ /**
+ * whether the day Of month field has a value
+ *
+ * @return if day Of month field has a value return true，else return false
+ */
+ protected boolean dayOfMonthFieldIsSetAll() {
+ return (dayOfMonthField.getExpression() instanceof Every || dayOfMonthField.getExpression() instanceof Always
+ || dayOfMonthField.getExpression() instanceof Between || dayOfMonthField.getExpression() instanceof And
+ || dayOfMonthField.getExpression() instanceof On);
+ }
+
+ /**
+ * whether the day Of Month field has a value of every or always
+ *
+ * @return if day Of Month field has a value of every or always return true，else return false
+ */
+ protected boolean dayOfMonthFieldIsEvery() {
+ return (dayOfMonthField.getExpression() instanceof Every || dayOfMonthField.getExpression() instanceof Always);
+ }
+
+ /**
+ * whether month field has a value
+ *
+ * @return if month field has a value return true，else return false
+ */
+ protected boolean monthFieldIsSetAll() {
+ FieldExpression monthFieldExpression = monthField.getExpression();
+ return (monthFieldExpression instanceof Every || monthFieldExpression instanceof Always
+ || monthFieldExpression instanceof Between || monthFieldExpression instanceof And
+ || monthFieldExpression instanceof On);
+ }
+
+ /**
+ * whether the month field has a value of every or always
+ *
+ * @return if month field has a value of every or always return true，else return false
+ */
+ protected boolean monthFieldIsEvery() {
+ FieldExpression monthFieldExpression = monthField.getExpression();
+ return (monthFieldExpression instanceof Every || monthFieldExpression instanceof Always);
+ }
+
+ /**
+ * whether the day Of week field has a value
+ *
+ * @return if day Of week field has a value return true，else return false
+ */
+ protected boolean dayofWeekFieldIsSetAll() {
+ FieldExpression dayOfWeekFieldExpression = dayOfWeekField.getExpression();
+ return (dayOfWeekFieldExpression instanceof Every || dayOfWeekFieldExpression instanceof Always
+ || dayOfWeekFieldExpression instanceof Between || dayOfWeekFieldExpression instanceof And
+ || dayOfWeekFieldExpression instanceof On);
+ }
+
+ /**
+ * whether the day Of week field has a value of every or always
+ *
+ * @return if day Of week field has a value of every or always return true，else return false
+ */
+ protected boolean dayofWeekFieldIsEvery() {
+ FieldExpression dayOfWeekFieldExpression = dayOfWeekField.getExpression();
+ return (dayOfWeekFieldExpression instanceof Every || dayOfWeekFieldExpression instanceof Always);
}
- this.cron = cron;
- this.minField = cron.retrieve(CronFieldName.MINUTE);
- this.hourField = cron.retrieve(CronFieldName.HOUR);
- this.dayOfMonthField = cron.retrieve(CronFieldName.DAY_OF_MONTH);
- this.dayOfWeekField = cron.retrieve(CronFieldName.DAY_OF_WEEK);
- this.monthField = cron.retrieve(CronFieldName.MONTH);
- this.yearField = cron.retrieve(CronFieldName.YEAR);
- }
-
- /**
- * whether the minute field has a value
- * @return if minute field has a value return true，else return false
- */
- protected boolean minFiledIsSetAll(){
- FieldExpression minFieldExpression = minField.getExpression();
- return (minFieldExpression instanceof Every || minFieldExpression instanceof Always
- || minFieldExpression instanceof Between || minFieldExpression instanceof And
- || minFieldExpression instanceof On);
- }
-
-
- /**
- * whether the minute field has a value of every or always
- * @return if minute field has a value of every or always return true，else return false
- */
- protected boolean minFiledIsEvery(){
- FieldExpression minFieldExpression = minField.getExpression();
- return (minFieldExpression instanceof Every || minFieldExpression instanceof Always);
- }
-
- /**
- * whether the hour field has a value
- * @return if hour field has a value return true，else return false
- */
- protected boolean hourFiledIsSetAll(){
- FieldExpression hourFieldExpression = hourField.getExpression();
- return (hourFieldExpression instanceof Every || hourFieldExpression instanceof Always
- || hourFieldExpression instanceof Between || hourFieldExpression instanceof And
- || hourFieldExpression instanceof On);
- }
-
- /**
- * whether the hour field has a value of every or always
- * @return if hour field has a value of every or always return true，else return false
- */
- protected boolean hourFiledIsEvery(){
- FieldExpression hourFieldExpression = hourField.getExpression();
- return (hourFieldExpression instanceof Every || hourFieldExpression instanceof Always);
- }
-
- /**
- * whether the day Of month field has a value
- * @return if day Of month field has a value return true，else return false
- */
- protected boolean dayOfMonthFieldIsSetAll(){
- return (dayOfMonthField.getExpression() instanceof Every || dayOfMonthField.getExpression() instanceof Always
- || dayOfMonthField.getExpression() instanceof Between || dayOfMonthField.getExpression() instanceof And
- || dayOfMonthField.getExpression() instanceof On);
- }
-
-
- /**
- * whether the day Of Month field has a value of every or always
- * @return if day Of Month field has a value of every or always return true，else return false
- */
- protected boolean dayOfMonthFieldIsEvery(){
- return (dayOfMonthField.getExpression() instanceof Every || dayOfMonthField.getExpression() instanceof Always);
- }
-
- /**
- * whether month field has a value
- * @return if month field has a value return true，else return false
- */
- protected boolean monthFieldIsSetAll(){
- FieldExpression monthFieldExpression = monthField.getExpression();
- return (monthFieldExpression instanceof Every || monthFieldExpression instanceof Always
- || monthFieldExpression instanceof Between || monthFieldExpression instanceof And
- || monthFieldExpression instanceof On);
- }
-
- /**
- * whether the month field has a value of every or always
- * @return if month field has a value of every or always return true，else return false
- */
- protected boolean monthFieldIsEvery(){
- FieldExpression monthFieldExpression = monthField.getExpression();
- return (monthFieldExpression instanceof Every || monthFieldExpression instanceof Always);
- }
-
- /**
- * whether the day Of week field has a value
- * @return if day Of week field has a value return true，else return false
- */
- protected boolean dayofWeekFieldIsSetAll(){
- FieldExpression dayOfWeekFieldExpression = dayOfWeekField.getExpression();
- return (dayOfWeekFieldExpression instanceof Every || dayOfWeekFieldExpression instanceof Always
- || dayOfWeekFieldExpression instanceof Between || dayOfWeekFieldExpression instanceof And
- || dayOfWeekFieldExpression instanceof On);
- }
-
- /**
- * whether the day Of week field has a value of every or always
- * @return if day Of week field has a value of every or always return true，else return false
- */
- protected boolean dayofWeekFieldIsEvery(){
- FieldExpression dayOfWeekFieldExpression = dayOfWeekField.getExpression();
- return (dayOfWeekFieldExpression instanceof Every || dayOfWeekFieldExpression instanceof Always);
- }
-
- /**
- * get cycle enum
- * @return CycleEnum
- */
- protected abstract CycleEnum getCycle();
-
- /**
- * get mini level cycle enum
- * @return CycleEnum
- */
- protected abstract CycleEnum getMiniCycle();
+ /**
+ * get cycle enum
+ *
+ * @return CycleEnum
+ */
+ protected abstract CycleEnum getCycle();
+
+ /**
+ * get mini level cycle enum
+ *
+ * @return CycleEnum
+ */
+ protected abstract CycleEnum getMiniCycle();
}
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/AbstractZKClient.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/AbstractZKClient.java
index 8a7d891c2ef58613739e144d1ff309ded6057c50..37d8f10c93d8cd1dc9802e0184bd15480543e240 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/AbstractZKClient.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/AbstractZKClient.java
@@ -14,322 +14,329 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.service.zk;
-import org.apache.curator.framework.recipes.locks.InterProcessMutex;
+import static org.apache.dolphinscheduler.common.Constants.ADD_ZK_OP;
+import static org.apache.dolphinscheduler.common.Constants.COLON;
+import static org.apache.dolphinscheduler.common.Constants.DELETE_ZK_OP;
+import static org.apache.dolphinscheduler.common.Constants.DIVISION_STRING;
+import static org.apache.dolphinscheduler.common.Constants.MASTER_PREFIX;
+import static org.apache.dolphinscheduler.common.Constants.SINGLE_SLASH;
+import static org.apache.dolphinscheduler.common.Constants.UNDERLINE;
+import static org.apache.dolphinscheduler.common.Constants.WORKER_PREFIX;
+
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.ZKNodeType;
import org.apache.dolphinscheduler.common.model.Server;
import org.apache.dolphinscheduler.common.utils.ResInfo;
import org.apache.dolphinscheduler.common.utils.StringUtils;
+
+import org.apache.curator.framework.recipes.locks.InterProcessMutex;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
-import java.util.*;
-
-import static org.apache.dolphinscheduler.common.Constants.*;
-
/**
* abstract zookeeper client
*/
@Component
public abstract class AbstractZKClient extends ZookeeperCachedOperator {
- private static final Logger logger = LoggerFactory.getLogger(AbstractZKClient.class);
-
-
- /**
- * remove dead server by host
- * @param host host
- * @param serverType serverType
- * @throws Exception
- */
- public void removeDeadServerByHost(String host, String serverType) throws Exception {
- List deadServers = super.getChildrenKeys(getDeadZNodeParentPath());
- for(String serverPath : deadServers){
- if(serverPath.startsWith(serverType+UNDERLINE+host)){
- String server = getDeadZNodeParentPath() + SINGLE_SLASH + serverPath;
- super.remove(server);
- logger.info("{} server {} deleted from zk dead server path success" , serverType , host);
- }
- }
- }
-
-
- /**
- * opType(add): if find dead server , then add to zk deadServerPath
- * opType(delete): delete path from zk
- *
- * @param zNode node path
- * @param zkNodeType master or worker
- * @param opType delete or add
- * @throws Exception errors
- */
- public void handleDeadServer(String zNode, ZKNodeType zkNodeType, String opType) throws Exception {
- String host = getHostByEventDataPath(zNode);
- String type = (zkNodeType == ZKNodeType.MASTER) ? MASTER_PREFIX : WORKER_PREFIX;
-
- //check server restart, if restart , dead server path in zk should be delete
- if(opType.equals(DELETE_ZK_OP)){
- removeDeadServerByHost(host, type);
-
- }else if(opType.equals(ADD_ZK_OP)){
- String deadServerPath = getDeadZNodeParentPath() + SINGLE_SLASH + type + UNDERLINE + host;
- if(!super.isExisted(deadServerPath)){
- //add dead server info to zk dead server path : /dead-servers/
-
- super.persist(deadServerPath,(type + UNDERLINE + host));
-
- logger.info("{} server dead , and {} added to zk dead server path success" ,
- zkNodeType.toString(), zNode);
- }
- }
-
- }
-
- /**
- * get active master num
- * @return active master number
- */
- public int getActiveMasterNum(){
- List childrenList = new ArrayList<>();
- try {
- // read master node parent path from conf
- if(super.isExisted(getZNodeParentPath(ZKNodeType.MASTER))){
- childrenList = super.getChildrenKeys(getZNodeParentPath(ZKNodeType.MASTER));
- }
- } catch (Exception e) {
- logger.error("getActiveMasterNum error",e);
- }
- return childrenList.size();
- }
-
- /**
- *
- * @return zookeeper quorum
- */
- public String getZookeeperQuorum(){
- return getZookeeperConfig().getServerList();
- }
-
- /**
- * get server list.
- * @param zkNodeType zookeeper node type
- * @return server list
- */
- public List getServersList(ZKNodeType zkNodeType){
- Map masterMap = getServerMaps(zkNodeType);
- String parentPath = getZNodeParentPath(zkNodeType);
-
- List masterServers = new ArrayList<>();
- for (Map.Entry entry : masterMap.entrySet()) {
- Server masterServer = ResInfo.parseHeartbeatForZKInfo(entry.getValue());
- if(masterServer == null){
- continue;
- }
- String key = entry.getKey();
- masterServer.setZkDirectory(parentPath + "/"+ key);
- //set host and port
- String[] hostAndPort=key.split(COLON);
- String[] hosts=hostAndPort[0].split(DIVISION_STRING);
- // fetch the last one
- masterServer.setHost(hosts[hosts.length-1]);
- masterServer.setPort(Integer.parseInt(hostAndPort[1]));
- masterServers.add(masterServer);
- }
- return masterServers;
- }
-
- /**
- * get master server list map.
- * @param zkNodeType zookeeper node type
- * @return result : {host : resource info}
- */
- public Map getServerMaps(ZKNodeType zkNodeType){
-
- Map masterMap = new HashMap<>();
- try {
- String path = getZNodeParentPath(zkNodeType);
- List serverList = super.getChildrenKeys(path);
- if(zkNodeType == ZKNodeType.WORKER){
- List workerList = new ArrayList<>();
- for(String group : serverList){
- List groupServers = super.getChildrenKeys(path + Constants.SLASH + group);
- for(String groupServer : groupServers){
- workerList.add(group + Constants.SLASH + groupServer);
- }
- }
- serverList = workerList;
- }
- for(String server : serverList){
- masterMap.putIfAbsent(server, super.get(path + Constants.SLASH + server));
- }
- } catch (Exception e) {
- logger.error("get server list failed", e);
- }
-
- return masterMap;
- }
-
- /**
- * check the zookeeper node already exists
- * @param host host
- * @param zkNodeType zookeeper node type
- * @return true if exists
- */
- public boolean checkZKNodeExists(String host, ZKNodeType zkNodeType) {
- String path = getZNodeParentPath(zkNodeType);
- if(StringUtils.isEmpty(path)){
- logger.error("check zk node exists error, host:{}, zk node type:{}",
- host, zkNodeType.toString());
- return false;
- }
- Map serverMaps = getServerMaps(zkNodeType);
- for(String hostKey : serverMaps.keySet()){
- if(hostKey.contains(host)){
- return true;
- }
- }
- return false;
- }
-
- /**
- *
- * @return get worker node parent path
- */
- protected String getWorkerZNodeParentPath(){
- return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_WORKERS;
- }
-
- /**
- *
- * @return get master node parent path
- */
- protected String getMasterZNodeParentPath(){
- return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_MASTERS;
- }
-
- /**
- *
- * @return get master lock path
- */
- public String getMasterLockPath(){
- return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_LOCK_MASTERS;
- }
-
- /**
- *
- * @param zkNodeType zookeeper node type
- * @return get zookeeper node parent path
- */
- public String getZNodeParentPath(ZKNodeType zkNodeType) {
- String path = "";
- switch (zkNodeType){
- case MASTER:
- return getMasterZNodeParentPath();
- case WORKER:
- return getWorkerZNodeParentPath();
- case DEAD_SERVER:
- return getDeadZNodeParentPath();
- default:
- break;
- }
- return path;
- }
-
- /**
- *
- * @return get dead server node parent path
- */
- protected String getDeadZNodeParentPath(){
- return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_DEAD_SERVERS;
- }
-
- /**
- *
- * @return get master start up lock path
- */
- public String getMasterStartUpLockPath(){
- return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_LOCK_FAILOVER_STARTUP_MASTERS;
- }
-
- /**
- *
- * @return get master failover lock path
- */
- public String getMasterFailoverLockPath(){
- return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_LOCK_FAILOVER_MASTERS;
- }
-
- /**
- *
- * @return get worker failover lock path
- */
- public String getWorkerFailoverLockPath(){
- return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_LOCK_FAILOVER_WORKERS;
- }
-
- /**
- * release mutex
- * @param mutex mutex
- */
- public void releaseMutex(InterProcessMutex mutex) {
- if (mutex != null){
- try {
- mutex.release();
- } catch (Exception e) {
- if("instance must be started before calling this method".equals(e.getMessage())){
- logger.warn("lock release");
- }else{
- logger.error("lock release failed",e);
- }
-
- }
- }
- }
-
- /**
- * init system znode
- */
- protected void initSystemZNode(){
- try {
- persist(getMasterZNodeParentPath(), "");
- persist(getWorkerZNodeParentPath(), "");
- persist(getDeadZNodeParentPath(), "");
-
- logger.info("initialize server nodes success.");
- } catch (Exception e) {
- logger.error("init system znode failed",e);
- }
- }
-
- /**
- * get host ip, string format: masterParentPath/ip
- * @param path path
- * @return host ip, string format: masterParentPath/ip
- */
- protected String getHostByEventDataPath(String path) {
- if(StringUtils.isEmpty(path)){
- logger.error("empty path!");
- return "";
- }
- String[] pathArray = path.split(SINGLE_SLASH);
- if(pathArray.length < 1){
- logger.error("parse ip error: {}", path);
- return "";
- }
- return pathArray[pathArray.length - 1];
-
- }
-
- @Override
- public String toString() {
- return "AbstractZKClient{" +
- "zkClient=" + getZkClient() +
- ", deadServerZNodeParentPath='" + getZNodeParentPath(ZKNodeType.DEAD_SERVER) + '\'' +
- ", masterZNodeParentPath='" + getZNodeParentPath(ZKNodeType.MASTER) + '\'' +
- ", workerZNodeParentPath='" + getZNodeParentPath(ZKNodeType.WORKER) + '\'' +
- '}';
- }
-}
\ No newline at end of file
+ private static final Logger logger = LoggerFactory.getLogger(AbstractZKClient.class);
+
+ /**
+ * remove dead server by host
+ *
+ * @param host host
+ * @param serverType serverType
+ */
+ public void removeDeadServerByHost(String host, String serverType) {
+ List deadServers = super.getChildrenKeys(getDeadZNodeParentPath());
+ for (String serverPath : deadServers) {
+ if (serverPath.startsWith(serverType + UNDERLINE + host)) {
+ String server = getDeadZNodeParentPath() + SINGLE_SLASH + serverPath;
+ super.remove(server);
+ logger.info("{} server {} deleted from zk dead server path success", serverType, host);
+ }
+ }
+ }
+
+ /**
+ * opType(add): if find dead server , then add to zk deadServerPath
+ * opType(delete): delete path from zk
+ *
+ * @param zNode node path
+ * @param zkNodeType master or worker
+ * @param opType delete or add
+ */
+ public void handleDeadServer(String zNode, ZKNodeType zkNodeType, String opType) {
+ String host = getHostByEventDataPath(zNode);
+ String type = (zkNodeType == ZKNodeType.MASTER) ? MASTER_PREFIX : WORKER_PREFIX;
+
+ //check server restart, if restart , dead server path in zk should be delete
+ if (opType.equals(DELETE_ZK_OP)) {
+ removeDeadServerByHost(host, type);
+
+ } else if (opType.equals(ADD_ZK_OP)) {
+ String deadServerPath = getDeadZNodeParentPath() + SINGLE_SLASH + type + UNDERLINE + host;
+ if (!super.isExisted(deadServerPath)) {
+ //add dead server info to zk dead server path : /dead-servers/
+
+ super.persist(deadServerPath, (type + UNDERLINE + host));
+
+ logger.info("{} server dead , and {} added to zk dead server path success",
+ zkNodeType, zNode);
+ }
+ }
+
+ }
+
+ /**
+ * get active master num
+ *
+ * @return active master number
+ */
+ public int getActiveMasterNum() {
+ List childrenList = new ArrayList<>();
+ try {
+ // read master node parent path from conf
+ if (super.isExisted(getZNodeParentPath(ZKNodeType.MASTER))) {
+ childrenList = super.getChildrenKeys(getZNodeParentPath(ZKNodeType.MASTER));
+ }
+ } catch (Exception e) {
+ logger.error("getActiveMasterNum error", e);
+ }
+ return childrenList.size();
+ }
+
+ /**
+ * @return zookeeper quorum
+ */
+ public String getZookeeperQuorum() {
+ return getZookeeperConfig().getServerList();
+ }
+
+ /**
+ * get server list.
+ *
+ * @param zkNodeType zookeeper node type
+ * @return server list
+ */
+ public List getServersList(ZKNodeType zkNodeType) {
+ Map masterMap = getServerMaps(zkNodeType);
+ String parentPath = getZNodeParentPath(zkNodeType);
+
+ List masterServers = new ArrayList<>();
+ for (Map.Entry entry : masterMap.entrySet()) {
+ Server masterServer = ResInfo.parseHeartbeatForZKInfo(entry.getValue());
+ if (masterServer == null) {
+ continue;
+ }
+ String key = entry.getKey();
+ masterServer.setZkDirectory(parentPath + "/" + key);
+ //set host and port
+ String[] hostAndPort = key.split(COLON);
+ String[] hosts = hostAndPort[0].split(DIVISION_STRING);
+ // fetch the last one
+ masterServer.setHost(hosts[hosts.length - 1]);
+ masterServer.setPort(Integer.parseInt(hostAndPort[1]));
+ masterServers.add(masterServer);
+ }
+ return masterServers;
+ }
+
+ /**
+ * get master server list map.
+ *
+ * @param zkNodeType zookeeper node type
+ * @return result : {host : resource info}
+ */
+ public Map getServerMaps(ZKNodeType zkNodeType) {
+
+ Map masterMap = new HashMap<>();
+ try {
+ String path = getZNodeParentPath(zkNodeType);
+ List serverList = super.getChildrenKeys(path);
+ if (zkNodeType == ZKNodeType.WORKER) {
+ List workerList = new ArrayList<>();
+ for (String group : serverList) {
+ List groupServers = super.getChildrenKeys(path + Constants.SLASH + group);
+ for (String groupServer : groupServers) {
+ workerList.add(group + Constants.SLASH + groupServer);
+ }
+ }
+ serverList = workerList;
+ }
+ for (String server : serverList) {
+ masterMap.putIfAbsent(server, super.get(path + Constants.SLASH + server));
+ }
+ } catch (Exception e) {
+ logger.error("get server list failed", e);
+ }
+
+ return masterMap;
+ }
+
+ /**
+ * check the zookeeper node already exists
+ *
+ * @param host host
+ * @param zkNodeType zookeeper node type
+ * @return true if exists
+ */
+ public boolean checkZKNodeExists(String host, ZKNodeType zkNodeType) {
+ String path = getZNodeParentPath(zkNodeType);
+ if (StringUtils.isEmpty(path)) {
+ logger.error("check zk node exists error, host:{}, zk node type:{}",
+ host, zkNodeType);
+ return false;
+ }
+ Map serverMaps = getServerMaps(zkNodeType);
+ for (String hostKey : serverMaps.keySet()) {
+ if (hostKey.contains(host)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * @return get worker node parent path
+ */
+ protected String getWorkerZNodeParentPath() {
+ return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_WORKERS;
+ }
+
+ /**
+ * @return get master node parent path
+ */
+ protected String getMasterZNodeParentPath() {
+ return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_MASTERS;
+ }
+
+ /**
+ * @return get master lock path
+ */
+ public String getMasterLockPath() {
+ return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_LOCK_MASTERS;
+ }
+
+ /**
+ * @param zkNodeType zookeeper node type
+ * @return get zookeeper node parent path
+ */
+ public String getZNodeParentPath(ZKNodeType zkNodeType) {
+ String path = "";
+ switch (zkNodeType) {
+ case MASTER:
+ return getMasterZNodeParentPath();
+ case WORKER:
+ return getWorkerZNodeParentPath();
+ case DEAD_SERVER:
+ return getDeadZNodeParentPath();
+ default:
+ break;
+ }
+ return path;
+ }
+
+ /**
+ * @return get dead server node parent path
+ */
+ protected String getDeadZNodeParentPath() {
+ return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_DEAD_SERVERS;
+ }
+
+ /**
+ * @return get master start up lock path
+ */
+ public String getMasterStartUpLockPath() {
+ return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_LOCK_FAILOVER_STARTUP_MASTERS;
+ }
+
+ /**
+ * @return get master failover lock path
+ */
+ public String getMasterFailoverLockPath() {
+ return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_LOCK_FAILOVER_MASTERS;
+ }
+
+ /**
+ * @return get worker failover lock path
+ */
+ public String getWorkerFailoverLockPath() {
+ return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_LOCK_FAILOVER_WORKERS;
+ }
+
+ /**
+ * release mutex
+ *
+ * @param mutex mutex
+ */
+ public void releaseMutex(InterProcessMutex mutex) {
+ if (mutex != null) {
+ try {
+ mutex.release();
+ } catch (Exception e) {
+ if ("instance must be started before calling this method".equals(e.getMessage())) {
+ logger.warn("lock release");
+ } else {
+ logger.error("lock release failed", e);
+ }
+
+ }
+ }
+ }
+
+ /**
+ * init system znode
+ */
+ protected void initSystemZNode() {
+ try {
+ persist(getMasterZNodeParentPath(), "");
+ persist(getWorkerZNodeParentPath(), "");
+ persist(getDeadZNodeParentPath(), "");
+
+ logger.info("initialize server nodes success.");
+ } catch (Exception e) {
+ logger.error("init system znode failed", e);
+ }
+ }
+
+ /**
+ * get host ip, string format: masterParentPath/ip
+ *
+ * @param path path
+ * @return host ip, string format: masterParentPath/ip
+ */
+ protected String getHostByEventDataPath(String path) {
+ if (StringUtils.isEmpty(path)) {
+ logger.error("empty path!");
+ return "";
+ }
+ String[] pathArray = path.split(SINGLE_SLASH);
+ if (pathArray.length < 1) {
+ logger.error("parse ip error: {}", path);
+ return "";
+ }
+ return pathArray[pathArray.length - 1];
+
+ }
+
+ @Override
+ public String toString() {
+ return "AbstractZKClient{"
+ + "zkClient=" + getZkClient()
+ + ", deadServerZNodeParentPath='" + getZNodeParentPath(ZKNodeType.DEAD_SERVER) + '\''
+ + ", masterZNodeParentPath='" + getZNodeParentPath(ZKNodeType.MASTER) + '\''
+ + ", workerZNodeParentPath='" + getZNodeParentPath(ZKNodeType.WORKER) + '\''
+ + '}';
+ }
+}
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/CuratorZookeeperClient.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/CuratorZookeeperClient.java
index 5a04c5a23bfc15ab834503cd8f2669283fe8184e..e25a22f03142f438bfd97b39a248490a43136435 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/CuratorZookeeperClient.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/CuratorZookeeperClient.java
@@ -14,9 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.service.zk;
-import org.apache.commons.lang.StringUtils;
+import static org.apache.dolphinscheduler.common.utils.Preconditions.checkNotNull;
+
+import org.apache.dolphinscheduler.common.utils.StringUtils;
+import org.apache.dolphinscheduler.service.exceptions.ServiceException;
+
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLProvider;
@@ -25,18 +30,16 @@ import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
-import java.nio.charset.StandardCharsets;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-
-import static org.apache.dolphinscheduler.common.utils.Preconditions.checkNotNull;
-
/**
* Shared Curator zookeeper client
*/
@@ -49,7 +52,6 @@ public class CuratorZookeeperClient implements InitializingBean {
private CuratorFramework zkClient;
-
@Override
public void afterPropertiesSet() throws Exception {
this.zkClient = buildClient();
@@ -91,7 +93,7 @@ public class CuratorZookeeperClient implements InitializingBean {
zkClient.blockUntilConnected(30, TimeUnit.SECONDS);
} catch (final Exception ex) {
- throw new RuntimeException(ex);
+ throw new ServiceException(ex);
}
return zkClient;
}
@@ -123,4 +125,4 @@ public class CuratorZookeeperClient implements InitializingBean {
public CuratorFramework getZkClient() {
return zkClient;
}
-}
\ No newline at end of file
+}
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZKServer.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZKServer.java
index c7a53ebdc0bf20fe29334cdf3245b7466de6f97e..7ac23a3c4df1cc40802f3c61a0a8a3cf82df1c59 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZKServer.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZKServer.java
@@ -14,19 +14,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.service.zk;
import org.apache.dolphinscheduler.common.utils.StringUtils;
+import org.apache.dolphinscheduler.service.exceptions.ServiceException;
+
import org.apache.zookeeper.server.ZooKeeperServer;
import org.apache.zookeeper.server.ZooKeeperServerMain;
import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* just speed experience version
@@ -51,10 +54,10 @@ public class ZKServer {
ZKServer zkServer;
if (args.length == 0) {
zkServer = new ZKServer();
- } else if (args.length == 1){
- zkServer = new ZKServer(Integer.valueOf(args[0]), "");
+ } else if (args.length == 1) {
+ zkServer = new ZKServer(Integer.parseInt(args[0]), "");
} else {
- zkServer = new ZKServer(Integer.valueOf(args[0]), args[1]);
+ zkServer = new ZKServer(Integer.parseInt(args[0]), args[1]);
}
zkServer.registerHook();
zkServer.start();
@@ -73,7 +76,7 @@ public class ZKServer {
}
private void registerHook() {
- /**
+ /*
* register hooks, which are called before the process exits
*/
Runtime.getRuntime().addShutdownHook(new Thread(this::stop));
@@ -90,7 +93,7 @@ public class ZKServer {
}
}
- public boolean isStarted(){
+ public boolean isStarted() {
return isStarted.get();
}
@@ -119,19 +122,19 @@ public class ZKServer {
if (file.exists()) {
logger.warn("The path of zk server exists");
}
- logger.info("zk server starting, data dir path:{}" , zkDataDir);
- startLocalZkServer(port, zkDataDir, ZooKeeperServer.DEFAULT_TICK_TIME,"60");
+ logger.info("zk server starting, data dir path:{}", zkDataDir);
+ startLocalZkServer(port, zkDataDir, ZooKeeperServer.DEFAULT_TICK_TIME, "60");
}
/**
* Starts a local Zk instance
*
- * @param port The port to listen on
+ * @param port The port to listen on
* @param dataDirPath The path for the Zk data directory
- * @param tickTime zk tick time
- * @param maxClientCnxns zk max client connections
+ * @param tickTime zk tick time
+ * @param maxClientCnxns zk max client connections
*/
- private void startLocalZkServer(final int port, final String dataDirPath,final int tickTime,String maxClientCnxns) {
+ private void startLocalZkServer(final int port, final String dataDirPath, final int tickTime, String maxClientCnxns) {
if (isStarted.compareAndSet(false, true)) {
zooKeeperServerMain = new PublicZooKeeperServerMain();
logger.info("Zookeeper data path : {} ", dataDirPath);
@@ -144,8 +147,7 @@ public class ZKServer {
zooKeeperServerMain.initializeAndRun(args);
} catch (QuorumPeerConfig.ConfigException | IOException e) {
- logger.warn("Caught exception while starting ZK", e);
- throw new RuntimeException(e);
+ throw new ServiceException("Caught exception while starting ZK", e);
}
}
}
@@ -159,7 +161,7 @@ public class ZKServer {
logger.info("zk server stopped");
} catch (Exception e) {
- logger.error("Failed to stop ZK ",e);
+ logger.error("Failed to stop ZK ", e);
}
}
@@ -180,8 +182,7 @@ public class ZKServer {
org.apache.commons.io.FileUtils.deleteDirectory(new File(dataDir));
}
} catch (Exception e) {
- logger.warn("Caught exception while stopping ZK server", e);
- throw new RuntimeException(e);
+ throw new ServiceException("Caught exception while starting ZK", e);
}
}
}
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZookeeperCachedOperator.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZookeeperCachedOperator.java
index 6dfce79a3aa3fcf62af3aed2e3e99988f36dcceb..88c339b04529a2823c3d88a36e8fbf174347e37b 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZookeeperCachedOperator.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZookeeperCachedOperator.java
@@ -14,21 +14,24 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.service.zk;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
+import org.apache.dolphinscheduler.service.exceptions.ServiceException;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
+
+import java.nio.charset.StandardCharsets;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
-import java.nio.charset.StandardCharsets;
-
@Component
public class ZookeeperCachedOperator extends ZookeeperOperator {
@@ -36,6 +39,7 @@ public class ZookeeperCachedOperator extends ZookeeperOperator {
private TreeCache treeCache;
+
/**
* register a unified listener of /${dsRoot},
*/
@@ -59,14 +63,16 @@ public class ZookeeperCachedOperator extends ZookeeperOperator {
treeCache.start();
} catch (Exception e) {
logger.error("add listener to zk path: {} failed", getZookeeperConfig().getDsRoot());
- throw new RuntimeException(e);
+ throw new ServiceException(e);
}
}
//for sub class
- protected void dataChanged(final CuratorFramework client, final TreeCacheEvent event, final String path){}
+ protected void dataChanged(final CuratorFramework client, final TreeCacheEvent event, final String path) {
+ // Used by sub class
+ }
- public String getFromCache(final String cachePath, final String key) {
+ public String getFromCache(final String key) {
ChildData resultInCache = treeCache.getCurrentData(key);
if (null != resultInCache) {
return null == resultInCache.getData() ? null : new String(resultInCache.getData(), StandardCharsets.UTF_8);
@@ -74,11 +80,11 @@ public class ZookeeperCachedOperator extends ZookeeperOperator {
return null;
}
- public TreeCache getTreeCache(final String cachePath) {
+ public TreeCache getTreeCache() {
return treeCache;
}
- public void addListener(TreeCacheListener listener){
+ public void addListener(TreeCacheListener listener) {
this.treeCache.getListenable().addListener(listener);
}
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZookeeperOperator.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZookeeperOperator.java
index e7b049f8bf2ff49174d701e657882e828113efd8..8a219837b788e581c106f35aa2c75e5c4424105a 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZookeeperOperator.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZookeeperOperator.java
@@ -14,13 +14,17 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.service.zk;
-import org.apache.commons.lang.StringUtils;
+import static org.apache.dolphinscheduler.common.utils.Preconditions.checkNotNull;
+
+import org.apache.dolphinscheduler.common.utils.StringUtils;
+import org.apache.dolphinscheduler.service.exceptions.ServiceException;
+
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLProvider;
-import org.apache.curator.framework.api.transaction.CuratorOp;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
@@ -29,18 +33,16 @@ import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
-import java.nio.charset.StandardCharsets;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-
-import static org.apache.dolphinscheduler.common.utils.Preconditions.checkNotNull;
-
/**
* zk base operator
*/
@@ -64,19 +66,23 @@ public class ZookeeperOperator implements InitializingBean {
/**
* this method is for sub class,
*/
- protected void registerListener(){}
+ protected void registerListener() {
+ // Used by sub class
+ }
- protected void treeCacheStart(){}
+ protected void treeCacheStart() {
+ // Used by sub class
+ }
public void initStateLister() {
checkNotNull(zkClient);
zkClient.getConnectionStateListenable().addListener((client, newState) -> {
- if(newState == ConnectionState.LOST){
+ if (newState == ConnectionState.LOST) {
logger.error("connection lost from zookeeper");
- } else if(newState == ConnectionState.RECONNECTED){
+ } else if (newState == ConnectionState.RECONNECTED) {
logger.info("reconnected to zookeeper");
- } else if(newState == ConnectionState.SUSPENDED){
+ } else if (newState == ConnectionState.SUSPENDED) {
logger.warn("connection SUSPENDED to zookeeper");
}
});
@@ -85,7 +91,8 @@ public class ZookeeperOperator implements InitializingBean {
private CuratorFramework buildClient() {
logger.info("zookeeper registry center init, server lists is: {}.", zookeeperConfig.getServerList());
- CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder().ensembleProvider(new DefaultEnsembleProvider(checkNotNull(zookeeperConfig.getServerList(),"zookeeper quorum can't be null")))
+ CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder().ensembleProvider(new DefaultEnsembleProvider(checkNotNull(zookeeperConfig.getServerList(),
+ "zookeeper quorum can't be null")))
.retryPolicy(new ExponentialBackoffRetry(zookeeperConfig.getBaseSleepTimeMs(), zookeeperConfig.getMaxRetries(), zookeeperConfig.getMaxSleepMs()));
//these has default value
@@ -114,7 +121,7 @@ public class ZookeeperOperator implements InitializingBean {
try {
zkClient.blockUntilConnected();
} catch (final Exception ex) {
- throw new RuntimeException(ex);
+ throw new ServiceException(ex);
}
return zkClient;
}
@@ -138,12 +145,12 @@ public class ZookeeperOperator implements InitializingBean {
throw new IllegalStateException(ex);
} catch (Exception ex) {
logger.error("getChildrenKeys key : {}", key, ex);
- throw new RuntimeException(ex);
+ throw new ServiceException(ex);
}
}
- public boolean hasChildren(final String key){
- Stat stat ;
+ public boolean hasChildren(final String key) {
+ Stat stat;
try {
stat = zkClient.checkExists().forPath(key);
return stat.getNumChildren() >= 1;
@@ -241,4 +248,4 @@ public class ZookeeperOperator implements InitializingBean {
public void close() {
CloseableUtils.closeQuietly(zkClient);
}
-}
\ No newline at end of file
+}
diff --git a/dolphinscheduler-ui/src/js/conf/home/pages/datasource/pages/list/index.vue b/dolphinscheduler-ui/src/js/conf/home/pages/datasource/pages/list/index.vue
index 04683ebe4e8b4a1b1568029c5f582989ce642625..545c6c0b70ad474ecaae9180653df2b9e38ee322 100644
--- a/dolphinscheduler-ui/src/js/conf/home/pages/datasource/pages/list/index.vue
+++ b/dolphinscheduler-ui/src/js/conf/home/pages/datasource/pages/list/index.vue
@@ -22,6 +22,7 @@
{{$t('Create Datasource')}}
diff --git a/pom.xml b/pom.xml
index 214138d9c7498cdc59cf2e597cef2cee9c2e808c..6d7446f8d13b9049feb737e9596abdf3b76a730d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -973,6 +973,8 @@
**/plugin/alert/script/ScriptSenderTest.java
**/plugin/alert/http/HttpAlertChannelFactoryTest.java
**/plugin/alert/http/HttpAlertChannelTest.java
+ **/plugin/alert/feishu/FeiShuAlertChannelFactoryTest.java
+ **/plugin/alert/feishu/FeiShuSenderTest.java
**/plugin/alert/http/HttpAlertPluginTest.java
**/plugin/alert/http/HttpSenderTest.java
**/spi/params/PluginParamsTransferTest.java
diff --git a/sql/upgrade/1.3.3_schema/postgresql/dolphinscheduler_ddl.sql b/sql/upgrade/1.3.3_schema/postgresql/dolphinscheduler_ddl.sql
index 477cb3bf6047153b6bc031cf42160d0efebee7fb..50f560aa13ab20820c5ddc0eb8bd28ad2de2a7a8 100644
--- a/sql/upgrade/1.3.3_schema/postgresql/dolphinscheduler_ddl.sql
+++ b/sql/upgrade/1.3.3_schema/postgresql/dolphinscheduler_ddl.sql
@@ -121,6 +121,7 @@ DROP FUNCTION IF EXISTS ct_dolphin_T_t_ds_process_definition_version();
-- add t_ds_resources_un
+delimiter d//
CREATE OR REPLACE FUNCTION uc_dolphin_T_t_ds_resources_un() RETURNS void AS $$
BEGIN
IF NOT EXISTS (
@@ -129,11 +130,13 @@ BEGIN
AND CONSTRAINT_NAME = 't_ds_resources_un'
)
THEN
-ALTER TABLE t_ds_resources ADD CONSTRAINT t_ds_resources_un UNIQUE (full_name,"type");
-END IF;
+ ALTER TABLE t_ds_resources ADD CONSTRAINT t_ds_resources_un UNIQUE (full_name,"type");
+ END IF;
END;
$$ LANGUAGE plpgsql;
+d//
+delimiter ;
SELECT uc_dolphin_T_t_ds_resources_un();
DROP FUNCTION IF EXISTS uc_dolphin_T_t_ds_resources_un();
diff --git a/sql/upgrade/1.4.0_schema/postgresql/dolphinscheduler_ddl.sql b/sql/upgrade/1.4.0_schema/postgresql/dolphinscheduler_ddl.sql
index 53a8c47d03af97a5fc716137e0bb223a85f143f8..e940cfaad201f3ab1a398c89f510baaabe382220 100644
--- a/sql/upgrade/1.4.0_schema/postgresql/dolphinscheduler_ddl.sql
+++ b/sql/upgrade/1.4.0_schema/postgresql/dolphinscheduler_ddl.sql
@@ -52,7 +52,8 @@ BEGIN
WHERE TABLE_NAME='t_ds_process_definition'
AND COLUMN_NAME ='warning_group_id')
THEN
- ALTER TABLE t_ds_process_definition ADD COLUMN `warning_group_id` int4 DEFAULT NULL COMMENT 'alert group id' AFTER `connects`;
+ ALTER TABLE t_ds_process_definition ADD COLUMN warning_group_id int4 DEFAULT NULL;
+ COMMENT ON COLUMN t_ds_process_definition.warning_group_id IS 'alert group id';
END IF;
END;
$$ LANGUAGE plpgsql;
@@ -70,7 +71,8 @@ BEGIN
WHERE TABLE_NAME='t_ds_process_definition_version'
AND COLUMN_NAME ='warning_group_id')
THEN
- ALTER TABLE t_ds_process_definition_version ADD COLUMN `warning_group_id` int4 DEFAULT NULL COMMENT 'alert group id' AFTER `connects`;
+ ALTER TABLE t_ds_process_definition_version ADD COLUMN warning_group_id int4 DEFAULT NULL;
+ COMMENT ON COLUMN t_ds_process_definition_version.warning_group_id IS 'alert group id';
END IF;
END;
$$ LANGUAGE plpgsql;
@@ -88,7 +90,8 @@ BEGIN
WHERE TABLE_NAME='t_ds_alertgroup'
AND COLUMN_NAME ='alert_instance_ids')
THEN
- ALTER TABLE t_ds_alertgroup ADD COLUMN `alert_instance_ids` varchar (255) DEFAULT NULL COMMENT 'alert instance ids' AFTER `id`;
+ ALTER TABLE t_ds_alertgroup ADD COLUMN alert_instance_ids varchar (255) DEFAULT NULL;
+ COMMENT ON COLUMN t_ds_alertgroup.alert_instance_ids IS 'alert instance ids';
END IF;
END;
$$ LANGUAGE plpgsql;
@@ -106,7 +109,8 @@ BEGIN
WHERE TABLE_NAME='t_ds_alertgroup'
AND COLUMN_NAME ='create_user_id')
THEN
- ALTER TABLE t_ds_alertgroup ADD COLUMN `create_user_id` int4 DEFAULT NULL COMMENT 'create user id' AFTER `alert_instance_ids`;
+ ALTER TABLE t_ds_alertgroup ADD COLUMN create_user_id int4 DEFAULT NULL;
+ COMMENT ON COLUMN t_ds_alertgroup.create_user_id IS 'create user id';
END IF;
END;
$$ LANGUAGE plpgsql;