未验证 提交 6c819ee1 编写于 作者: Z zwZjut 提交者: GitHub

[Bug] [dolphinscheduler-server] workflow is always running when worker...

[Bug] [dolphinscheduler-server] workflow is always running when worker sendResult success but master not received  (#7610)

* [Feature][dolphinscheduler-api] parse traceId in http header for Cross system delivery to #7237 (#7238)

* to #7237

* rerun test
Co-authored-by: Nhonghuo.zw <honghuo.zw@alibaba-inc.com>

* chery-pick 05aef279 and handle conflicts

* to #7065: fix ExecutorService and schedulerService (#7072)
Co-authored-by: Nhonghuo.zw <honghuo.zw@alibaba-inc.com>

* [Feature][dolphinscheduler-api] access control of taskDefinition and taskInstance in project to #7081  (#7082)

* to #7081

* fix #7081

* to #7081
Co-authored-by: Nhonghuo.zw <honghuo.zw@alibaba-inc.com>

* chery-pick 8ebe0606 and handle conflicts

* cherry-pick 1f184440 and handle conflicts

* fix #6807: dolphinscheduler.zookeeper.env_vars - > dolphinscheduler.registry.env_vars (#6808)
Co-authored-by: Nhonghuo.zw <honghuo.zw@alibaba-inc.com>
Co-authored-by: NKirs <acm_master@163.com>

* add default constructor (#6780)
Co-authored-by: Nhonghuo.zw <honghuo.zw@alibaba-inc.com>

* to #7108 (#7109)

* to #7609
Co-authored-by: Nhonghuo.zw <honghuo.zw@alibaba-inc.com>
Co-authored-by: NKirs <acm_master@163.com>
上级 f8942bf7
......@@ -47,13 +47,15 @@ public class DBTaskResponseProcessor implements NettyRequestProcessor {
DBTaskResponseCommand taskResponseCommand = JSONUtils.parseObject(
command.getBody(), DBTaskResponseCommand.class);
if (taskResponseCommand == null){
if (taskResponseCommand == null) {
return;
}
if (taskResponseCommand.getStatus() == ExecutionStatus.SUCCESS.getCode()){
if (taskResponseCommand.getStatus() == ExecutionStatus.SUCCESS.getCode()) {
ResponceCache.get().removeResponseCache(taskResponseCommand.getTaskInstanceId());
logger.debug("removeResponseCache: taskinstance id:{}", taskResponseCommand.getTaskInstanceId());
TaskCallbackService.remove(taskResponseCommand.getTaskInstanceId());
logger.debug("remove REMOTE_CHANNELS, task instance id:{}", taskResponseCommand.getTaskInstanceId());
}
}
......
......@@ -19,14 +19,13 @@ package org.apache.dolphinscheduler.server.worker.processor;
import static org.apache.dolphinscheduler.common.Constants.SLEEP_TIME_MILLIS;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.dolphinscheduler.remote.NettyRemotingClient;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.config.NettyClientConfig;
import org.apache.dolphinscheduler.remote.processor.NettyRemoteChannel;
import org.apache.dolphinscheduler.service.registry.RegistryClient;
import java.util.concurrent.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -125,7 +124,7 @@ public class TaskCallbackService {
*
* @param taskInstanceId taskInstanceId
*/
public void remove(int taskInstanceId) {
public static void remove(int taskInstanceId) {
REMOTE_CHANNELS.remove(taskInstanceId);
}
......@@ -156,7 +155,7 @@ public class TaskCallbackService {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
remove(taskInstanceId);
// remove(taskInstanceId);
return;
}
}
......
......@@ -99,6 +99,8 @@ public class TaskKillProcessor implements NettyRequestProcessor {
TaskKillResponseCommand taskKillResponseCommand = buildKillTaskResponseCommand(killCommand, result);
taskCallbackService.sendResult(taskKillResponseCommand.getTaskInstanceId(), taskKillResponseCommand.convert2Command());
TaskExecutionContextCacheManager.removeByTaskInstanceId(taskKillResponseCommand.getTaskInstanceId());
TaskCallbackService.remove(killCommand.getTaskInstanceId());
logger.info("remove REMOTE_CHANNELS, task instance id:{}", killCommand.getTaskInstanceId());
}
/**
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册