From d6798c6db7fe50189cd83e4b2a619587c7c5a47a Mon Sep 17 00:00:00 2001 From: qiaozhanwei Date: Thu, 16 Apr 2020 18:16:19 +0800 Subject: [PATCH 1/2] =?UTF-8?q?1=EF=BC=8Ctask=20status=20statistics=20and?= =?UTF-8?q?=20process=20status=20statistics=20bug=20fix=20(#2357)=20=202?= =?UTF-8?q?=EF=BC=8Cworker=20group=20bug=20fix=20(#2430)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * dispatch task fail will set task status failed * 1，no worker condition , master will while ture wait for worker startup 2，worker response task status sync wait for result * 1，no worker condition , master will while ture wait for worker startup 2，worker response task status sync wait for result * 1，no worker condition , master will while ture wait for worker startup 2，worker response task status sync wait for result * 1，no worker condition , master will while ture wait for worker startup 2，worker response task status sync wait for result * 1，no worker condition , master will while ture wait for worker startup 2，worker response task status sync wait for result * 1，no worker condition , master will while ture wait for worker startup 2，worker response task status sync wait for result * 1，no worker condition , master will while ture wait for worker startup 2，worker response task status sync wait for result * 1，no worker condition , master will while ture wait for worker startup 2，worker response task status sync wait for result * 1，no worker condition , master will while ture wait for worker startup 2，worker response task status sync wait for result * 1，no worker condition , master will while ture wait for worker startup 2，worker response task status sync wait for result * 1，no worker condition , master will while ture wait for worker startup 2，worker response task status sync wait for result * 1，task status statistics and process status statistics bug fix (#2357) 2，worker group bug fix * 1，task status statistics and process status statistics bug fix (#2357) 2，worker group bug fix * 1，task status statistics and process status statistics bug fix (#2357) 2，worker group bug fix * 1，task status statistics and process status statistics bug fix (#2357) 2，worker group bug fix Co-authored-by: qiaozhanwei --- .../api/service/DataAnalysisService.java | 14 +++++--------- .../api/service/DataAnalysisServiceTest.java | 7 ------- .../common/thread/ThreadUtils.java | 10 ++++++++++ .../dolphinscheduler/dao/entity/Command.java | 2 +- .../dao/mapper/CommandMapperTest.java | 13 ++++--------- .../master/consumer/TaskPriorityQueueConsumer.java | 5 ++--- .../server/master/processor/TaskAckProcessor.java | 6 ++---- .../master/processor/TaskResponseProcessor.java | 6 ++---- .../worker/processor/TaskCallbackService.java | 10 ++++------ sql/dolphinscheduler-postgre.sql | 6 +++--- sql/dolphinscheduler_mysql.sql | 4 ++-- 11 files changed, 35 insertions(+), 48 deletions(-) diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/DataAnalysisService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/DataAnalysisService.java index 0c93e00a80..39bec56357 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/DataAnalysisService.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/DataAnalysisService.java @@ -106,14 +106,12 @@ public class DataAnalysisService extends BaseService{ List taskInstanceStateCounts = taskInstanceMapper.countTaskInstanceStateByUser(start, end, projectIds); - if (taskInstanceStateCounts != null && !taskInstanceStateCounts.isEmpty()) { + if (taskInstanceStateCounts != null) { TaskCountDto taskCountResult = new TaskCountDto(taskInstanceStateCounts); result.put(Constants.DATA_LIST, taskCountResult); putMsg(result, Status.SUCCESS); - } else { - putMsg(result, Status.TASK_INSTANCE_STATE_COUNT_ERROR); } - return result; + return result; } private void putErrorRequestParamsMsg(Map result) { @@ -153,14 +151,12 @@ public class DataAnalysisService extends BaseService{ processInstanceMapper.countInstanceStateByUser(start, end, projectIdArray); - if (processInstanceStateCounts != null && !processInstanceStateCounts.isEmpty()) { + if (processInstanceStateCounts != null) { TaskCountDto taskCountResult = new TaskCountDto(processInstanceStateCounts); result.put(Constants.DATA_LIST, taskCountResult); putMsg(result, Status.SUCCESS); - } else { - putMsg(result, Status.COUNT_PROCESS_INSTANCE_STATE_ERROR); } - return result; + return result; } @@ -234,7 +230,7 @@ public class DataAnalysisService extends BaseService{ // count error command state List errorCommandStateCounts = errorCommandMapper.countCommandState( - start, end, projectIdArray); + start, end, projectIdArray); // Map> dataMap = new HashMap<>(); diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/DataAnalysisServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/DataAnalysisServiceTest.java index 35cc6ae9a6..14612fcef8 100644 --- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/DataAnalysisServiceTest.java +++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/DataAnalysisServiceTest.java @@ -114,9 +114,6 @@ public class DataAnalysisServiceTest { Map result = dataAnalysisService.countTaskStateByProject(user, 2, startDate, endDate); Assert.assertTrue(result.isEmpty()); - // task instance state count error - result = dataAnalysisService.countTaskStateByProject(user, 1, startDate, endDate); - Assert.assertEquals(Status.TASK_INSTANCE_STATE_COUNT_ERROR,result.get(Constants.STATUS)); //SUCCESS Mockito.when(taskInstanceMapper.countTaskInstanceStateByUser(DateUtils.getScheduleDate(startDate), @@ -137,10 +134,6 @@ public class DataAnalysisServiceTest { Map result = dataAnalysisService.countProcessInstanceStateByProject(user,2,startDate,endDate); Assert.assertTrue(result.isEmpty()); - //COUNT_PROCESS_INSTANCE_STATE_ERROR - result = dataAnalysisService.countProcessInstanceStateByProject(user,1,startDate,endDate); - Assert.assertEquals(Status.COUNT_PROCESS_INSTANCE_STATE_ERROR,result.get(Constants.STATUS)); - //SUCCESS Mockito.when(processInstanceMapper.countInstanceStateByUser(DateUtils.getScheduleDate(startDate), DateUtils.getScheduleDate(endDate), new Integer[]{1})).thenReturn(getTaskInstanceStateCounts()); diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/thread/ThreadUtils.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/thread/ThreadUtils.java index e348835363..a9a124547a 100644 --- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/thread/ThreadUtils.java +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/thread/ThreadUtils.java @@ -223,4 +223,14 @@ public class ThreadUtils { } return id + " (" + name + ")"; } + + /** + * sleep + * @param millis millis + */ + public static void sleep(final long millis) { + try { + Thread.sleep(millis); + } catch (final InterruptedException ignore) {} + } } diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java index 5a6974803c..7d52dc93f3 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java @@ -111,7 +111,7 @@ public class Command { /** * worker group */ - @TableField(exist = false) + @TableField("worker_group") private String workerGroup; public Command() { diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/CommandMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/CommandMapperTest.java index 468824bf6e..297ea66c94 100644 --- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/CommandMapperTest.java +++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/CommandMapperTest.java @@ -76,7 +76,8 @@ public class CommandMapperTest { //query Command actualCommand = commandMapper.selectById(expectedCommand.getId()); - assertEquals(expectedCommand, actualCommand); + assertNotNull(actualCommand); + assertEquals(expectedCommand.getProcessDefinitionId(), actualCommand.getProcessDefinitionId()); } /** @@ -94,7 +95,8 @@ public class CommandMapperTest { Command actualCommand = commandMapper.selectById(expectedCommand.getId()); - assertEquals(expectedCommand,actualCommand); + assertNotNull(actualCommand); + assertEquals(expectedCommand.getUpdateTime(),actualCommand.getUpdateTime()); } @@ -127,13 +129,6 @@ public class CommandMapperTest { List actualCommands = commandMapper.selectList(null); assertThat(actualCommands.size(), greaterThanOrEqualTo(count)); - - for (Command actualCommand : actualCommands){ - Command expectedCommand = commandMap.get(actualCommand.getId()); - if (expectedCommand != null){ - assertEquals(expectedCommand,actualCommand); - } - } } /** diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java index b2cf53a575..cdd9ff2219 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java @@ -31,6 +31,7 @@ import org.apache.dolphinscheduler.common.task.sqoop.SqoopParameters; import org.apache.dolphinscheduler.common.task.sqoop.sources.SourceMysqlParameter; import org.apache.dolphinscheduler.common.task.sqoop.targets.TargetMysqlParameter; import org.apache.dolphinscheduler.common.thread.Stopper; +import org.apache.dolphinscheduler.common.thread.ThreadUtils; import org.apache.dolphinscheduler.common.utils.*; import org.apache.dolphinscheduler.dao.entity.*; import org.apache.dolphinscheduler.server.builder.TaskExecutionContextBuilder; @@ -122,9 +123,7 @@ public class TaskPriorityQueueConsumer extends Thread{ result = dispatcher.dispatch(executionContext); } catch (ExecuteException e) { logger.error("dispatch error",e); - try { - Thread.sleep(SLEEP_TIME_MILLIS); - } catch (InterruptedException e1) {} + ThreadUtils.sleep(SLEEP_TIME_MILLIS); } if (result){ diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskAckProcessor.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskAckProcessor.java index 7af9cdc2cc..3460248dfb 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskAckProcessor.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskAckProcessor.java @@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.server.master.processor; import io.netty.channel.Channel; import org.apache.dolphinscheduler.common.enums.ExecutionStatus; import org.apache.dolphinscheduler.common.thread.Stopper; +import org.apache.dolphinscheduler.common.thread.ThreadUtils; import org.apache.dolphinscheduler.common.utils.Preconditions; import org.apache.dolphinscheduler.dao.entity.TaskInstance; import org.apache.dolphinscheduler.remote.command.Command; @@ -101,10 +102,7 @@ public class TaskAckProcessor implements NettyRequestProcessor { if (taskInstance != null && ackStatus.typeIsRunning()){ break; } - - try { - Thread.sleep(SLEEP_TIME_MILLIS); - } catch (InterruptedException e) {} + ThreadUtils.sleep(SLEEP_TIME_MILLIS); } } diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskResponseProcessor.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskResponseProcessor.java index ecb8646ad0..721b146d86 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskResponseProcessor.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskResponseProcessor.java @@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.server.master.processor; import io.netty.channel.Channel; import org.apache.dolphinscheduler.common.enums.ExecutionStatus; import org.apache.dolphinscheduler.common.thread.Stopper; +import org.apache.dolphinscheduler.common.thread.ThreadUtils; import org.apache.dolphinscheduler.common.utils.Preconditions; import org.apache.dolphinscheduler.dao.entity.TaskInstance; import org.apache.dolphinscheduler.remote.command.Command; @@ -99,10 +100,7 @@ public class TaskResponseProcessor implements NettyRequestProcessor { if (taskInstance != null && responseStatus.typeIsFinished()){ break; } - - try { - Thread.sleep(SLEEP_TIME_MILLIS); - } catch (InterruptedException e) {} + ThreadUtils.sleep(SLEEP_TIME_MILLIS); } } diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackService.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackService.java index 7cd25cba65..ecae9edc1c 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackService.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackService.java @@ -18,11 +18,11 @@ package org.apache.dolphinscheduler.server.worker.processor; -import com.alibaba.fastjson.JSONObject; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFutureListener; import org.apache.dolphinscheduler.common.thread.Stopper; +import org.apache.dolphinscheduler.common.thread.ThreadUtils; import org.apache.dolphinscheduler.common.utils.CollectionUtils; import org.apache.dolphinscheduler.remote.NettyRemotingClient; import org.apache.dolphinscheduler.remote.command.Command; @@ -37,6 +37,8 @@ import org.springframework.stereotype.Service; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import static org.apache.dolphinscheduler.common.Constants.SLEEP_TIME_MILLIS; + /** * taks callback service */ @@ -98,11 +100,7 @@ public class TaskCallbackService { while (Stopper.isRunning()) { if (CollectionUtils.isEmpty(masterNodes)) { logger.error("no available master node"); - try { - Thread.sleep(1000); - }catch (Exception e){ - - } + ThreadUtils.sleep(SLEEP_TIME_MILLIS); }else { break; } diff --git a/sql/dolphinscheduler-postgre.sql b/sql/dolphinscheduler-postgre.sql index b48ddde042..b26520e29c 100644 --- a/sql/dolphinscheduler-postgre.sql +++ b/sql/dolphinscheduler-postgre.sql @@ -234,7 +234,7 @@ CREATE TABLE t_ds_command ( dependence varchar(255) DEFAULT NULL , update_time timestamp DEFAULT NULL , process_instance_priority int DEFAULT NULL , - worker_group_id int DEFAULT '-1' , + worker_group varchar(64), PRIMARY KEY (id) ) ; @@ -275,7 +275,7 @@ CREATE TABLE t_ds_error_command ( update_time timestamp DEFAULT NULL , dependence text , process_instance_priority int DEFAULT NULL , - worker_group_id int DEFAULT '-1' , + worker_group varchar(64), message text , PRIMARY KEY (id) ); @@ -748,7 +748,7 @@ CREATE SEQUENCE t_ds_worker_server_id_sequence; ALTER TABLE t_ds_worker_server ALTER COLUMN id SET DEFAULT NEXTVAL('t_ds_worker_server_id_sequence'); --- Records of t_ds_user，user : admin , password : dolphinscheduler123 +-- Records of t_ds_user?user : admin , password : dolphinscheduler123 INSERT INTO t_ds_user(user_name,user_password,user_type,email,phone,tenant_id,create_time,update_time) VALUES ('admin', '7ad2410b2f4c074479a8937a28a22b8f', '0', 'xxx@qq.com', 'xx', '0', '2018-03-27 15:48:50', '2018-10-24 17:40:22'); -- Records of t_ds_alertgroup，dolphinscheduler warning group diff --git a/sql/dolphinscheduler_mysql.sql b/sql/dolphinscheduler_mysql.sql index 559e684133..c0c81dcb21 100644 --- a/sql/dolphinscheduler_mysql.sql +++ b/sql/dolphinscheduler_mysql.sql @@ -333,7 +333,7 @@ CREATE TABLE `t_ds_command` ( `dependence` varchar(255) DEFAULT NULL COMMENT 'dependence', `update_time` datetime DEFAULT NULL COMMENT 'update time', `process_instance_priority` int(11) DEFAULT NULL COMMENT 'process instance priority: 0 Highest,1 High,2 Medium,3 Low,4 Lowest', - `worker_group_id` int(11) DEFAULT '-1' COMMENT 'worker group id', + `worker_group` varchar(64) COMMENT 'worker group', PRIMARY KEY (`id`) ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8; @@ -380,7 +380,7 @@ CREATE TABLE `t_ds_error_command` ( `update_time` datetime DEFAULT NULL COMMENT 'update time', `dependence` text COMMENT 'dependence', `process_instance_priority` int(11) DEFAULT NULL COMMENT 'process instance priority, 0 Highest,1 High,2 Medium,3 Low,4 Lowest', - `worker_group_id` int(11) DEFAULT '-1' COMMENT 'worker group id', + `worker_group` varchar(64) COMMENT 'worker group', `message` text COMMENT 'message', PRIMARY KEY (`id`) USING BTREE ) ENGINE=InnoDB DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC; From f929c6f3b7a43b78db977206c3425d88ae3c357b Mon Sep 17 00:00:00 2001 From: songgg <1172417734@qq.com> Date: Thu, 16 Apr 2020 22:50:26 +0800 Subject: [PATCH 2/2] =?UTF-8?q?[optimization]=EF=BC=9Aprevent=20repeated?= =?UTF-8?q?=20database=20updates=20(#2396)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * sqlTask failed to run * prevent repeated database updates * prevent repeated database updates * prevent repeated database updates Co-authored-by: songqh Co-authored-by: dailidong --- .../api/service/ExecutorService.java | 25 +++++++++---------- .../master/runner/MasterExecThread.java | 2 +- 2 files changed, 13 insertions(+), 14 deletions(-) diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java index 7ce7497e98..d290886572 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java @@ -225,20 +225,14 @@ public class ExecutorService extends BaseService{ if (processInstance.getState() == ExecutionStatus.READY_STOP) { putMsg(result, Status.PROCESS_INSTANCE_ALREADY_CHANGED, processInstance.getName(), processInstance.getState()); } else { - processInstance.setCommandType(CommandType.STOP); - processInstance.addHistoryCmd(CommandType.STOP); - processService.updateProcessInstance(processInstance); - result = updateProcessInstanceState(processInstanceId, ExecutionStatus.READY_STOP); + result = updateProcessInstancePrepare(processInstance, CommandType.STOP, ExecutionStatus.READY_STOP); } break; case PAUSE: if (processInstance.getState() == ExecutionStatus.READY_PAUSE) { putMsg(result, Status.PROCESS_INSTANCE_ALREADY_CHANGED, processInstance.getName(), processInstance.getState()); } else { - processInstance.setCommandType(CommandType.PAUSE); - processInstance.addHistoryCmd(CommandType.PAUSE); - processService.updateProcessInstance(processInstance); - result = updateProcessInstanceState(processInstanceId, ExecutionStatus.READY_PAUSE); + result = updateProcessInstancePrepare(processInstance, CommandType.PAUSE, ExecutionStatus.READY_PAUSE); } break; default: @@ -308,22 +302,27 @@ public class ExecutorService extends BaseService{ } /** - * update process instance state + * prepare to update process instance command type and status * - * @param processInstanceId process instance id + * @param processInstance process instance + * @param commandType command type * @param executionStatus execute status * @return update result */ - private Map updateProcessInstanceState(Integer processInstanceId, ExecutionStatus executionStatus) { + private Map updateProcessInstancePrepare(ProcessInstance processInstance, CommandType commandType, ExecutionStatus executionStatus) { Map result = new HashMap<>(5); - int update = processService.updateProcessInstanceState(processInstanceId, executionStatus); + processInstance.setCommandType(commandType); + processInstance.addHistoryCmd(commandType); + processInstance.setState(executionStatus); + int update = processService.updateProcessInstance(processInstance); + + // determine whether the process is normal if (update > 0) { putMsg(result, Status.SUCCESS); } else { putMsg(result, Status.EXECUTE_PROCESS_INSTANCE_ERROR); } - return result; } diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java index b1ac73cb54..e1e8c090fd 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java @@ -914,7 +914,7 @@ public class MasterExecThread implements Runnable { processInstance.getId(), processInstance.getName(), processInstance.getState(), state, processInstance.getCommandType()); - processInstance.setState(state); + ProcessInstance instance = processService.findProcessInstanceById(processInstance.getId()); instance.setState(state); instance.setProcessDefinition(processInstance.getProcessDefinition());