|
|
@ -19,11 +19,11 @@ package org.apache.dolphinscheduler.server.worker.processor; |
|
|
|
|
|
|
|
|
|
|
|
import com.google.common.base.Preconditions; |
|
|
|
import com.google.common.base.Preconditions; |
|
|
|
import com.google.common.base.Strings; |
|
|
|
import com.google.common.base.Strings; |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import io.micrometer.core.lang.NonNull; |
|
|
|
import io.netty.channel.Channel; |
|
|
|
import io.netty.channel.Channel; |
|
|
|
import io.netty.channel.ChannelFuture; |
|
|
|
import io.netty.channel.ChannelFuture; |
|
|
|
import io.netty.channel.ChannelFutureListener; |
|
|
|
import io.netty.channel.ChannelFutureListener; |
|
|
|
import lombok.NonNull; |
|
|
|
|
|
|
|
import org.apache.commons.collections4.CollectionUtils; |
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.common.utils.JSONUtils; |
|
|
|
import org.apache.dolphinscheduler.common.utils.JSONUtils; |
|
|
|
import org.apache.dolphinscheduler.common.utils.OSUtils; |
|
|
|
import org.apache.dolphinscheduler.common.utils.OSUtils; |
|
|
|
import org.apache.dolphinscheduler.plugin.task.api.AbstractTask; |
|
|
|
import org.apache.dolphinscheduler.plugin.task.api.AbstractTask; |
|
|
@ -43,6 +43,11 @@ import org.apache.dolphinscheduler.server.worker.message.MessageRetryRunner; |
|
|
|
import org.apache.dolphinscheduler.server.worker.runner.WorkerManagerThread; |
|
|
|
import org.apache.dolphinscheduler.server.worker.runner.WorkerManagerThread; |
|
|
|
import org.apache.dolphinscheduler.server.worker.runner.WorkerTaskExecuteRunnable; |
|
|
|
import org.apache.dolphinscheduler.server.worker.runner.WorkerTaskExecuteRunnable; |
|
|
|
import org.apache.dolphinscheduler.service.log.LogClientService; |
|
|
|
import org.apache.dolphinscheduler.service.log.LogClientService; |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import org.apache.commons.collections.CollectionUtils; |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import org.slf4j.Logger; |
|
|
|
import org.slf4j.Logger; |
|
|
|
import org.slf4j.LoggerFactory; |
|
|
|
import org.slf4j.LoggerFactory; |
|
|
|
import org.springframework.beans.factory.annotation.Autowired; |
|
|
|
import org.springframework.beans.factory.annotation.Autowired; |
|
|
@ -123,7 +128,9 @@ public class TaskKillProcessor implements NettyRequestProcessor { |
|
|
|
private void sendTaskKillResponseCommand(Channel channel, TaskExecutionContext taskExecutionContext) { |
|
|
|
private void sendTaskKillResponseCommand(Channel channel, TaskExecutionContext taskExecutionContext) { |
|
|
|
TaskKillResponseCommand taskKillResponseCommand = new TaskKillResponseCommand(); |
|
|
|
TaskKillResponseCommand taskKillResponseCommand = new TaskKillResponseCommand(); |
|
|
|
taskKillResponseCommand.setStatus(taskExecutionContext.getCurrentExecutionStatus()); |
|
|
|
taskKillResponseCommand.setStatus(taskExecutionContext.getCurrentExecutionStatus()); |
|
|
|
|
|
|
|
if (taskExecutionContext.getAppIds() != null) { |
|
|
|
taskKillResponseCommand.setAppIds(Arrays.asList(taskExecutionContext.getAppIds().split(TaskConstants.COMMA))); |
|
|
|
taskKillResponseCommand.setAppIds(Arrays.asList(taskExecutionContext.getAppIds().split(TaskConstants.COMMA))); |
|
|
|
|
|
|
|
} |
|
|
|
taskKillResponseCommand.setTaskInstanceId(taskExecutionContext.getTaskInstanceId()); |
|
|
|
taskKillResponseCommand.setTaskInstanceId(taskExecutionContext.getTaskInstanceId()); |
|
|
|
taskKillResponseCommand.setHost(taskExecutionContext.getHost()); |
|
|
|
taskKillResponseCommand.setHost(taskExecutionContext.getHost()); |
|
|
|
taskKillResponseCommand.setProcessId(taskExecutionContext.getProcessId()); |
|
|
|
taskKillResponseCommand.setProcessId(taskExecutionContext.getProcessId()); |
|
|
@ -138,9 +145,15 @@ public class TaskKillProcessor implements NettyRequestProcessor { |
|
|
|
}); |
|
|
|
}); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
|
|
|
* do kill |
|
|
|
|
|
|
|
* |
|
|
|
|
|
|
|
* @return kill result |
|
|
|
|
|
|
|
*/ |
|
|
|
private Pair<Boolean, List<String>> doKill(TaskExecutionContext taskExecutionContext) { |
|
|
|
private Pair<Boolean, List<String>> doKill(TaskExecutionContext taskExecutionContext) { |
|
|
|
// kill system process
|
|
|
|
// kill system process
|
|
|
|
boolean processFlag = killProcess(taskExecutionContext.getTenantCode(), taskExecutionContext.getProcessId()); |
|
|
|
boolean processFlag = killProcess(taskExecutionContext.getTenantCode(), taskExecutionContext.getProcessId()); |
|
|
|
|
|
|
|
|
|
|
|
// find log and kill yarn job
|
|
|
|
// find log and kill yarn job
|
|
|
|
Pair<Boolean, List<String>> yarnResult = killYarnJob(Host.of(taskExecutionContext.getHost()), |
|
|
|
Pair<Boolean, List<String>> yarnResult = killYarnJob(Host.of(taskExecutionContext.getHost()), |
|
|
|
taskExecutionContext.getLogPath(), |
|
|
|
taskExecutionContext.getLogPath(), |
|
|
|