seg_yinzy
/
DolphinScheduler
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

optimize worker taskProcessor logger. (#8242)

switch/metrics
yimaixinchen 2022-02-11 09:48:04 +08:00 committed by GitHub
parent 71eeb9c879
commit da09c00cba
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 15 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/processor/DBTaskAckProcessor.java
View File

@ -46,8 +46,10 @@ public class DBTaskAckProcessor implements NettyRequestProcessor {
command.getBody(), DBTaskAckCommand.class);
if (taskAckCommand == null){
logger.error("dBTask ACK request command is null");
return;
}
logger.info("dBTask ACK request command : {}", taskAckCommand);
if (taskAckCommand.getStatus() == ExecutionStatus.SUCCESS.getCode()){
ResponseCache.get().removeAckCache(taskAckCommand.getTaskInstanceId());

3
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/processor/DBTaskResponseProcessor.java
View File

@ -50,8 +50,11 @@ public class DBTaskResponseProcessor implements NettyRequestProcessor {
command.getBody(), DBTaskResponseCommand.class);
if (taskResponseCommand == null) {
logger.error("dBTask Response command is null");
return;
}
logger.info("dBTask Response command : {}", taskResponseCommand);
if (taskResponseCommand.getStatus() == ExecutionStatus.SUCCESS.getCode()) {
ResponseCache.get().removeResponseCache(taskResponseCommand.getTaskInstanceId());

4
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/processor/HostUpdateProcessor.java
View File

@ -49,6 +49,10 @@ public class HostUpdateProcessor implements NettyRequestProcessor {
public void process(Channel channel, Command command) {
Preconditions.checkArgument(CommandType.PROCESS_HOST_UPDATE_REQUEST == command.getType(), String.format("invalid command type : %s", command.getType()));
HostUpdateCommand updateCommand = JSONUtils.parseObject(command.getBody(), HostUpdateCommand.class);
if (updateCommand == null){
logger.error("host update command is null");
return;
}
logger.info("received host update command : {}", updateCommand);
taskCallbackService.changeRemoteChannel(updateCommand.getTaskInstanceId(), new NettyRemoteChannel(channel, command.getOpaque()));

3
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskExecuteProcessor.java
View File

@ -110,12 +110,11 @@ public class TaskExecuteProcessor implements NettyRequestProcessor {
TaskExecuteRequestCommand taskRequestCommand = JSONUtils.parseObject(
command.getBody(), TaskExecuteRequestCommand.class);
logger.info("received command : {}", taskRequestCommand);
if (taskRequestCommand == null) {
logger.error("task execute request command is null");
return;
}
logger.info("task execute request command : {}", taskRequestCommand);
String contextJson = taskRequestCommand.getTaskExecutionContext();
TaskExecutionContext taskExecutionContext = JSONUtils.parseObject(contextJson, TaskExecutionContext.class);

6
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskKillProcessor.java
View File

@ -80,7 +80,11 @@ public class TaskKillProcessor implements NettyRequestProcessor {
public void process(Channel channel, Command command) {
Preconditions.checkArgument(CommandType.TASK_KILL_REQUEST == command.getType(), String.format("invalid command type : %s", command.getType()));
TaskKillRequestCommand killCommand = JSONUtils.parseObject(command.getBody(), TaskKillRequestCommand.class);
logger.info("received kill command : {}", killCommand);
if (killCommand == null){
logger.error("task kill request command is null");
return;
}
logger.info("task kill command : {}", killCommand);
Pair<Boolean, List<String>> result = doKill(killCommand);