Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
apache
DolphinScheduler
提交
8b00582e
DolphinScheduler
项目概览
apache
/
DolphinScheduler
上一次同步 1 年多
通知
705
Star
9572
Fork
3514
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
DolphinScheduler
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
前往新版Gitcode,体验更适合开发者的 AI 搜索 >>
未验证
提交
8b00582e
编写于
4月 16, 2020
作者:
D
dailidong
提交者:
GitHub
4月 16, 2020
浏览文件
操作
浏览文件
下载
差异文件
Merge branch 'dev' into fixbug-#2439
上级
aec883ce
f929c6f3
变更
13
隐藏空白更改
内联
并排
Showing
13 changed file
with
48 addition
and
62 deletion
+48
-62
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/DataAnalysisService.java
...che/dolphinscheduler/api/service/DataAnalysisService.java
+5
-9
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java
.../apache/dolphinscheduler/api/service/ExecutorService.java
+12
-13
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/DataAnalysisServiceTest.java
...dolphinscheduler/api/service/DataAnalysisServiceTest.java
+0
-7
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/thread/ThreadUtils.java
...rg/apache/dolphinscheduler/common/thread/ThreadUtils.java
+10
-0
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java
.../java/org/apache/dolphinscheduler/dao/entity/Command.java
+1
-1
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/CommandMapperTest.java
...apache/dolphinscheduler/dao/mapper/CommandMapperTest.java
+4
-9
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java
...ler/server/master/consumer/TaskPriorityQueueConsumer.java
+2
-3
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskAckProcessor.java
...inscheduler/server/master/processor/TaskAckProcessor.java
+2
-4
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskResponseProcessor.java
...eduler/server/master/processor/TaskResponseProcessor.java
+2
-4
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
...lphinscheduler/server/master/runner/MasterExecThread.java
+1
-1
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackService.java
...cheduler/server/worker/processor/TaskCallbackService.java
+4
-6
sql/dolphinscheduler-postgre.sql
sql/dolphinscheduler-postgre.sql
+3
-3
sql/dolphinscheduler_mysql.sql
sql/dolphinscheduler_mysql.sql
+2
-2
未找到文件。
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/DataAnalysisService.java
浏览文件 @
8b00582e
...
...
@@ -106,14 +106,12 @@ public class DataAnalysisService extends BaseService{
List
<
ExecuteStatusCount
>
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
<
String
,
Object
>
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
<
CommandCount
>
errorCommandStateCounts
=
errorCommandMapper
.
countCommandState
(
start
,
end
,
projectIdArray
);
start
,
end
,
projectIdArray
);
//
Map
<
CommandType
,
Map
<
String
,
Integer
>>
dataMap
=
new
HashMap
<>();
...
...
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java
浏览文件 @
8b00582e
...
...
@@ -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
<
String
,
Object
>
updateProcessInstance
State
(
Integer
processInstanceId
,
ExecutionStatus
executionStatus
)
{
private
Map
<
String
,
Object
>
updateProcessInstance
Prepare
(
ProcessInstance
processInstance
,
CommandType
commandType
,
ExecutionStatus
executionStatus
)
{
Map
<
String
,
Object
>
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
;
}
...
...
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/DataAnalysisServiceTest.java
浏览文件 @
8b00582e
...
...
@@ -114,9 +114,6 @@ public class DataAnalysisServiceTest {
Map
<
String
,
Object
>
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
<
String
,
Object
>
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
());
...
...
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/thread/ThreadUtils.java
浏览文件 @
8b00582e
...
...
@@ -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
)
{}
}
}
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java
浏览文件 @
8b00582e
...
...
@@ -111,7 +111,7 @@ public class Command {
/**
* worker group
*/
@TableField
(
exist
=
false
)
@TableField
(
"worker_group"
)
private
String
workerGroup
;
public
Command
()
{
...
...
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/CommandMapperTest.java
浏览文件 @
8b00582e
...
...
@@ -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
<
Command
>
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
);
}
}
}
/**
...
...
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java
浏览文件 @
8b00582e
...
...
@@ -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
){
...
...
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskAckProcessor.java
浏览文件 @
8b00582e
...
...
@@ -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
);
}
}
...
...
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskResponseProcessor.java
浏览文件 @
8b00582e
...
...
@@ -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
);
}
}
...
...
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
浏览文件 @
8b00582e
...
...
@@ -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
());
...
...
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackService.java
浏览文件 @
8b00582e
...
...
@@ -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
;
}
...
...
sql/dolphinscheduler-postgre.sql
浏览文件 @
8b00582e
...
...
@@ -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
...
...
sql/dolphinscheduler_mysql.sql
浏览文件 @
8b00582e
...
...
@@ -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
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录