Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
DolphinScheduler社区
DolphinScheduler
提交
64d538e1
DolphinScheduler
项目概览
DolphinScheduler社区
/
DolphinScheduler
与 Fork 源项目一致
Fork自
apache / DolphinScheduler
通知
67
Star
1
Fork
0
代码
文件
提交
分支
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 搜索 >>
提交
64d538e1
编写于
2月 21, 2020
作者:
T
Technoboy-
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
updates
上级
d6ea202e
变更
16
隐藏空白更改
内联
并排
Showing
16 changed file
with
696 addition
and
84 deletion
+696
-84
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/entity/TaskExecutionContext.java
.../dolphinscheduler/remote/entity/TaskExecutionContext.java
+3
-3
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/host/Host.java
.../org/apache/dolphinscheduler/server/master/host/Host.java
+95
-0
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/host/HostManager.java
...ache/dolphinscheduler/server/master/host/HostManager.java
+27
-0
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/host/RoundRobinHostManager.java
...inscheduler/server/master/host/RoundRobinHostManager.java
+57
-0
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/host/assign/RandomSelector.java
...inscheduler/server/master/host/assign/RandomSelector.java
+45
-0
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/host/assign/RoundRobinSelector.java
...heduler/server/master/host/assign/RoundRobinSelector.java
+40
-0
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/host/assign/Selector.java
.../dolphinscheduler/server/master/host/assign/Selector.java
+26
-0
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistry.java
...lphinscheduler/server/master/registry/MasterRegistry.java
+104
-0
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterBaseTaskExecThread.java
...eduler/server/master/runner/MasterBaseTaskExecThread.java
+22
-21
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/ZookeeperNodeManager.java
...olphinscheduler/server/registry/ZookeeperNodeManager.java
+159
-0
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/ZookeeperRegistryCenter.java
...hinscheduler/server/registry/ZookeeperRegistryCenter.java
+25
-0
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/WorkerRequestProcessor.java
...duler/server/worker/processor/WorkerRequestProcessor.java
+12
-18
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistry.java
...lphinscheduler/server/worker/registry/WorkerRegistry.java
+1
-1
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskScheduleThread.java
...hinscheduler/server/worker/runner/TaskScheduleThread.java
+38
-40
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/AbstractListener.java
.../apache/dolphinscheduler/service/zk/AbstractListener.java
+36
-0
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZookeeperCachedOperator.java
.../dolphinscheduler/service/zk/ZookeeperCachedOperator.java
+6
-1
未找到文件。
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/
command/TaskInfo
.java
→
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/
entity/TaskExecutionContext
.java
浏览文件 @
64d538e1
...
...
@@ -15,7 +15,7 @@
* limitations under the License.
*/
package
org.apache.dolphinscheduler.remote.
command
;
package
org.apache.dolphinscheduler.remote.
entity
;
import
java.io.Serializable
;
import
java.util.Date
;
...
...
@@ -23,7 +23,7 @@ import java.util.Date;
/**
* master/worker task transport
*/
public
class
Task
Info
implements
Serializable
{
public
class
Task
ExecutionContext
implements
Serializable
{
/**
* task instance id
...
...
@@ -229,7 +229,7 @@ public class TaskInfo implements Serializable{
@Override
public
String
toString
()
{
return
"Task
Info
{"
+
return
"Task
ExecutionContext
{"
+
"taskId="
+
taskId
+
", taskName='"
+
taskName
+
'\''
+
", startTime="
+
startTime
+
...
...
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/host/Host.java
0 → 100644
浏览文件 @
64d538e1
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package
org.apache.dolphinscheduler.server.master.host
;
import
java.util.Objects
;
public
class
Host
{
private
String
address
;
private
String
ip
;
private
int
port
;
public
Host
()
{
}
public
Host
(
String
ip
,
int
port
)
{
this
.
ip
=
ip
;
this
.
port
=
port
;
this
.
address
=
ip
+
":"
+
port
;
}
public
String
getAddress
()
{
return
address
;
}
public
void
setAddress
(
String
address
)
{
this
.
address
=
address
;
}
public
String
getIp
()
{
return
ip
;
}
public
void
setIp
(
String
ip
)
{
this
.
ip
=
ip
;
this
.
address
=
ip
+
":"
+
port
;
}
public
int
getPort
()
{
return
port
;
}
public
void
setPort
(
int
port
)
{
this
.
port
=
port
;
this
.
address
=
ip
+
":"
+
port
;
}
public
static
Host
of
(
String
address
){
String
[]
parts
=
address
.
split
(
":"
);
if
(
parts
.
length
!=
2
)
{
throw
new
IllegalArgumentException
(
String
.
format
(
"Address : %s illegal."
,
address
));
}
Host
host
=
new
Host
(
parts
[
0
],
Integer
.
parseInt
(
parts
[
1
]));
return
host
;
}
@Override
public
String
toString
()
{
return
"Host{"
+
"address='"
+
address
+
'\''
+
'}'
;
}
@Override
public
boolean
equals
(
Object
o
)
{
if
(
this
==
o
)
return
true
;
if
(
o
==
null
||
getClass
()
!=
o
.
getClass
())
return
false
;
Host
host
=
(
Host
)
o
;
return
Objects
.
equals
(
getAddress
(),
host
.
getAddress
());
}
@Override
public
int
hashCode
()
{
return
Objects
.
hash
(
getAddress
());
}
}
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/host/HostManager.java
0 → 100644
浏览文件 @
64d538e1
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package
org.apache.dolphinscheduler.server.master.host
;
import
org.apache.dolphinscheduler.remote.entity.TaskExecutionContext
;
public
interface
HostManager
{
Host
select
(
TaskExecutionContext
context
);
}
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/host/RoundRobinHostManager.java
0 → 100644
浏览文件 @
64d538e1
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package
org.apache.dolphinscheduler.server.master.host
;
import
org.apache.dolphinscheduler.common.utils.CollectionUtils
;
import
org.apache.dolphinscheduler.remote.entity.TaskExecutionContext
;
import
org.apache.dolphinscheduler.server.master.host.assign.RoundRobinSelector
;
import
org.apache.dolphinscheduler.server.registry.ZookeeperNodeManager
;
import
org.slf4j.Logger
;
import
org.slf4j.LoggerFactory
;
import
org.springframework.beans.factory.annotation.Autowired
;
import
org.springframework.stereotype.Service
;
import
java.util.ArrayList
;
import
java.util.Collection
;
import
java.util.List
;
@Service
public
class
RoundRobinHostManager
implements
HostManager
{
private
final
Logger
logger
=
LoggerFactory
.
getLogger
(
RoundRobinHostManager
.
class
);
@Autowired
private
RoundRobinSelector
<
Host
>
selector
;
@Autowired
private
ZookeeperNodeManager
zookeeperNodeManager
;
@Override
public
Host
select
(
TaskExecutionContext
context
){
Host
host
=
new
Host
();
Collection
<
String
>
nodes
=
zookeeperNodeManager
.
getWorkerNodes
();
if
(
CollectionUtils
.
isEmpty
(
nodes
)){
return
host
;
}
List
<
Host
>
candidateHosts
=
new
ArrayList
<>(
nodes
.
size
());
nodes
.
stream
().
forEach
(
node
->
candidateHosts
.
add
(
Host
.
of
(
node
)));
return
selector
.
select
(
candidateHosts
);
}
}
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/host/assign/RandomSelector.java
0 → 100644
浏览文件 @
64d538e1
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package
org.apache.dolphinscheduler.server.master.host.assign
;
import
java.util.Collection
;
import
java.util.Random
;
public
class
RandomSelector
<
T
>
implements
Selector
<
T
>
{
private
final
Random
random
=
new
Random
();
@Override
public
T
select
(
final
Collection
<
T
>
source
)
{
if
(
source
==
null
||
source
.
size
()
==
0
)
{
throw
new
IllegalArgumentException
(
"Empty source."
);
}
if
(
source
.
size
()
==
1
)
{
return
(
T
)
source
.
toArray
()[
0
];
}
int
size
=
source
.
size
();
int
randomIndex
=
random
.
nextInt
(
size
);
return
(
T
)
source
.
toArray
()[
randomIndex
];
}
}
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/host/assign/RoundRobinSelector.java
0 → 100644
浏览文件 @
64d538e1
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package
org.apache.dolphinscheduler.server.master.host.assign
;
import
java.util.Collection
;
import
java.util.concurrent.atomic.AtomicInteger
;
public
class
RoundRobinSelector
<
T
>
implements
Selector
<
T
>
{
private
final
AtomicInteger
index
=
new
AtomicInteger
(
0
);
@Override
public
T
select
(
Collection
<
T
>
source
)
{
if
(
source
==
null
||
source
.
size
()
==
0
)
{
throw
new
IllegalArgumentException
(
"Empty source."
);
}
if
(
source
.
size
()
==
1
)
{
return
(
T
)
source
.
toArray
()[
0
];
}
int
size
=
source
.
size
();
return
(
T
)
source
.
toArray
()[
index
.
getAndIncrement
()
%
size
];
}
}
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/host/assign/Selector.java
0 → 100644
浏览文件 @
64d538e1
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package
org.apache.dolphinscheduler.server.master.host.assign
;
import
java.util.Collection
;
public
interface
Selector
<
T
>
{
T
select
(
Collection
<
T
>
source
);
}
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistry.java
0 → 100644
浏览文件 @
64d538e1
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package
org.apache.dolphinscheduler.server.master.registry
;
import
org.apache.curator.framework.CuratorFramework
;
import
org.apache.curator.framework.state.ConnectionState
;
import
org.apache.curator.framework.state.ConnectionStateListener
;
import
org.apache.dolphinscheduler.remote.utils.Constants
;
import
org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter
;
import
org.slf4j.Logger
;
import
org.slf4j.LoggerFactory
;
/**
* master registry
*/
public
class
MasterRegistry
{
private
final
Logger
logger
=
LoggerFactory
.
getLogger
(
MasterRegistry
.
class
);
/**
* zookeeper registry center
*/
private
final
ZookeeperRegistryCenter
zookeeperRegistryCenter
;
/**
* port
*/
private
final
int
port
;
/**
* construct
* @param zookeeperRegistryCenter zookeeperRegistryCenter
* @param port port
*/
public
MasterRegistry
(
ZookeeperRegistryCenter
zookeeperRegistryCenter
,
int
port
){
this
.
zookeeperRegistryCenter
=
zookeeperRegistryCenter
;
this
.
port
=
port
;
}
/**
* registry
*/
public
void
registry
()
{
String
address
=
Constants
.
LOCAL_ADDRESS
;
String
localNodePath
=
getWorkerPath
();
zookeeperRegistryCenter
.
getZookeeperCachedOperator
().
persist
(
localNodePath
,
""
);
zookeeperRegistryCenter
.
getZookeeperCachedOperator
().
getZkClient
().
getConnectionStateListenable
().
addListener
(
new
ConnectionStateListener
()
{
@Override
public
void
stateChanged
(
CuratorFramework
client
,
ConnectionState
newState
)
{
if
(
newState
==
ConnectionState
.
LOST
){
logger
.
error
(
"master : {} connection lost from zookeeper"
,
address
);
}
else
if
(
newState
==
ConnectionState
.
RECONNECTED
){
logger
.
info
(
"master : {} reconnected to zookeeper"
,
address
);
zookeeperRegistryCenter
.
getZookeeperCachedOperator
().
persist
(
localNodePath
,
""
);
}
else
if
(
newState
==
ConnectionState
.
SUSPENDED
){
logger
.
warn
(
"master : {} connection SUSPENDED "
,
address
);
}
}
});
logger
.
info
(
"master node : {} registry to ZK successfully."
,
address
);
}
/**
* remove registry info
*/
public
void
unRegistry
()
{
String
address
=
getLocalAddress
();
String
localNodePath
=
getWorkerPath
();
zookeeperRegistryCenter
.
getZookeeperCachedOperator
().
remove
(
localNodePath
);
logger
.
info
(
"worker node : {} unRegistry to ZK."
,
address
);
}
/**
* get worker path
* @return
*/
private
String
getWorkerPath
()
{
String
address
=
getLocalAddress
();
String
localNodePath
=
this
.
zookeeperRegistryCenter
.
getWorkerPath
()
+
"/"
+
address
;
return
localNodePath
;
}
/**
* get local address
* @return
*/
private
String
getLocalAddress
(){
return
Constants
.
LOCAL_ADDRESS
+
":"
+
port
;
}
}
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterBaseTaskExecThread.java
浏览文件 @
64d538e1
...
...
@@ -28,7 +28,7 @@ import org.apache.dolphinscheduler.remote.NettyRemotingClient;
import
org.apache.dolphinscheduler.remote.command.Command
;
import
org.apache.dolphinscheduler.remote.command.ExecuteTaskAckCommand
;
import
org.apache.dolphinscheduler.remote.command.ExecuteTaskRequestCommand
;
import
org.apache.dolphinscheduler.remote.
command.TaskInfo
;
import
org.apache.dolphinscheduler.remote.
entity.TaskExecutionContext
;
import
org.apache.dolphinscheduler.remote.config.NettyClientConfig
;
import
org.apache.dolphinscheduler.remote.exceptions.RemotingException
;
import
org.apache.dolphinscheduler.remote.utils.Address
;
...
...
@@ -137,7 +137,7 @@ public class MasterBaseTaskExecThread implements Callable<Boolean> {
FastJsonSerializer
.
serializeToString
(
convertToTaskInfo
(
destTaskInstance
)));
try
{
Command
responseCommand
=
nettyRemotingClient
.
sendSync
(
address
,
taskRequestCommand
.
convert2Command
(),
Integer
.
MAX_VALUE
);
taskRequestCommand
.
convert2Command
(),
2000
);
ExecuteTaskAckCommand
taskAckCommand
=
FastJsonSerializer
.
deserialize
(
responseCommand
.
getBody
(),
ExecuteTaskAckCommand
.
class
);
...
...
@@ -156,6 +156,7 @@ public class MasterBaseTaskExecThread implements Callable<Boolean> {
}
/**
* set task instance relation
*
...
...
@@ -203,25 +204,25 @@ public class MasterBaseTaskExecThread implements Callable<Boolean> {
* @param taskInstance taskInstance
* @return taskInfo
*/
private
Task
Info
convertToTaskInfo
(
TaskInstance
taskInstance
){
Task
Info
taskInfo
=
new
TaskInfo
();
task
Info
.
setTaskId
(
taskInstance
.
getId
());
task
Info
.
setTaskName
(
taskInstance
.
getName
());
task
Info
.
setStartTime
(
taskInstance
.
getStartTime
());
task
Info
.
setTaskType
(
taskInstance
.
getTaskType
());
task
Info
.
setExecutePath
(
getExecLocalPath
(
taskInstance
));
task
Info
.
setTaskJson
(
taskInstance
.
getTaskJson
());
task
Info
.
setProcessInstanceId
(
taskInstance
.
getProcessInstance
().
getId
());
task
Info
.
setScheduleTime
(
taskInstance
.
getProcessInstance
().
getScheduleTime
());
task
Info
.
setGlobalParams
(
taskInstance
.
getProcessInstance
().
getGlobalParams
());
task
Info
.
setExecutorId
(
taskInstance
.
getProcessInstance
().
getExecutorId
());
task
Info
.
setCmdTypeIfComplement
(
taskInstance
.
getProcessInstance
().
getCmdTypeIfComplement
().
getCode
());
task
Info
.
setTenantCode
(
taskInstance
.
getProcessInstance
().
getTenantCode
());
task
Info
.
setQueue
(
taskInstance
.
getProcessInstance
().
getQueue
());
task
Info
.
setProcessDefineId
(
taskInstance
.
getProcessDefine
().
getId
());
task
Info
.
setProjectId
(
taskInstance
.
getProcessDefine
().
getProjectId
());
return
task
Info
;
private
Task
ExecutionContext
convertToTaskInfo
(
TaskInstance
taskInstance
){
Task
ExecutionContext
taskExecutionContext
=
new
TaskExecutionContext
();
task
ExecutionContext
.
setTaskId
(
taskInstance
.
getId
());
task
ExecutionContext
.
setTaskName
(
taskInstance
.
getName
());
task
ExecutionContext
.
setStartTime
(
taskInstance
.
getStartTime
());
task
ExecutionContext
.
setTaskType
(
taskInstance
.
getTaskType
());
task
ExecutionContext
.
setExecutePath
(
getExecLocalPath
(
taskInstance
));
task
ExecutionContext
.
setTaskJson
(
taskInstance
.
getTaskJson
());
task
ExecutionContext
.
setProcessInstanceId
(
taskInstance
.
getProcessInstance
().
getId
());
task
ExecutionContext
.
setScheduleTime
(
taskInstance
.
getProcessInstance
().
getScheduleTime
());
task
ExecutionContext
.
setGlobalParams
(
taskInstance
.
getProcessInstance
().
getGlobalParams
());
task
ExecutionContext
.
setExecutorId
(
taskInstance
.
getProcessInstance
().
getExecutorId
());
task
ExecutionContext
.
setCmdTypeIfComplement
(
taskInstance
.
getProcessInstance
().
getCmdTypeIfComplement
().
getCode
());
task
ExecutionContext
.
setTenantCode
(
taskInstance
.
getProcessInstance
().
getTenantCode
());
task
ExecutionContext
.
setQueue
(
taskInstance
.
getProcessInstance
().
getQueue
());
task
ExecutionContext
.
setProcessDefineId
(
taskInstance
.
getProcessDefine
().
getId
());
task
ExecutionContext
.
setProjectId
(
taskInstance
.
getProcessDefine
().
getProjectId
());
return
task
ExecutionContext
;
}
...
...
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/ZookeeperNodeManager.java
0 → 100644
浏览文件 @
64d538e1
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package
org.apache.dolphinscheduler.server.registry
;
import
org.apache.curator.framework.CuratorFramework
;
import
org.apache.curator.framework.recipes.cache.TreeCacheEvent
;
import
org.apache.dolphinscheduler.service.zk.AbstractListener
;
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.Service
;
import
java.util.Collections
;
import
java.util.HashSet
;
import
java.util.Set
;
import
java.util.concurrent.locks.Lock
;
import
java.util.concurrent.locks.ReentrantLock
;
@Service
public
abstract
class
ZookeeperNodeManager
implements
InitializingBean
{
private
final
Logger
logger
=
LoggerFactory
.
getLogger
(
ZookeeperNodeManager
.
class
);
private
final
Lock
masterLock
=
new
ReentrantLock
();
private
final
Lock
workerLock
=
new
ReentrantLock
();
private
final
Set
<
String
>
workerNodes
=
new
HashSet
<>();
private
final
Set
<
String
>
masterNodes
=
new
HashSet
<>();
@Autowired
private
ZookeeperRegistryCenter
registryCenter
;
@Override
public
void
afterPropertiesSet
()
throws
Exception
{
load
();
registryCenter
.
getZookeeperCachedOperator
().
addListener
(
new
MasterNodeListener
());
registryCenter
.
getZookeeperCachedOperator
().
addListener
(
new
WorkerNodeListener
());
}
private
void
load
(){
Set
<
String
>
schedulerNodes
=
registryCenter
.
getMasterNodesDirectly
();
syncMasterNodes
(
schedulerNodes
);
Set
<
String
>
workersNodes
=
registryCenter
.
getWorkerNodesDirectly
();
syncWorkerNodes
(
workersNodes
);
}
class
WorkerNodeListener
extends
AbstractListener
{
@Override
protected
void
dataChanged
(
CuratorFramework
client
,
TreeCacheEvent
event
,
String
path
)
{
if
(
registryCenter
.
isWorkerPath
(
path
)){
try
{
if
(
event
.
getType
()
==
TreeCacheEvent
.
Type
.
NODE_ADDED
)
{
logger
.
info
(
"worker node : {} added."
,
path
);
Set
<
String
>
previousNodes
=
new
HashSet
<>(
workerNodes
);
Set
<
String
>
currentNodes
=
registryCenter
.
getWorkerNodesDirectly
();
syncWorkerNodes
(
currentNodes
);
}
else
if
(
event
.
getType
()
==
TreeCacheEvent
.
Type
.
NODE_REMOVED
)
{
logger
.
info
(
"worker node : {} down."
,
path
);
Set
<
String
>
previousNodes
=
new
HashSet
<>(
workerNodes
);
Set
<
String
>
currentNodes
=
registryCenter
.
getWorkerNodesDirectly
();
syncWorkerNodes
(
currentNodes
);
}
}
catch
(
IllegalArgumentException
ignore
)
{
logger
.
warn
(
ignore
.
getMessage
());
}
catch
(
Exception
ex
)
{
logger
.
error
(
"WorkerListener capture data change and get data failed"
,
ex
);
}
}
}
}
class
MasterNodeListener
extends
AbstractListener
{
@Override
protected
void
dataChanged
(
CuratorFramework
client
,
TreeCacheEvent
event
,
String
path
)
{
if
(
registryCenter
.
isMasterPath
(
path
))
{
try
{
if
(
event
.
getType
()
==
TreeCacheEvent
.
Type
.
NODE_ADDED
)
{
logger
.
info
(
"master node : {} added."
,
path
);
Set
<
String
>
previousNodes
=
new
HashSet
<>(
masterNodes
);
Set
<
String
>
currentNodes
=
registryCenter
.
getMasterNodesDirectly
();
syncMasterNodes
(
currentNodes
);
}
else
if
(
event
.
getType
()
==
TreeCacheEvent
.
Type
.
NODE_REMOVED
)
{
logger
.
info
(
"master node : {} down."
,
path
);
Set
<
String
>
previousNodes
=
new
HashSet
<>(
masterNodes
);
Set
<
String
>
currentNodes
=
registryCenter
.
getMasterNodesDirectly
();
syncMasterNodes
(
currentNodes
);
}
}
catch
(
Exception
ex
)
{
logger
.
error
(
"MasterNodeListener capture data change and get data failed."
,
ex
);
}
}
}
}
public
Set
<
String
>
getMasterNodes
()
{
masterLock
.
lock
();
try
{
return
Collections
.
unmodifiableSet
(
masterNodes
);
}
finally
{
masterLock
.
unlock
();
}
}
private
void
syncMasterNodes
(
Set
<
String
>
nodes
){
masterLock
.
lock
();
try
{
masterNodes
.
clear
();
masterNodes
.
addAll
(
nodes
);
}
finally
{
masterLock
.
unlock
();
}
}
private
void
syncWorkerNodes
(
Set
<
String
>
nodes
){
workerLock
.
lock
();
try
{
workerNodes
.
clear
();
workerNodes
.
addAll
(
nodes
);
}
finally
{
workerLock
.
unlock
();
}
}
public
Set
<
String
>
getWorkerNodes
(){
workerLock
.
lock
();
try
{
return
Collections
.
unmodifiableSet
(
workerNodes
);
}
finally
{
workerLock
.
unlock
();
}
}
public
void
close
(){
registryCenter
.
close
();
}
}
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/ZookeeperRegistryCenter.java
浏览文件 @
64d538e1
...
...
@@ -22,6 +22,9 @@ import org.springframework.beans.factory.InitializingBean;
import
org.springframework.beans.factory.annotation.Autowired
;
import
org.springframework.stereotype.Service
;
import
java.util.HashSet
;
import
java.util.List
;
import
java.util.Set
;
import
java.util.concurrent.atomic.AtomicBoolean
;
@Service
...
...
@@ -76,6 +79,28 @@ public class ZookeeperRegistryCenter implements InitializingBean {
return
WORKER_PATH
;
}
public
Set
<
String
>
getMasterNodesDirectly
()
{
List
<
String
>
masters
=
getChildrenKeys
(
MASTER_PATH
);
return
new
HashSet
<>(
masters
);
}
public
Set
<
String
>
getWorkerNodesDirectly
()
{
List
<
String
>
workers
=
getChildrenKeys
(
WORKER_PATH
);
return
new
HashSet
<>(
workers
);
}
public
boolean
isWorkerPath
(
String
path
)
{
return
path
!=
null
&&
path
.
contains
(
WORKER_PATH
);
}
public
boolean
isMasterPath
(
String
path
)
{
return
path
!=
null
&&
path
.
contains
(
MASTER_PATH
);
}
public
List
<
String
>
getChildrenKeys
(
final
String
key
)
{
return
zookeeperCachedOperator
.
getChildrenKeys
(
key
);
}
public
ZookeeperCachedOperator
getZookeeperCachedOperator
()
{
return
zookeeperCachedOperator
;
}
...
...
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/WorkerRequestProcessor.java
浏览文件 @
64d538e1
...
...
@@ -19,18 +19,13 @@ package org.apache.dolphinscheduler.server.worker.processor;
import
com.alibaba.fastjson.JSONObject
;
import
io.netty.channel.Channel
;
import
org.apache.dolphinscheduler.common.enums.ExecutionStatus
;
import
org.apache.dolphinscheduler.common.thread.ThreadUtils
;
import
org.apache.dolphinscheduler.common.utils.FileUtils
;
import
org.apache.dolphinscheduler.common.utils.OSUtils
;
import
org.apache.dolphinscheduler.common.utils.Preconditions
;
import
org.apache.dolphinscheduler.common.utils.StringUtils
;
import
org.apache.dolphinscheduler.dao.entity.TaskInstance
;
import
org.apache.dolphinscheduler.dao.entity.Tenant
;
import
org.apache.dolphinscheduler.remote.command.Command
;
import
org.apache.dolphinscheduler.remote.command.CommandType
;
import
org.apache.dolphinscheduler.remote.command.ExecuteTaskRequestCommand
;
import
org.apache.dolphinscheduler.remote.
command.TaskInfo
;
import
org.apache.dolphinscheduler.remote.
entity.TaskExecutionContext
;
import
org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor
;
import
org.apache.dolphinscheduler.remote.utils.FastJsonSerializer
;
import
org.apache.dolphinscheduler.server.worker.config.WorkerConfig
;
...
...
@@ -40,7 +35,6 @@ import org.apache.dolphinscheduler.service.process.ProcessService;
import
org.slf4j.Logger
;
import
org.slf4j.LoggerFactory
;
import
java.util.Date
;
import
java.util.concurrent.ExecutorService
;
/**
...
...
@@ -87,23 +81,23 @@ public class WorkerRequestProcessor implements NettyRequestProcessor {
String
taskInstanceJson
=
taskRequestCommand
.
getTaskInfoJson
();
Task
Info
taskInfo
=
JSONObject
.
parseObject
(
taskInstanceJson
,
TaskInfo
.
class
);
Task
ExecutionContext
taskExecutionContext
=
JSONObject
.
parseObject
(
taskInstanceJson
,
TaskExecutionContext
.
class
);
// local execute path
String
execLocalPath
=
getExecLocalPath
(
task
Info
);
String
execLocalPath
=
getExecLocalPath
(
task
ExecutionContext
);
logger
.
info
(
"task instance local execute path : {} "
,
execLocalPath
);
try
{
FileUtils
.
createWorkDirAndUserIfAbsent
(
execLocalPath
,
task
Info
.
getTenantCode
());
FileUtils
.
createWorkDirAndUserIfAbsent
(
execLocalPath
,
task
ExecutionContext
.
getTenantCode
());
}
catch
(
Exception
ex
){
logger
.
error
(
String
.
format
(
"create execLocalPath : %s"
,
execLocalPath
),
ex
);
}
taskCallbackService
.
addCallbackChannel
(
task
Info
.
getTaskId
(),
taskCallbackService
.
addCallbackChannel
(
task
ExecutionContext
.
getTaskId
(),
new
CallbackChannel
(
channel
,
command
.
getOpaque
()));
// submit task
workerExecService
.
submit
(
new
TaskScheduleThread
(
task
Info
,
workerExecService
.
submit
(
new
TaskScheduleThread
(
task
ExecutionContext
,
processService
,
taskCallbackService
));
}
...
...
@@ -111,13 +105,13 @@ public class WorkerRequestProcessor implements NettyRequestProcessor {
/**
* get execute local path
*
* @param task
Info taskInfo
* @param task
ExecutionContext taskExecutionContext
* @return execute local path
*/
private
String
getExecLocalPath
(
Task
Info
taskInfo
){
return
FileUtils
.
getProcessExecDir
(
task
Info
.
getProjectId
(),
task
Info
.
getProcessDefineId
(),
task
Info
.
getProcessInstanceId
(),
task
Info
.
getTaskId
());
private
String
getExecLocalPath
(
Task
ExecutionContext
taskExecutionContext
){
return
FileUtils
.
getProcessExecDir
(
task
ExecutionContext
.
getProjectId
(),
task
ExecutionContext
.
getProcessDefineId
(),
task
ExecutionContext
.
getProcessInstanceId
(),
task
ExecutionContext
.
getTaskId
());
}
}
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistry.java
浏览文件 @
64d538e1
...
...
@@ -71,7 +71,7 @@ public class WorkerRegistry {
}
}
});
logger
.
info
(
"
schedul
er node : {} registry to ZK successfully."
,
address
);
logger
.
info
(
"
work
er node : {} registry to ZK successfully."
,
address
);
}
/**
...
...
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskScheduleThread.java
浏览文件 @
64d538e1
...
...
@@ -31,11 +31,9 @@ import org.apache.dolphinscheduler.common.process.Property;
import
org.apache.dolphinscheduler.common.task.AbstractParameters
;
import
org.apache.dolphinscheduler.common.task.TaskTimeoutParameter
;
import
org.apache.dolphinscheduler.common.utils.*
;
import
org.apache.dolphinscheduler.dao.entity.ProcessInstance
;
import
org.apache.dolphinscheduler.dao.entity.TaskInstance
;
import
org.apache.dolphinscheduler.remote.command.ExecuteTaskAckCommand
;
import
org.apache.dolphinscheduler.remote.command.ExecuteTaskResponseCommand
;
import
org.apache.dolphinscheduler.remote.
command.TaskInfo
;
import
org.apache.dolphinscheduler.remote.
entity.TaskExecutionContext
;
import
org.apache.dolphinscheduler.server.worker.processor.TaskCallbackService
;
import
org.apache.dolphinscheduler.server.worker.task.AbstractTask
;
import
org.apache.dolphinscheduler.server.worker.task.TaskManager
;
...
...
@@ -63,7 +61,7 @@ public class TaskScheduleThread implements Runnable {
/**
* task instance
*/
private
Task
Info
taskInfo
;
private
Task
ExecutionContext
taskExecutionContext
;
/**
* process service
...
...
@@ -83,67 +81,67 @@ public class TaskScheduleThread implements Runnable {
/**
* constructor
*
* @param task
Info taskInfo
* @param task
ExecutionContext taskExecutionContext
* @param processService processService
* @param taskInstanceCallbackService taskInstanceCallbackService
*/
public
TaskScheduleThread
(
Task
Info
taskInfo
,
ProcessService
processService
,
TaskCallbackService
taskInstanceCallbackService
){
public
TaskScheduleThread
(
Task
ExecutionContext
taskExecutionContext
,
ProcessService
processService
,
TaskCallbackService
taskInstanceCallbackService
){
this
.
processService
=
processService
;
this
.
task
Info
=
taskInfo
;
this
.
task
ExecutionContext
=
taskExecutionContext
;
this
.
taskInstanceCallbackService
=
taskInstanceCallbackService
;
}
@Override
public
void
run
()
{
ExecuteTaskResponseCommand
responseCommand
=
new
ExecuteTaskResponseCommand
(
task
Instance
.
get
Id
());
ExecuteTaskResponseCommand
responseCommand
=
new
ExecuteTaskResponseCommand
(
task
ExecutionContext
.
getTask
Id
());
try
{
// tell master that task is in executing
ExecuteTaskAckCommand
ackCommand
=
buildAckCommand
(
task
Info
.
getTaskType
());
taskInstanceCallbackService
.
sendAck
(
task
Info
.
getTaskId
(),
ackCommand
);
ExecuteTaskAckCommand
ackCommand
=
buildAckCommand
(
task
ExecutionContext
.
getTaskType
());
taskInstanceCallbackService
.
sendAck
(
task
ExecutionContext
.
getTaskId
(),
ackCommand
);
logger
.
info
(
"script path : {}"
,
task
Info
.
getExecutePath
());
logger
.
info
(
"script path : {}"
,
task
ExecutionContext
.
getExecutePath
());
// task node
TaskNode
taskNode
=
JSONObject
.
parseObject
(
task
Info
.
getTaskJson
(),
TaskNode
.
class
);
TaskNode
taskNode
=
JSONObject
.
parseObject
(
task
ExecutionContext
.
getTaskJson
(),
TaskNode
.
class
);
// get resource files
List
<
String
>
resourceFiles
=
createProjectResFiles
(
taskNode
);
// copy hdfs/minio file to local
downloadResource
(
task
Info
.
getExecutePath
(),
task
ExecutionContext
.
getExecutePath
(),
resourceFiles
,
logger
);
// set task props
TaskProps
taskProps
=
new
TaskProps
(
taskNode
.
getParams
(),
task
Info
.
getExecutePath
(),
task
Info
.
getScheduleTime
(),
task
Info
.
getTaskName
(),
task
Info
.
getTaskType
(),
task
Info
.
getTaskId
(),
task
ExecutionContext
.
getExecutePath
(),
task
ExecutionContext
.
getScheduleTime
(),
task
ExecutionContext
.
getTaskName
(),
task
ExecutionContext
.
getTaskType
(),
task
ExecutionContext
.
getTaskId
(),
CommonUtils
.
getSystemEnvPath
(),
task
Info
.
getTenantCode
(),
task
Info
.
getQueue
(),
task
Info
.
getStartTime
(),
task
ExecutionContext
.
getTenantCode
(),
task
ExecutionContext
.
getQueue
(),
task
ExecutionContext
.
getStartTime
(),
getGlobalParamsMap
(),
null
,
CommandType
.
of
(
task
Info
.
getCmdTypeIfComplement
()));
CommandType
.
of
(
task
ExecutionContext
.
getCmdTypeIfComplement
()));
// set task timeout
setTaskTimeout
(
taskProps
,
taskNode
);
taskProps
.
setTaskAppId
(
String
.
format
(
"%s_%s_%s"
,
task
Info
.
getProcessDefineId
(),
task
Info
.
getProcessInstanceId
(),
task
Info
.
getTaskId
()));
task
ExecutionContext
.
getProcessDefineId
(),
task
ExecutionContext
.
getProcessInstanceId
(),
task
ExecutionContext
.
getTaskId
()));
// custom logger
Logger
taskLogger
=
LoggerFactory
.
getLogger
(
LoggerUtils
.
buildTaskId
(
LoggerUtils
.
TASK_LOGGER_INFO_PREFIX
,
task
Info
.
getProcessDefineId
(),
task
Info
.
getProcessInstanceId
(),
task
Info
.
getTaskId
()));
task
ExecutionContext
.
getProcessDefineId
(),
task
ExecutionContext
.
getProcessInstanceId
(),
task
ExecutionContext
.
getTaskId
()));
task
=
TaskManager
.
newTask
(
task
Info
.
getTaskType
(),
task
=
TaskManager
.
newTask
(
task
ExecutionContext
.
getTaskType
(),
taskProps
,
taskLogger
);
...
...
@@ -159,14 +157,14 @@ public class TaskScheduleThread implements Runnable {
//
responseCommand
.
setStatus
(
task
.
getExitStatus
().
getCode
());
responseCommand
.
setEndTime
(
new
Date
());
logger
.
info
(
"task instance id : {},task final status : {}"
,
task
Info
.
getTaskId
(),
task
.
getExitStatus
());
logger
.
info
(
"task instance id : {},task final status : {}"
,
task
ExecutionContext
.
getTaskId
(),
task
.
getExitStatus
());
}
catch
(
Exception
e
){
logger
.
error
(
"task scheduler failure"
,
e
);
kill
();
responseCommand
.
setStatus
(
ExecutionStatus
.
FAILURE
.
getCode
());
responseCommand
.
setEndTime
(
new
Date
());
}
finally
{
taskInstanceCallbackService
.
sendResult
(
task
Info
.
getTaskId
(),
responseCommand
);
taskInstanceCallbackService
.
sendResult
(
task
ExecutionContext
.
getTaskId
(),
responseCommand
);
}
}
...
...
@@ -178,7 +176,7 @@ public class TaskScheduleThread implements Runnable {
Map
<
String
,
String
>
globalParamsMap
=
new
HashMap
<>(
16
);
// global params string
String
globalParamsStr
=
task
Info
.
getGlobalParams
();
String
globalParamsStr
=
task
ExecutionContext
.
getGlobalParams
();
if
(
globalParamsStr
!=
null
)
{
List
<
Property
>
globalParamsList
=
JSONObject
.
parseArray
(
globalParamsStr
,
Property
.
class
);
...
...
@@ -199,7 +197,7 @@ public class TaskScheduleThread implements Runnable {
if
(
taskType
.
equals
(
TaskType
.
SQL
.
name
())
||
taskType
.
equals
(
TaskType
.
PROCEDURE
.
name
())){
ackCommand
.
setExecutePath
(
null
);
}
else
{
ackCommand
.
setExecutePath
(
task
Info
.
getExecutePath
());
ackCommand
.
setExecutePath
(
task
ExecutionContext
.
getExecutePath
());
}
return
ackCommand
;
}
...
...
@@ -215,15 +213,15 @@ public class TaskScheduleThread implements Runnable {
.
getDiscriminator
()).
getLogBase
();
if
(
baseLog
.
startsWith
(
Constants
.
SINGLE_SLASH
)){
return
baseLog
+
Constants
.
SINGLE_SLASH
+
task
Info
.
getProcessDefineId
()
+
Constants
.
SINGLE_SLASH
+
task
Info
.
getProcessInstanceId
()
+
Constants
.
SINGLE_SLASH
+
task
Info
.
getTaskId
()
+
".log"
;
task
ExecutionContext
.
getProcessDefineId
()
+
Constants
.
SINGLE_SLASH
+
task
ExecutionContext
.
getProcessInstanceId
()
+
Constants
.
SINGLE_SLASH
+
task
ExecutionContext
.
getTaskId
()
+
".log"
;
}
return
System
.
getProperty
(
"user.dir"
)
+
Constants
.
SINGLE_SLASH
+
baseLog
+
Constants
.
SINGLE_SLASH
+
task
Info
.
getProcessDefineId
()
+
Constants
.
SINGLE_SLASH
+
task
Info
.
getProcessInstanceId
()
+
Constants
.
SINGLE_SLASH
+
task
Info
.
getTaskId
()
+
".log"
;
task
ExecutionContext
.
getProcessDefineId
()
+
Constants
.
SINGLE_SLASH
+
task
ExecutionContext
.
getProcessInstanceId
()
+
Constants
.
SINGLE_SLASH
+
task
ExecutionContext
.
getTaskId
()
+
".log"
;
}
/**
...
...
@@ -329,7 +327,7 @@ public class TaskScheduleThread implements Runnable {
* @throws Exception exception
*/
private
void
checkDownloadPermission
(
List
<
String
>
projectRes
)
throws
Exception
{
int
userId
=
task
Info
.
getExecutorId
();
int
userId
=
task
ExecutionContext
.
getExecutorId
();
String
[]
resNames
=
projectRes
.
toArray
(
new
String
[
projectRes
.
size
()]);
PermissionCheck
<
String
>
permissionCheck
=
new
PermissionCheck
<>(
AuthorizationType
.
RESOURCE_FILE
,
processService
,
resNames
,
userId
,
logger
);
permissionCheck
.
checkPermission
();
...
...
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/AbstractListener.java
0 → 100644
浏览文件 @
64d538e1
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* 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.CuratorFramework
;
import
org.apache.curator.framework.recipes.cache.TreeCacheEvent
;
import
org.apache.curator.framework.recipes.cache.TreeCacheListener
;
public
abstract
class
AbstractListener
implements
TreeCacheListener
{
@Override
public
final
void
childEvent
(
final
CuratorFramework
client
,
final
TreeCacheEvent
event
)
throws
Exception
{
String
path
=
null
==
event
.
getData
()
?
""
:
event
.
getData
().
getPath
();
if
(
path
.
isEmpty
())
{
return
;
}
dataChanged
(
client
,
event
,
path
);
}
protected
abstract
void
dataChanged
(
final
CuratorFramework
client
,
final
TreeCacheEvent
event
,
final
String
path
);
}
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZookeeperCachedOperator.java
浏览文件 @
64d538e1
...
...
@@ -20,6 +20,7 @@ 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
org.slf4j.Logger
;
import
org.slf4j.LoggerFactory
;
import
org.springframework.stereotype.Component
;
...
...
@@ -32,7 +33,7 @@ public class ZookeeperCachedOperator extends ZookeeperOperator {
private
final
Logger
logger
=
LoggerFactory
.
getLogger
(
ZookeeperCachedOperator
.
class
);
TreeCache
treeCache
;
private
TreeCache
treeCache
;
/**
* register a unified listener of /${dsRoot},
*/
...
...
@@ -72,6 +73,10 @@ public class ZookeeperCachedOperator extends ZookeeperOperator {
return
treeCache
;
}
public
void
addListener
(
TreeCacheListener
listener
){
this
.
treeCache
.
getListenable
().
addListener
(
listener
);
}
@Override
public
void
close
()
{
treeCache
.
close
();
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录