natas_hw
/
dolphinscheduler
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[Improvement] Move delay calculation to Master (#15278)

This commit is contained in:
Wenjun Ruan 2023-12-06 09:37:14 +08:00 committed by GitHub
parent 43f5f24529
commit 2119e41800
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
65 changed files with 401 additions and 577 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskTimeoutStateEventHandler.java
View File

@ -22,8 +22,8 @@ import org.apache.dolphinscheduler.common.enums.TimeoutFlag;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy;
import org.apache.dolphinscheduler.server.master.metrics.TaskMetrics;
import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnable;
import java.util.Map;

43
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/LogicITaskInstanceDispatchOperationFunction.java
View File

@ -17,18 +17,14 @@
package org.apache.dolphinscheduler.server.master.rpc;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskDispatchRequest;
import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskDispatchResponse;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
import org.apache.dolphinscheduler.server.master.runner.MasterDelayTaskExecuteRunnableDelayQueue;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterDelayTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecuteRunnableFactoryBuilder;
import org.apache.dolphinscheduler.server.master.runner.GlobalMasterTaskExecuteRunnableQueue;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecutionContextHolder;
import java.util.concurrent.TimeUnit;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecutor;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecutorFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
@ -42,10 +38,10 @@ public class LogicITaskInstanceDispatchOperationFunction
ITaskInstanceOperationFunction<LogicTaskDispatchRequest, LogicTaskDispatchResponse> {
@Autowired
private MasterTaskExecuteRunnableFactoryBuilder masterTaskExecuteRunnableFactoryBuilder;
private MasterTaskExecutorFactoryBuilder masterTaskExecutorFactoryBuilder;
@Autowired
private MasterDelayTaskExecuteRunnableDelayQueue masterDelayTaskExecuteRunnableDelayQueue;
private GlobalMasterTaskExecuteRunnableQueue globalMasterTaskExecuteRunnableQueue;
@Override
public LogicTaskDispatchResponse operate(LogicTaskDispatchRequest taskDispatchRequest) {
@ -63,34 +59,17 @@ public class LogicITaskInstanceDispatchOperationFunction
MasterTaskExecutionContextHolder.putTaskExecutionContext(taskExecutionContext);
int delayTime = taskExecutionContext.getDelayTime();
if (delayTime > 0) {
// todo: calculate the delay in master dispatcher then we don't need to use a queue to store the task
final long remainTime =
DateUtils.getRemainTime(DateUtils.timeStampToDate(taskExecutionContext.getFirstSubmitTime()),
TimeUnit.SECONDS.toMillis(delayTime));
if (remainTime > 0) {
log.info(
"Current taskInstance: {} is choosing delay execution, delay time: {}/ms, remainTime: {}/ms",
taskExecutionContext.getTaskName(),
TimeUnit.SECONDS.toMillis(taskExecutionContext.getDelayTime()), remainTime);
taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.DELAY_EXECUTION);
// todo: send delay execution message
return LogicTaskDispatchResponse.success(taskExecutionContext.getTaskInstanceId());
}
}
final MasterDelayTaskExecuteRunnable masterDelayTaskExecuteRunnable =
masterTaskExecuteRunnableFactoryBuilder
.createWorkerDelayTaskExecuteRunnableFactory(taskExecutionContext.getTaskType())
.createWorkerTaskExecuteRunnable(taskExecutionContext);
if (masterDelayTaskExecuteRunnableDelayQueue
.submitMasterDelayTaskExecuteRunnable(masterDelayTaskExecuteRunnable)) {
MasterTaskExecutor masterTaskExecutor = masterTaskExecutorFactoryBuilder
.createMasterTaskExecutorFactory(taskExecutionContext.getTaskType())
.createMasterTaskExecutor(taskExecutionContext);
if (globalMasterTaskExecuteRunnableQueue
.submitMasterTaskExecuteRunnable(masterTaskExecutor)) {
log.info("Submit LogicTask: {} to MasterDelayTaskExecuteRunnableDelayQueue success", taskInstanceName);
return LogicTaskDispatchResponse.success(taskInstanceId);
} else {
log.error(
"Submit LogicTask: {} to MasterDelayTaskExecuteRunnableDelayQueue failed, current task waiting queue size: {} is full",
taskInstanceName, masterDelayTaskExecuteRunnableDelayQueue.size());
taskInstanceName, globalMasterTaskExecuteRunnableQueue.size());
return LogicTaskDispatchResponse.failed(taskInstanceId,
"MasterDelayTaskExecuteRunnableDelayQueue is full");
}

22
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/LogicITaskInstanceKillOperationFunction.java
View File

@ -21,10 +21,10 @@ import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskKillReque
import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskKillResponse;
import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
import org.apache.dolphinscheduler.server.master.exception.MasterTaskExecuteException;
import org.apache.dolphinscheduler.server.master.runner.MasterDelayTaskExecuteRunnableDelayQueue;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecuteRunnableHolder;
import org.apache.dolphinscheduler.server.master.runner.GlobalMasterTaskExecuteRunnableQueue;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecutionContextHolder;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecutor;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecutorHolder;
import lombok.extern.slf4j.Slf4j;
@ -38,7 +38,7 @@ public class LogicITaskInstanceKillOperationFunction
ITaskInstanceOperationFunction<LogicTaskKillRequest, LogicTaskKillResponse> {
@Autowired
private MasterDelayTaskExecuteRunnableDelayQueue masterDelayTaskExecuteRunnableDelayQueue;
private GlobalMasterTaskExecuteRunnableQueue globalMasterTaskExecuteRunnableQueue;
@Override
public LogicTaskKillResponse operate(LogicTaskKillRequest taskKillRequest) {
@ -46,16 +46,16 @@ public class LogicITaskInstanceKillOperationFunction
try {
LogUtils.setTaskInstanceIdMDC(taskKillRequest.getTaskInstanceId());
log.info("Received killLogicTask request: {}", taskKillRequest);
final MasterTaskExecuteRunnable masterTaskExecuteRunnable =
MasterTaskExecuteRunnableHolder.getMasterTaskExecuteRunnable(taskInstanceId);
if (masterTaskExecuteRunnable == null) {
final MasterTaskExecutor masterTaskExecutor =
MasterTaskExecutorHolder.getMasterTaskExecutor(taskInstanceId);
if (masterTaskExecutor == null) {
log.error("Cannot find the MasterTaskExecuteRunnable, this task may already been killed");
return LogicTaskKillResponse.fail("Cannot find the MasterTaskExecuteRunnable");
}
try {
masterTaskExecuteRunnable.cancelTask();
masterDelayTaskExecuteRunnableDelayQueue
.removeMasterDelayTaskExecuteRunnable(masterTaskExecuteRunnable);
masterTaskExecutor.cancelTask();
globalMasterTaskExecuteRunnableQueue
.removeMasterTaskExecuteRunnable(masterTaskExecutor);
return LogicTaskKillResponse.success();
} catch (MasterTaskExecuteException e) {
log.error("Cancel MasterTaskExecuteRunnable failed ", e);
@ -63,7 +63,7 @@ public class LogicITaskInstanceKillOperationFunction
} finally {
// todo: If cancel failed, we cannot remove the context?
MasterTaskExecutionContextHolder.removeTaskExecutionContext(taskInstanceId);
MasterTaskExecuteRunnableHolder.removeMasterTaskExecuteRunnable(taskInstanceId);
MasterTaskExecutorHolder.removeMasterTaskExecutor(taskInstanceId);
}
} finally {
LogUtils.removeTaskInstanceIdMDC();

14
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/LogicITaskInstancePauseOperationFunction.java
View File

@ -22,8 +22,8 @@ import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskPauseResp
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
import org.apache.dolphinscheduler.server.master.exception.MasterTaskExecuteException;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecuteRunnableHolder;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecutor;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecutorHolder;
import lombok.extern.slf4j.Slf4j;
@ -39,16 +39,16 @@ public class LogicITaskInstancePauseOperationFunction
public LogicTaskPauseResponse operate(LogicTaskPauseRequest taskPauseRequest) {
try {
LogUtils.setTaskInstanceIdMDC(taskPauseRequest.getTaskInstanceId());
final MasterTaskExecuteRunnable masterTaskExecuteRunnable =
MasterTaskExecuteRunnableHolder.getMasterTaskExecuteRunnable(taskPauseRequest.getTaskInstanceId());
if (masterTaskExecuteRunnable == null) {
final MasterTaskExecutor masterTaskExecutor =
MasterTaskExecutorHolder.getMasterTaskExecutor(taskPauseRequest.getTaskInstanceId());
if (masterTaskExecutor == null) {
log.info("Cannot find the MasterTaskExecuteRunnable");
return LogicTaskPauseResponse.fail("Cannot find the MasterTaskExecuteRunnable");
}
final TaskExecutionContext taskExecutionContext = masterTaskExecuteRunnable.getTaskExecutionContext();
final TaskExecutionContext taskExecutionContext = masterTaskExecutor.getTaskExecutionContext();
LogUtils.setWorkflowAndTaskInstanceIDMDC(taskExecutionContext.getProcessInstanceId(),
taskExecutionContext.getTaskInstanceId());
masterTaskExecuteRunnable.pauseTask();
masterTaskExecutor.pauseTask();
return LogicTaskPauseResponse.success();
} catch (MasterTaskExecuteException e) {
log.error("Pause MasterTaskExecuteRunnable failed", e);

1
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/BaseTaskDispatcher.java
View File

@ -30,7 +30,6 @@ import org.apache.dolphinscheduler.server.master.exception.TaskDispatchException
import org.apache.dolphinscheduler.server.master.processor.queue.TaskEvent;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskEventService;
import org.apache.dolphinscheduler.server.master.runner.dispatcher.TaskDispatcher;
import org.apache.dolphinscheduler.server.master.runner.execute.TaskExecuteRunnable;
import java.util.Date;
import java.util.Optional;

55
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/BaseTaskExecuteRunnable.java Normal file
View File

@ -0,0 +1,55 @@
/*
* 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.runner;
import static com.google.common.base.Preconditions.checkNotNull;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
public abstract class BaseTaskExecuteRunnable implements TaskExecuteRunnable {
protected final ProcessInstance workflowInstance;
protected final TaskInstance taskInstance;
protected final TaskExecutionContext taskExecutionContext;
public BaseTaskExecuteRunnable(ProcessInstance workflowInstance,
TaskInstance taskInstance,
TaskExecutionContext taskExecutionContext) {
this.taskInstance = checkNotNull(taskInstance);
this.workflowInstance = checkNotNull(workflowInstance);
this.taskExecutionContext = checkNotNull(taskExecutionContext);
}
@Override
public ProcessInstance getWorkflowInstance() {
return workflowInstance;
}
@Override
public TaskInstance getTaskInstance() {
return taskInstance;
}
@Override
public TaskExecutionContext getTaskExecutionContext() {
return taskExecutionContext;
}
}

4
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/DefaultTaskExecuteRunnable.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/DefaultTaskExecuteRunnable.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master.runner.execute;
package org.apache.dolphinscheduler.server.master.runner;
import static com.google.common.base.Preconditions.checkNotNull;
@ -24,7 +24,7 @@ import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.server.master.runner.operator.TaskExecuteRunnableOperatorManager;
public class DefaultTaskExecuteRunnable extends PriorityTaskExecuteRunnable {
public class DefaultTaskExecuteRunnable extends PriorityDelayTaskExecuteRunnable {
private final TaskExecuteRunnableOperatorManager taskExecuteRunnableOperatorManager;

52
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/GlobalMasterTaskExecuteRunnableQueue.java Normal file
View File

@ -0,0 +1,52 @@
/*
* 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.runner;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecutor;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.springframework.stereotype.Component;
/**
*
*/
@Component
public class GlobalMasterTaskExecuteRunnableQueue {
private final BlockingQueue<MasterTaskExecutor> masterTaskExecutorBlockingQueue =
new LinkedBlockingQueue<>();
public boolean submitMasterTaskExecuteRunnable(MasterTaskExecutor masterTaskExecutor) {
return masterTaskExecutorBlockingQueue.offer(masterTaskExecutor);
}
public MasterTaskExecutor takeMasterTaskExecuteRunnable() throws InterruptedException {
return masterTaskExecutorBlockingQueue.take();
}
public boolean removeMasterTaskExecuteRunnable(MasterTaskExecutor masterTaskExecutor) {
return masterTaskExecutorBlockingQueue.remove(masterTaskExecutor);
}
public int size() {
return masterTaskExecutorBlockingQueue.size();
}
}

24
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterDelayTaskExecuteRunnableDelayQueueLooper.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/GlobalMasterTaskExecuteRunnableQueueLooper.java
View File

@ -18,9 +18,9 @@
package org.apache.dolphinscheduler.server.master.runner;
import org.apache.dolphinscheduler.common.thread.BaseDaemonThread;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterDelayTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecuteRunnableHolder;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecuteRunnableThreadPool;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecutor;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecutorHolder;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecutorThreadPool;
import java.util.concurrent.atomic.AtomicBoolean;
@ -31,17 +31,17 @@ import org.springframework.stereotype.Component;
@Slf4j
@Component
public class MasterDelayTaskExecuteRunnableDelayQueueLooper extends BaseDaemonThread implements AutoCloseable {
public class GlobalMasterTaskExecuteRunnableQueueLooper extends BaseDaemonThread implements AutoCloseable {
@Autowired
private MasterDelayTaskExecuteRunnableDelayQueue masterDelayTaskExecuteRunnableDelayQueue;
private GlobalMasterTaskExecuteRunnableQueue globalMasterTaskExecuteRunnableQueue;
@Autowired
private MasterTaskExecuteRunnableThreadPool masterTaskExecuteRunnableThreadPool;
private MasterTaskExecutorThreadPool masterTaskExecutorThreadPool;
private final AtomicBoolean RUNNING_FLAG = new AtomicBoolean(false);
public MasterDelayTaskExecuteRunnableDelayQueueLooper() {
public GlobalMasterTaskExecuteRunnableQueueLooper() {
super("MasterDelayTaskExecuteRunnableDelayQueueLooper");
}
@ -53,7 +53,7 @@ public class MasterDelayTaskExecuteRunnableDelayQueueLooper extends BaseDaemonTh
}
log.info("MasterDelayTaskExecuteRunnableDelayQueueLooper starting...");
super.start();
masterTaskExecuteRunnableThreadPool.start();
masterTaskExecutorThreadPool.start();
log.info("MasterDelayTaskExecuteRunnableDelayQueueLooper started...");
}
@ -61,10 +61,10 @@ public class MasterDelayTaskExecuteRunnableDelayQueueLooper extends BaseDaemonTh
public void run() {
while (RUNNING_FLAG.get()) {
try {
final MasterDelayTaskExecuteRunnable masterDelayTaskExecuteRunnable =
masterDelayTaskExecuteRunnableDelayQueue.takeMasterDelayTaskExecuteRunnable();
masterTaskExecuteRunnableThreadPool.submitMasterTaskExecuteRunnable(masterDelayTaskExecuteRunnable);
MasterTaskExecuteRunnableHolder.putMasterTaskExecuteRunnable(masterDelayTaskExecuteRunnable);
final MasterTaskExecutor masterTaskExecutor =
globalMasterTaskExecuteRunnableQueue.takeMasterTaskExecuteRunnable();
masterTaskExecutorThreadPool.submitMasterTaskExecutor(masterTaskExecutor);
MasterTaskExecutorHolder.putMasterTaskExecuteRunnable(masterTaskExecutor);
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
log.warn("MasterDelayTaskExecuteRunnableDelayQueueLooper has been interrupted, will stop loop");

6
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/GlobalTaskDispatchWaitingQueue.java
View File

@ -17,9 +17,7 @@
package org.apache.dolphinscheduler.server.master.runner;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnable;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.DelayQueue;
import lombok.extern.slf4j.Slf4j;
@ -29,7 +27,7 @@ import org.springframework.stereotype.Component;
@Component
public class GlobalTaskDispatchWaitingQueue {
private final PriorityBlockingQueue<DefaultTaskExecuteRunnable> queue = new PriorityBlockingQueue<>();
private final DelayQueue<DefaultTaskExecuteRunnable> queue = new DelayQueue<>();
public void submitNeedToDispatchTaskExecuteRunnable(DefaultTaskExecuteRunnable priorityTaskExecuteRunnable) {
queue.put(priorityTaskExecuteRunnable);

1
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/GlobalTaskDispatchWaitingQueueLooper.java
View File

@ -21,7 +21,6 @@ import org.apache.dolphinscheduler.common.thread.BaseDaemonThread;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.server.master.runner.dispatcher.TaskDispatchFactory;
import org.apache.dolphinscheduler.server.master.runner.dispatcher.TaskDispatcher;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnable;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

54
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterDelayTaskExecuteRunnableDelayQueue.java
View File

@ -1,54 +0,0 @@
/*
* 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.runner;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterDelayTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.execute.MasterTaskExecuteRunnable;
import java.util.concurrent.DelayQueue;
import org.springframework.stereotype.Component;
/**
*
*/
@Component
public class MasterDelayTaskExecuteRunnableDelayQueue {
private final DelayQueue<MasterDelayTaskExecuteRunnable> masterDelayTaskExecuteRunnableDelayQueue =
new DelayQueue<>();
public boolean submitMasterDelayTaskExecuteRunnable(MasterDelayTaskExecuteRunnable masterDelayTaskExecuteRunnable) {
return masterDelayTaskExecuteRunnableDelayQueue.offer(masterDelayTaskExecuteRunnable);
}
public MasterDelayTaskExecuteRunnable takeMasterDelayTaskExecuteRunnable() throws InterruptedException {
return masterDelayTaskExecuteRunnableDelayQueue.take();
}
// todo: if we move the delay process to master, than we don't need this method, since dispatchProcess can directly
// submit to thread pool
public boolean removeMasterDelayTaskExecuteRunnable(MasterTaskExecuteRunnable masterTaskExecuteRunnable) {
return masterDelayTaskExecuteRunnableDelayQueue.remove(masterTaskExecuteRunnable);
}
public int size() {
return masterDelayTaskExecuteRunnableDelayQueue.size();
}
}

8
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterTaskExecutorBootstrap.java
View File

@ -32,7 +32,7 @@ public class MasterTaskExecutorBootstrap implements AutoCloseable {
private GlobalTaskDispatchWaitingQueueLooper globalTaskDispatchWaitingQueueLooper;
@Autowired
private MasterDelayTaskExecuteRunnableDelayQueueLooper masterDelayTaskExecuteRunnableDelayQueueLooper;
private GlobalMasterTaskExecuteRunnableQueueLooper globalMasterTaskExecuteRunnableQueueLooper;
@Autowired
private AsyncMasterTaskDelayQueueLooper asyncMasterTaskDelayQueueLooper;
@ -40,7 +40,7 @@ public class MasterTaskExecutorBootstrap implements AutoCloseable {
public synchronized void start() {
log.info("MasterTaskExecutorBootstrap starting...");
globalTaskDispatchWaitingQueueLooper.start();
masterDelayTaskExecuteRunnableDelayQueueLooper.start();
globalMasterTaskExecuteRunnableQueueLooper.start();
asyncMasterTaskDelayQueueLooper.start();
log.info("MasterTaskExecutorBootstrap started...");
}
@ -51,8 +51,8 @@ public class MasterTaskExecutorBootstrap implements AutoCloseable {
try (
final GlobalTaskDispatchWaitingQueueLooper globalTaskDispatchWaitingQueueLooper1 =
globalTaskDispatchWaitingQueueLooper;
final MasterDelayTaskExecuteRunnableDelayQueueLooper masterDelayTaskExecuteRunnableDelayQueueLooper1 =
masterDelayTaskExecuteRunnableDelayQueueLooper;
final GlobalMasterTaskExecuteRunnableQueueLooper globalMasterTaskExecuteRunnableQueueLooper1 =
globalMasterTaskExecuteRunnableQueueLooper;
final AsyncMasterTaskDelayQueueLooper asyncMasterTaskDelayQueueLooper1 =
asyncMasterTaskDelayQueueLooper) {
// closed the resource

73
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/PriorityTaskExecuteRunnable.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/PriorityDelayTaskExecuteRunnable.java
View File

@ -15,62 +15,57 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master.runner.execute;
import static com.google.common.base.Preconditions.checkNotNull;
package org.apache.dolphinscheduler.server.master.runner;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.jetbrains.annotations.NotNull;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public abstract class PriorityTaskExecuteRunnable implements TaskExecuteRunnable, Comparable<TaskExecuteRunnable> {
public abstract class PriorityDelayTaskExecuteRunnable extends BaseTaskExecuteRunnable implements Delayed {
private final ProcessInstance workflowInstance;
private final TaskInstance taskInstance;
private final TaskExecutionContext taskExecutionContext;
public PriorityTaskExecuteRunnable(ProcessInstance workflowInstance,
TaskInstance taskInstance,
TaskExecutionContext taskExecutionContext) {
this.taskInstance = checkNotNull(taskInstance);
this.workflowInstance = checkNotNull(workflowInstance);
this.taskExecutionContext = checkNotNull(taskExecutionContext);
public PriorityDelayTaskExecuteRunnable(ProcessInstance workflowInstance,
TaskInstance taskInstance,
TaskExecutionContext taskExecutionContext) {
super(workflowInstance, taskInstance, taskExecutionContext);
}
@Override
public ProcessInstance getWorkflowInstance() {
return workflowInstance;
public long getDelay(TimeUnit unit) {
return unit.convert(
DateUtils.getRemainTime(taskExecutionContext.getFirstSubmitTime(),
taskExecutionContext.getDelayTime() * 60L),
TimeUnit.SECONDS);
}
@Override
public TaskInstance getTaskInstance() {
return taskInstance;
}
@Override
public TaskExecutionContext getTaskExecutionContext() {
return taskExecutionContext;
}
@Override
public int compareTo(@NotNull TaskExecuteRunnable other) {
public int compareTo(Delayed o) {
if (o == null) {
return 1;
}
int delayTimeCompareResult =
Long.compare(this.getDelay(TimeUnit.MILLISECONDS), o.getDelay(TimeUnit.MILLISECONDS));
if (delayTimeCompareResult != 0) {
return delayTimeCompareResult;
}
PriorityDelayTaskExecuteRunnable other = (PriorityDelayTaskExecuteRunnable) o;
// the smaller dispatch fail times, the higher priority
int dispatchFailTimesCompareResult = taskExecutionContext.getDispatchFailTimes()
- other.getTaskExecutionContext().getDispatchFailTimes();
if (dispatchFailTimesCompareResult != 0) {
return dispatchFailTimesCompareResult;
}
int workflowInstancePriorityCompareResult = workflowInstance.getProcessInstancePriority().getCode()
- other.getWorkflowInstance().getProcessInstancePriority().getCode();
if (workflowInstancePriorityCompareResult != 0) {
return workflowInstancePriorityCompareResult;
}
int workflowInstanceIdCompareResult = workflowInstance.getId() - other.getWorkflowInstance().getId();
long workflowInstanceIdCompareResult = workflowInstance.getId().compareTo(other.getWorkflowInstance().getId());
if (workflowInstanceIdCompareResult != 0) {
return workflowInstanceIdCompareResult;
return workflowInstancePriorityCompareResult;
}
int taskInstancePriorityCompareResult = taskInstance.getTaskInstancePriority().getCode()
- other.getTaskInstance().getTaskInstancePriority().getCode();
@ -84,21 +79,7 @@ public abstract class PriorityTaskExecuteRunnable implements TaskExecuteRunnable
return -taskGroupPriorityCompareResult;
}
// The task instance shouldn't be equals
return taskInstance.getId() - other.getTaskInstance().getId();
}
@Override
public boolean equals(Object obj) {
if (obj instanceof PriorityTaskExecuteRunnable) {
PriorityTaskExecuteRunnable other = (PriorityTaskExecuteRunnable) obj;
return compareTo(other) == 0;
}
return false;
}
@Override
public int hashCode() {
return taskInstance.getId();
return taskInstance.getId().compareTo(other.getTaskInstance().getId());
}
}

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/StreamTaskExecuteRunnable.java
View File

@ -57,9 +57,7 @@ import org.apache.dolphinscheduler.server.master.event.StateEventHandleException
import org.apache.dolphinscheduler.server.master.metrics.TaskMetrics;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskEvent;
import org.apache.dolphinscheduler.server.master.runner.dispatcher.WorkerTaskDispatcher;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnableFactory;
import org.apache.dolphinscheduler.server.master.runner.execute.TaskExecutionContextFactory;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.spi.enums.ResourceType;

4
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/TaskExecuteRunnable.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/TaskExecuteRunnable.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master.runner.execute;
package org.apache.dolphinscheduler.server.master.runner;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
@ -25,7 +25,7 @@ import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
* This interface is used to define a task which is executing.
* todo: split to MasterTaskExecuteRunnable and WorkerTaskExecuteRunnable
*/
public interface TaskExecuteRunnable extends Comparable<TaskExecuteRunnable> {
public interface TaskExecuteRunnable {
void dispatch();

7
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/TaskExecuteRunnableFactory.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/TaskExecuteRunnableFactory.java
View File

@ -15,11 +15,16 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master.runner.execute;
package org.apache.dolphinscheduler.server.master.runner;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.server.master.exception.TaskExecuteRunnableCreateException;
/**
* Use to create TaskExecuteRunnable
*
* @param <T> TaskExecuteRunnable
*/
public interface TaskExecuteRunnableFactory<T extends TaskExecuteRunnable> {
T createTaskExecuteRunnable(TaskInstance taskInstance) throws TaskExecuteRunnableCreateException;

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/TaskExecutionContextFactory.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/TaskExecutionContextFactory.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master.runner.execute;
package org.apache.dolphinscheduler.server.master.runner;
import static org.apache.dolphinscheduler.common.constants.Constants.ADDRESS;
import static org.apache.dolphinscheduler.common.constants.Constants.DATABASE;

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java
View File

@ -80,9 +80,7 @@ import org.apache.dolphinscheduler.server.master.event.TaskStateEvent;
import org.apache.dolphinscheduler.server.master.event.WorkflowStateEvent;
import org.apache.dolphinscheduler.server.master.graph.IWorkflowGraph;
import org.apache.dolphinscheduler.server.master.metrics.TaskMetrics;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnableFactory;
import org.apache.dolphinscheduler.server.master.runner.execute.TaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.utils.TaskUtils;
import org.apache.dolphinscheduler.server.master.utils.WorkflowInstanceUtils;
import org.apache.dolphinscheduler.service.alert.ListenerEventAlertManager;

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/dispatcher/MasterTaskDispatcher.java
View File

@ -27,7 +27,7 @@ import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.exception.TaskDispatchException;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskEventService;
import org.apache.dolphinscheduler.server.master.runner.BaseTaskDispatcher;
import org.apache.dolphinscheduler.server.master.runner.execute.TaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.TaskExecuteRunnable;
import java.util.Optional;

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/dispatcher/TaskDispatcher.java
View File

@ -19,7 +19,7 @@ package org.apache.dolphinscheduler.server.master.runner.dispatcher;
import org.apache.dolphinscheduler.server.master.dispatch.exceptions.WorkerGroupNotFoundException;
import org.apache.dolphinscheduler.server.master.exception.TaskDispatchException;
import org.apache.dolphinscheduler.server.master.runner.execute.TaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.TaskExecuteRunnable;
/**
* Used to do task dispatcher.

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/dispatcher/WorkerTaskDispatcher.java
View File

@ -31,7 +31,7 @@ import org.apache.dolphinscheduler.server.master.dispatch.host.HostManager;
import org.apache.dolphinscheduler.server.master.exception.TaskDispatchException;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskEventService;
import org.apache.dolphinscheduler.server.master.runner.BaseTaskDispatcher;
import org.apache.dolphinscheduler.server.master.runner.execute.TaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.TaskExecuteRunnable;
import java.util.Optional;

10
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/AsyncMasterDelayTaskExecuteRunnable.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/AsyncMasterTaskExecutor.java
View File

@ -26,14 +26,14 @@ import org.apache.dolphinscheduler.server.master.runner.task.LogicTaskPluginFact
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class AsyncMasterDelayTaskExecuteRunnable extends MasterDelayTaskExecuteRunnable {
public class AsyncMasterTaskExecutor extends MasterTaskExecutor {
private final AsyncMasterTaskDelayQueue asyncMasterTaskDelayQueue;
public AsyncMasterDelayTaskExecuteRunnable(TaskExecutionContext taskExecutionContext,
LogicTaskPluginFactoryBuilder logicTaskPluginFactoryBuilder,
LogicTaskInstanceExecutionEventSenderManager logicTaskInstanceExecutionEventSenderManager,
AsyncMasterTaskDelayQueue asyncTaskDelayQueue) {
public AsyncMasterTaskExecutor(TaskExecutionContext taskExecutionContext,
LogicTaskPluginFactoryBuilder logicTaskPluginFactoryBuilder,
LogicTaskInstanceExecutionEventSenderManager logicTaskInstanceExecutionEventSenderManager,
AsyncMasterTaskDelayQueue asyncTaskDelayQueue) {
super(taskExecutionContext, logicTaskPluginFactoryBuilder, logicTaskInstanceExecutionEventSenderManager);
this.asyncMasterTaskDelayQueue = asyncTaskDelayQueue;
}

8
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/AsyncMasterDelayTaskExecuteRunnableFactory.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/AsyncMasterTaskExecutorFactory.java
View File

@ -25,9 +25,9 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class AsyncMasterDelayTaskExecuteRunnableFactory
public class AsyncMasterTaskExecutorFactory
implements
MasterDelayTaskExecuteRunnableFactory<AsyncMasterDelayTaskExecuteRunnable> {
MasterTaskExecutorFactory<AsyncMasterTaskExecutor> {
@Autowired
private LogicTaskPluginFactoryBuilder logicTaskPluginFactoryBuilder;
@ -39,8 +39,8 @@ public class AsyncMasterDelayTaskExecuteRunnableFactory
private AsyncMasterTaskDelayQueue asyncTaskDelayQueue;
@Override
public AsyncMasterDelayTaskExecuteRunnable createWorkerTaskExecuteRunnable(TaskExecutionContext taskExecutionContext) {
return new AsyncMasterDelayTaskExecuteRunnable(taskExecutionContext,
public AsyncMasterTaskExecutor createMasterTaskExecutor(TaskExecutionContext taskExecutionContext) {
return new AsyncMasterTaskExecutor(taskExecutionContext,
logicTaskPluginFactoryBuilder,
logicTaskInstanceExecutionEventSenderManager,
asyncTaskDelayQueue);

22
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/AsyncTaskCallbackFunctionImpl.java
View File

@ -26,41 +26,41 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class AsyncTaskCallbackFunctionImpl implements AsyncTaskCallbackFunction {
private final AsyncMasterDelayTaskExecuteRunnable asyncMasterDelayTaskExecuteRunnable;
private final AsyncMasterTaskExecutor asyncMasterTaskExecuteRunnable;
public AsyncTaskCallbackFunctionImpl(@NonNull AsyncMasterDelayTaskExecuteRunnable asyncMasterDelayTaskExecuteRunnable) {
this.asyncMasterDelayTaskExecuteRunnable = asyncMasterDelayTaskExecuteRunnable;
public AsyncTaskCallbackFunctionImpl(@NonNull AsyncMasterTaskExecutor asyncMasterTaskExecuteRunnable) {
this.asyncMasterTaskExecuteRunnable = asyncMasterTaskExecuteRunnable;
}
@Override
public void executeSuccess() {
asyncMasterDelayTaskExecuteRunnable.getTaskExecutionContext()
asyncMasterTaskExecuteRunnable.getTaskExecutionContext()
.setCurrentExecutionStatus(TaskExecutionStatus.SUCCESS);
executeFinished();
}
@Override
public void executeFailed() {
asyncMasterDelayTaskExecuteRunnable.getTaskExecutionContext()
asyncMasterTaskExecuteRunnable.getTaskExecutionContext()
.setCurrentExecutionStatus(TaskExecutionStatus.FAILURE);
executeFinished();
}
@Override
public void executeThrowing(Throwable throwable) {
asyncMasterDelayTaskExecuteRunnable.afterThrowing(throwable);
asyncMasterTaskExecuteRunnable.afterThrowing(throwable);
}
private void executeFinished() {
TaskInstanceLogHeader.printFinalizeTaskHeader();
int taskInstanceId = asyncMasterDelayTaskExecuteRunnable.getTaskExecutionContext().getTaskInstanceId();
int taskInstanceId = asyncMasterTaskExecuteRunnable.getTaskExecutionContext().getTaskInstanceId();
MasterTaskExecutionContextHolder.removeTaskExecutionContext(taskInstanceId);
MasterTaskExecuteRunnableHolder.removeMasterTaskExecuteRunnable(taskInstanceId);
MasterTaskExecutorHolder.removeMasterTaskExecutor(taskInstanceId);
log.info("Task execute finished, removed the TaskExecutionContext");
asyncMasterDelayTaskExecuteRunnable.sendTaskResult();
asyncMasterTaskExecuteRunnable.sendTaskResult();
log.info(
"Execute task finished, will send the task execute result to master, the current task execute result is {}",
asyncMasterDelayTaskExecuteRunnable.getTaskExecutionContext().getCurrentExecutionStatus().name());
asyncMasterDelayTaskExecuteRunnable.closeLogAppender();
asyncMasterTaskExecuteRunnable.getTaskExecutionContext().getCurrentExecutionStatus().name());
asyncMasterTaskExecuteRunnable.closeLogAppender();
}
}

3
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/DefaultTaskExecuteRunnableFactory.java
View File

@ -21,6 +21,9 @@ import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.server.master.cache.ProcessInstanceExecCacheManager;
import org.apache.dolphinscheduler.server.master.exception.TaskExecuteRunnableCreateException;
import org.apache.dolphinscheduler.server.master.exception.TaskExecutionContextCreateException;
import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.TaskExecuteRunnableFactory;
import org.apache.dolphinscheduler.server.master.runner.TaskExecutionContextFactory;
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.operator.TaskExecuteRunnableOperatorManager;

68
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/MasterDelayTaskExecuteRunnable.java
View File

@ -1,68 +0,0 @@
/*
* 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.runner.execute;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.server.master.runner.message.LogicTaskInstanceExecutionEventSenderManager;
import org.apache.dolphinscheduler.server.master.runner.task.LogicTaskPluginFactoryBuilder;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public abstract class MasterDelayTaskExecuteRunnable extends MasterTaskExecuteRunnable implements Delayed {
public MasterDelayTaskExecuteRunnable(TaskExecutionContext taskExecutionContext,
LogicTaskPluginFactoryBuilder logicTaskPluginFactoryBuilder,
LogicTaskInstanceExecutionEventSenderManager logicTaskInstanceExecutionEventSenderManager) {
super(taskExecutionContext, logicTaskPluginFactoryBuilder, logicTaskInstanceExecutionEventSenderManager);
}
@Override
public boolean equals(Object obj) {
if (!(obj instanceof MasterDelayTaskExecuteRunnable)) {
return false;
}
MasterDelayTaskExecuteRunnable other = (MasterDelayTaskExecuteRunnable) obj;
return other.getTaskExecutionContext().getTaskInstanceId() == this.getTaskExecutionContext()
.getTaskInstanceId();
}
@Override
public int hashCode() {
return this.getTaskExecutionContext().getTaskInstanceId();
}
@Override
public long getDelay(TimeUnit unit) {
TaskExecutionContext taskExecutionContext = getTaskExecutionContext();
return unit.convert(
DateUtils.getRemainTime(
taskExecutionContext.getFirstSubmitTime(), taskExecutionContext.getDelayTime() * 60L),
TimeUnit.SECONDS);
}
@Override
public int compareTo(Delayed o) {
if (o == null) {
return 1;
}
return Long.compare(this.getDelay(TimeUnit.MILLISECONDS), o.getDelay(TimeUnit.MILLISECONDS));
}
}

10
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/MasterTaskExecuteRunnable.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/MasterTaskExecutor.java
View File

@ -35,16 +35,16 @@ import org.apache.dolphinscheduler.server.master.runner.task.LogicTaskPluginFact
import lombok.extern.slf4j.Slf4j;
@Slf4j
public abstract class MasterTaskExecuteRunnable implements Runnable {
public abstract class MasterTaskExecutor implements Runnable {
protected final TaskExecutionContext taskExecutionContext;
protected final LogicTaskPluginFactoryBuilder logicTaskPluginFactoryBuilder;
protected final LogicTaskInstanceExecutionEventSenderManager logicTaskInstanceExecutionEventSenderManager;
protected ILogicTask logicTask;
public MasterTaskExecuteRunnable(TaskExecutionContext taskExecutionContext,
LogicTaskPluginFactoryBuilder logicTaskPluginFactoryBuilder,
LogicTaskInstanceExecutionEventSenderManager logicTaskInstanceExecutionEventSenderManager) {
public MasterTaskExecutor(TaskExecutionContext taskExecutionContext,
LogicTaskPluginFactoryBuilder logicTaskPluginFactoryBuilder,
LogicTaskInstanceExecutionEventSenderManager logicTaskInstanceExecutionEventSenderManager) {
this.taskExecutionContext = taskExecutionContext;
this.logicTaskPluginFactoryBuilder = logicTaskPluginFactoryBuilder;
this.logicTaskInstanceExecutionEventSenderManager = logicTaskInstanceExecutionEventSenderManager;
@ -68,7 +68,7 @@ public abstract class MasterTaskExecuteRunnable implements Runnable {
"Get a exception when execute the task, sent the task execute result to master, the current task execute result is {}",
taskExecutionContext.getCurrentExecutionStatus());
MasterTaskExecutionContextHolder.removeTaskExecutionContext(taskExecutionContext.getTaskInstanceId());
MasterTaskExecuteRunnableHolder.removeMasterTaskExecuteRunnable(taskExecutionContext.getTaskInstanceId());
MasterTaskExecutorHolder.removeMasterTaskExecutor(taskExecutionContext.getTaskInstanceId());
log.info("Get a exception when execute the task, removed the TaskExecutionContext");
closeLogAppender();
}

4
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/MasterDelayTaskExecuteRunnableFactory.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/MasterTaskExecutorFactory.java
View File

@ -19,8 +19,8 @@ package org.apache.dolphinscheduler.server.master.runner.execute;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
public interface MasterDelayTaskExecuteRunnableFactory<T extends MasterDelayTaskExecuteRunnable> {
public interface MasterTaskExecutorFactory<T extends MasterTaskExecutor> {
T createWorkerTaskExecuteRunnable(TaskExecutionContext taskExecutionContext);
T createMasterTaskExecutor(TaskExecutionContext taskExecutionContext);
}

15
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/MasterTaskExecuteRunnableFactoryBuilder.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/MasterTaskExecutorFactoryBuilder.java
View File

@ -28,24 +28,27 @@ import org.springframework.stereotype.Component;
import com.google.common.collect.Sets;
/**
* Use to create MasterTaskExecutorFactory
*/
@Component
public class MasterTaskExecuteRunnableFactoryBuilder {
public class MasterTaskExecutorFactoryBuilder {
@Autowired
private AsyncMasterDelayTaskExecuteRunnableFactory asyncMasterDelayTaskExecuteRunnableFactory;
private AsyncMasterTaskExecutorFactory asyncMasterTaskExecutorFactory;
@Autowired
private SyncMasterDelayTaskExecuteRunnableFactory syncMasterDelayTaskExecuteRunnableFactory;
private SyncMasterTaskExecutorFactory syncMasterTaskExecutorFactory;
private static final Set<String> ASYNC_TASK_TYPE = Sets.newHashSet(
DependentLogicTask.TASK_TYPE,
SubWorkflowLogicTask.TASK_TYPE,
DynamicLogicTask.TASK_TYPE);
public MasterDelayTaskExecuteRunnableFactory<? extends MasterDelayTaskExecuteRunnable> createWorkerDelayTaskExecuteRunnableFactory(String taskType) {
public MasterTaskExecutorFactory<? extends MasterTaskExecutor> createMasterTaskExecutorFactory(String taskType) {
if (ASYNC_TASK_TYPE.contains(taskType)) {
return asyncMasterDelayTaskExecuteRunnableFactory;
return asyncMasterTaskExecutorFactory;
}
return syncMasterDelayTaskExecuteRunnableFactory;
return syncMasterTaskExecutorFactory;
}
}

14
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/MasterTaskExecuteRunnableHolder.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/MasterTaskExecutorHolder.java
View File

@ -25,20 +25,20 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
@UtilityClass
public class MasterTaskExecuteRunnableHolder {
public class MasterTaskExecutorHolder {
private static final Map<Integer, MasterTaskExecuteRunnable> SUBMITTED_MASTER_TASK_MAP = new ConcurrentHashMap<>();
private static final Map<Integer, MasterTaskExecutor> SUBMITTED_MASTER_TASK_MAP = new ConcurrentHashMap<>();
public void putMasterTaskExecuteRunnable(MasterTaskExecuteRunnable masterTaskExecuteRunnable) {
SUBMITTED_MASTER_TASK_MAP.put(masterTaskExecuteRunnable.getTaskExecutionContext().getTaskInstanceId(),
masterTaskExecuteRunnable);
public void putMasterTaskExecuteRunnable(MasterTaskExecutor masterTaskExecutor) {
SUBMITTED_MASTER_TASK_MAP.put(masterTaskExecutor.getTaskExecutionContext().getTaskInstanceId(),
masterTaskExecutor);
}
public MasterTaskExecuteRunnable getMasterTaskExecuteRunnable(Integer taskInstanceId) {
public MasterTaskExecutor getMasterTaskExecutor(Integer taskInstanceId) {
return SUBMITTED_MASTER_TASK_MAP.get(taskInstanceId);
}
public void removeMasterTaskExecuteRunnable(Integer taskInstanceId) {
public void removeMasterTaskExecutor(Integer taskInstanceId) {
SUBMITTED_MASTER_TASK_MAP.remove(taskInstanceId);
}

6
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/MasterTaskExecuteRunnableThreadPool.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/MasterTaskExecutorThreadPool.java
View File

@ -30,7 +30,7 @@ import com.google.common.util.concurrent.MoreExecutors;
@Slf4j
@Component
public class MasterTaskExecuteRunnableThreadPool {
public class MasterTaskExecutorThreadPool {
@Autowired
private MasterConfig masterConfig;
@ -44,8 +44,8 @@ public class MasterTaskExecuteRunnableThreadPool {
log.info("MasterTaskExecuteRunnableThreadPool started...");
}
public void submitMasterTaskExecuteRunnable(MasterTaskExecuteRunnable masterTaskExecuteRunnable) {
listeningExecutorService.submit(masterTaskExecuteRunnable);
public void submitMasterTaskExecutor(MasterTaskExecutor masterTaskExecutor) {
listeningExecutorService.submit(masterTaskExecutor);
}
}

10
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/SyncMasterDelayTaskExecuteRunnable.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/SyncMasterTaskExecutor.java
View File

@ -27,11 +27,11 @@ import org.apache.dolphinscheduler.server.master.runner.task.LogicTaskPluginFact
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class SyncMasterDelayTaskExecuteRunnable extends MasterDelayTaskExecuteRunnable {
public class SyncMasterTaskExecutor extends MasterTaskExecutor {
public SyncMasterDelayTaskExecuteRunnable(TaskExecutionContext taskExecutionContext,
LogicTaskPluginFactoryBuilder logicTaskPluginFactoryBuilder,
LogicTaskInstanceExecutionEventSenderManager logicTaskInstanceExecutionEventSenderManager) {
public SyncMasterTaskExecutor(TaskExecutionContext taskExecutionContext,
LogicTaskPluginFactoryBuilder logicTaskPluginFactoryBuilder,
LogicTaskInstanceExecutionEventSenderManager logicTaskInstanceExecutionEventSenderManager) {
super(taskExecutionContext, logicTaskPluginFactoryBuilder, logicTaskInstanceExecutionEventSenderManager);
}
@ -56,7 +56,7 @@ public class SyncMasterDelayTaskExecuteRunnable extends MasterDelayTaskExecuteRu
taskExecutionContext.getCurrentExecutionStatus().name());
closeLogAppender();
MasterTaskExecutionContextHolder.removeTaskExecutionContext(taskExecutionContext.getTaskInstanceId());
MasterTaskExecuteRunnableHolder.removeMasterTaskExecuteRunnable(taskExecutionContext.getTaskInstanceId());
MasterTaskExecutorHolder.removeMasterTaskExecutor(taskExecutionContext.getTaskInstanceId());
}
}

8
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/SyncMasterDelayTaskExecuteRunnableFactory.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/SyncMasterTaskExecutorFactory.java
View File

@ -28,9 +28,9 @@ import org.springframework.stereotype.Component;
@Slf4j
@Component
public class SyncMasterDelayTaskExecuteRunnableFactory
public class SyncMasterTaskExecutorFactory
implements
MasterDelayTaskExecuteRunnableFactory<SyncMasterDelayTaskExecuteRunnable> {
MasterTaskExecutorFactory<SyncMasterTaskExecutor> {
@Autowired
private LogicTaskPluginFactoryBuilder logicTaskPluginFactoryBuilder;
@ -38,8 +38,8 @@ public class SyncMasterDelayTaskExecuteRunnableFactory
private LogicTaskInstanceExecutionEventSenderManager logicTaskInstanceExecutionEventSenderManager;
@Override
public SyncMasterDelayTaskExecuteRunnable createWorkerTaskExecuteRunnable(TaskExecutionContext taskExecutionContext) {
return new SyncMasterDelayTaskExecuteRunnable(taskExecutionContext, logicTaskPluginFactoryBuilder,
public SyncMasterTaskExecutor createMasterTaskExecutor(TaskExecutionContext taskExecutionContext) {
return new SyncMasterTaskExecutor(taskExecutionContext, logicTaskPluginFactoryBuilder,
logicTaskInstanceExecutionEventSenderManager);
}
}

26
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/operator/BaseTaskExecuteRunnableDispatchOperator.java
View File

@ -17,19 +17,41 @@
package org.apache.dolphinscheduler.server.master.runner.operator;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.GlobalTaskDispatchWaitingQueue;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnable;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public abstract class BaseTaskExecuteRunnableDispatchOperator implements TaskExecuteRunnableOperator {
private final GlobalTaskDispatchWaitingQueue globalTaskDispatchWaitingQueue;
public BaseTaskExecuteRunnableDispatchOperator(GlobalTaskDispatchWaitingQueue globalTaskDispatchWaitingQueue) {
private final TaskInstanceDao taskInstanceDao;
public BaseTaskExecuteRunnableDispatchOperator(
GlobalTaskDispatchWaitingQueue globalTaskDispatchWaitingQueue,
TaskInstanceDao taskInstanceDao) {
this.globalTaskDispatchWaitingQueue = globalTaskDispatchWaitingQueue;
this.taskInstanceDao = taskInstanceDao;
}
@Override
public void operate(DefaultTaskExecuteRunnable taskExecuteRunnable) {
long remainTime = taskExecuteRunnable.getDelay(TimeUnit.SECONDS);
TaskInstance taskInstance = taskExecuteRunnable.getTaskInstance();
if (remainTime > 0) {
taskInstance.setState(TaskExecutionStatus.DELAY_EXECUTION);
taskInstanceDao.updateById(taskInstance);
log.info("Current taskInstance: {} is choose delay execution, delay time: {}/s, remainTime: {}/s",
taskInstance.getName(),
taskInstance.getDelayTime(), remainTime);
}
globalTaskDispatchWaitingQueue.submitNeedToDispatchTaskExecuteRunnable(taskExecuteRunnable);
}
}

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/operator/BaseTaskExecuteRunnableKillOperator.java
View File

@ -20,7 +20,7 @@ package org.apache.dolphinscheduler.server.master.runner.operator;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
import java.util.Date;

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/operator/BaseTaskExecuteRunnablePauseOperator.java
View File

@ -18,7 +18,7 @@
package org.apache.dolphinscheduler.server.master.runner.operator;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
import lombok.extern.slf4j.Slf4j;

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/operator/BaseTaskExecuteRunnableTimeoutOperator.java
View File

@ -21,7 +21,7 @@ import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
import java.util.Date;

6
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/operator/LogicTaskExecuteRunnableDispatchOperator.java
View File

@ -17,6 +17,7 @@
package org.apache.dolphinscheduler.server.master.runner.operator;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.server.master.runner.GlobalTaskDispatchWaitingQueue;
import org.springframework.stereotype.Component;
@ -24,8 +25,9 @@ import org.springframework.stereotype.Component;
@Component
public class LogicTaskExecuteRunnableDispatchOperator extends BaseTaskExecuteRunnableDispatchOperator {
public LogicTaskExecuteRunnableDispatchOperator(GlobalTaskDispatchWaitingQueue globalTaskDispatchWaitingQueue) {
super(globalTaskDispatchWaitingQueue);
public LogicTaskExecuteRunnableDispatchOperator(GlobalTaskDispatchWaitingQueue globalTaskDispatchWaitingQueue,
TaskInstanceDao taskInstanceDao) {
super(globalTaskDispatchWaitingQueue, taskInstanceDao);
}
}

14
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/operator/TaskExecuteRunnableDispatchOperator.java
View File

@ -17,20 +17,16 @@
package org.apache.dolphinscheduler.server.master.runner.operator;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.server.master.runner.GlobalTaskDispatchWaitingQueue;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnable;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class TaskExecuteRunnableDispatchOperator implements TaskExecuteRunnableOperator {
public class TaskExecuteRunnableDispatchOperator extends BaseTaskExecuteRunnableDispatchOperator {
@Autowired
private GlobalTaskDispatchWaitingQueue globalTaskDispatchWaitingQueue;
@Override
public void operate(DefaultTaskExecuteRunnable taskExecuteRunnable) {
globalTaskDispatchWaitingQueue.submitNeedToDispatchTaskExecuteRunnable(taskExecuteRunnable);
public TaskExecuteRunnableDispatchOperator(GlobalTaskDispatchWaitingQueue globalTaskDispatchWaitingQueue,
TaskInstanceDao taskInstanceDao) {
super(globalTaskDispatchWaitingQueue, taskInstanceDao);
}
}

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/operator/TaskExecuteRunnableOperator.java
View File

@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.master.runner.operator;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
public interface TaskExecuteRunnableOperator {

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/operator/TaskExecuteRunnableOperatorManager.java
View File

@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.master.runner.operator;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.utils.TaskUtils;
import org.springframework.beans.factory.annotation.Autowired;

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/operator/TaskExecuteRunnablePauseOperator.java
View File

@ -22,7 +22,7 @@ import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcCli
import org.apache.dolphinscheduler.extract.worker.ITaskInstanceOperator;
import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstancePauseRequest;
import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstancePauseResponse;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
import org.apache.commons.lang3.StringUtils;

1
dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnableTest.java
View File

@ -39,7 +39,6 @@ import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.graph.IWorkflowGraph;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnableFactory;
import org.apache.dolphinscheduler.server.master.runner.execute.TaskExecuteRunnable;
import org.apache.dolphinscheduler.service.alert.ListenerEventAlertManager;
import org.apache.dolphinscheduler.service.alert.ProcessAlertManager;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;

2
dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/dispatcher/MasterTaskDispatcherTest.java
View File

@ -20,7 +20,7 @@ package org.apache.dolphinscheduler.server.master.runner.dispatcher;
import org.apache.dolphinscheduler.extract.base.utils.Host;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskEventService;
import org.apache.dolphinscheduler.server.master.runner.execute.TaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.TaskExecuteRunnable;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

2
dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/dispatcher/WorkerTaskDispatcherTest.java
View File

@ -23,7 +23,7 @@ import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.dispatch.exceptions.WorkerGroupNotFoundException;
import org.apache.dolphinscheduler.server.master.dispatch.host.HostManager;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskEventService;
import org.apache.dolphinscheduler.server.master.runner.execute.TaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.TaskExecuteRunnable;
import java.util.Optional;

8
dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/execute/PriorityTaskExecuteRunnableTest.java → dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/execute/PriorityDelayTaskExecuteRunnableTest.java
View File

@ -21,12 +21,14 @@ import org.apache.dolphinscheduler.common.enums.Priority;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.PriorityDelayTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.operator.TaskExecuteRunnableOperatorManager;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
public class PriorityTaskExecuteRunnableTest {
public class PriorityDelayTaskExecuteRunnableTest {
@Test
public void testCompareTo() {
@ -46,9 +48,9 @@ public class PriorityTaskExecuteRunnableTest {
TaskExecutionContext context1 = new TaskExecutionContext();
TaskExecutionContext context2 = new TaskExecutionContext();
PriorityTaskExecuteRunnable p1 =
PriorityDelayTaskExecuteRunnable p1 =
new DefaultTaskExecuteRunnable(workflowInstance, t1, context1, taskOperatorManager);
PriorityTaskExecuteRunnable p2 =
PriorityDelayTaskExecuteRunnable p2 =
new DefaultTaskExecuteRunnable(workflowInstance, t2, context2, taskOperatorManager);
Assertions.assertEquals(0, p1.compareTo(p2));

7
dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/shell/BaseLinuxShellInterceptorBuilder.java
View File

@ -17,6 +17,7 @@
package org.apache.dolphinscheduler.plugin.task.api.shell;
import org.apache.dolphinscheduler.common.constants.TenantConstants;
import org.apache.dolphinscheduler.common.exception.FileOperateException;
import org.apache.dolphinscheduler.common.utils.FileUtils;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
@ -67,8 +68,10 @@ public abstract class BaseLinuxShellInterceptorBuilder<T extends BaseLinuxShellI
protected List<String> generateBootstrapCommand() throws FileOperateException {
if (sudoEnable) {
// Set the tenant owner as the working directory
FileUtils.setDirectoryOwner(Paths.get(shellDirectory), runUser);
if (!TenantConstants.BOOTSTRAPT_SYSTEM_USER.equals(runUser)) {
// Set the tenant owner as the working directory
FileUtils.setDirectoryOwner(Paths.get(shellDirectory), runUser);
}
return bootstrapCommandInSudoMode();
}
return bootstrapCommandInNormalMode();

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

@ -29,7 +29,7 @@ import org.apache.dolphinscheduler.plugin.task.api.utils.ProcessUtils;
import org.apache.dolphinscheduler.server.worker.message.MessageRetryRunner;
import org.apache.dolphinscheduler.server.worker.registry.WorkerRegistryClient;
import org.apache.dolphinscheduler.server.worker.rpc.WorkerRpcServer;
import org.apache.dolphinscheduler.server.worker.runner.GlobalTaskInstanceDispatchQueueLooper;
import org.apache.dolphinscheduler.server.worker.runner.GlobalTaskInstanceWaitingQueueLooper;
import org.apache.dolphinscheduler.server.worker.runner.WorkerManagerThread;
import org.apache.commons.collections4.CollectionUtils;
@ -68,7 +68,7 @@ public class WorkerServer implements IStoppable {
private MessageRetryRunner messageRetryRunner;
@Autowired
private GlobalTaskInstanceDispatchQueueLooper globalTaskInstanceDispatchQueueLooper;
private GlobalTaskInstanceWaitingQueueLooper globalTaskInstanceWaitingQueueLooper;
/**
* worker server startup, not use web service
@ -91,7 +91,7 @@ public class WorkerServer implements IStoppable {
this.workerManagerThread.start();
this.messageRetryRunner.start();
this.globalTaskInstanceDispatchQueueLooper.start();
this.globalTaskInstanceWaitingQueueLooper.start();
/*
* registry hooks, which are called before the process exits

6
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerWaitingStrategy.java
View File

@ -25,7 +25,7 @@ import org.apache.dolphinscheduler.registry.api.StrategyType;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.message.MessageRetryRunner;
import org.apache.dolphinscheduler.server.worker.rpc.WorkerRpcServer;
import org.apache.dolphinscheduler.server.worker.runner.GlobalTaskInstanceDispatchQueue;
import org.apache.dolphinscheduler.server.worker.runner.GlobalTaskInstanceWaitingQueue;
import org.apache.dolphinscheduler.server.worker.runner.WorkerManagerThread;
import java.time.Duration;
@ -57,7 +57,7 @@ public class WorkerWaitingStrategy implements WorkerConnectStrategy {
private WorkerManagerThread workerManagerThread;
@Autowired
private GlobalTaskInstanceDispatchQueue globalTaskInstanceDispatchQueue;
private GlobalTaskInstanceWaitingQueue globalTaskInstanceWaitingQueue;
@Override
public void disconnect() {
@ -121,7 +121,7 @@ public class WorkerWaitingStrategy implements WorkerConnectStrategy {
workerRpcServer.close();
log.warn("Worker server close the RPC server due to lost connection from registry");
workerManagerThread.clearTask();
globalTaskInstanceDispatchQueue.clearTask();
globalTaskInstanceWaitingQueue.clearTask();
log.warn("Worker server clear the tasks due to lost connection from registry");
messageRetryRunner.clearMessage();
log.warn("Worker server clear the retry message due to lost connection from registry");

8
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/StreamingTaskInstanceOperatorImpl.java
View File

@ -26,7 +26,7 @@ import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContextCacheMana
import org.apache.dolphinscheduler.plugin.task.api.stream.StreamTask;
import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
import org.apache.dolphinscheduler.server.worker.runner.WorkerManagerThread;
import org.apache.dolphinscheduler.server.worker.runner.WorkerTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.worker.runner.WorkerTaskExecutor;
import lombok.extern.slf4j.Slf4j;
@ -53,12 +53,12 @@ public class StreamingTaskInstanceOperatorImpl implements IStreamingTaskInstance
log.error("Cannot find TaskExecutionContext for taskInstance: {}", taskInstanceId);
return TaskInstanceTriggerSavepointResponse.fail("Cannot find TaskExecutionContext");
}
WorkerTaskExecuteRunnable workerTaskExecuteRunnable = workerManager.getTaskExecuteThread(taskInstanceId);
if (workerTaskExecuteRunnable == null) {
WorkerTaskExecutor workerTaskExecutor = workerManager.getTaskExecuteThread(taskInstanceId);
if (workerTaskExecutor == null) {
log.error("Cannot find WorkerTaskExecuteRunnable for taskInstance: {}", taskInstanceId);
return TaskInstanceTriggerSavepointResponse.fail("Cannot find WorkerTaskExecuteRunnable");
}
AbstractTask task = workerTaskExecuteRunnable.getTask();
AbstractTask task = workerTaskExecutor.getTask();
if (task == null) {
log.error("Cannot find StreamTask for taskInstance:{}", taskInstanceId);
return TaskInstanceTriggerSavepointResponse.fail("Cannot find StreamTask");

59
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnableFactory.java
View File

@ -1,59 +0,0 @@
/*
* 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.worker.runner;
import org.apache.dolphinscheduler.plugin.storage.api.StorageOperate;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.TaskPluginManager;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.registry.WorkerRegistryClient;
import org.apache.dolphinscheduler.server.worker.rpc.WorkerMessageSender;
import javax.annotation.Nullable;
import lombok.NonNull;
public class DefaultWorkerDelayTaskExecuteRunnableFactory
extends
WorkerDelayTaskExecuteRunnableFactory<DefaultWorkerDelayTaskExecuteRunnable> {
protected DefaultWorkerDelayTaskExecuteRunnableFactory(@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull WorkerMessageSender workerMessageSender,
@NonNull TaskPluginManager taskPluginManager,
@Nullable StorageOperate storageOperate,
@NonNull WorkerRegistryClient workerRegistryClient) {
super(taskExecutionContext,
workerConfig,
workerMessageSender,
taskPluginManager,
storageOperate,
workerRegistryClient);
}
@Override
public DefaultWorkerDelayTaskExecuteRunnable createWorkerTaskExecuteRunnable() {
return new DefaultWorkerDelayTaskExecuteRunnable(
taskExecutionContext,
workerConfig,
workerMessageSender,
taskPluginManager,
storageOperate,
workerRegistryClient);
}
}

14
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnable.java → dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerTaskExecutor.java
View File

@ -30,14 +30,14 @@ import javax.annotation.Nullable;
import lombok.NonNull;
public class DefaultWorkerDelayTaskExecuteRunnable extends WorkerDelayTaskExecuteRunnable {
public class DefaultWorkerTaskExecutor extends WorkerTaskExecutor {
public DefaultWorkerDelayTaskExecuteRunnable(@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull WorkerMessageSender workerMessageSender,
@NonNull TaskPluginManager taskPluginManager,
@Nullable StorageOperate storageOperate,
@NonNull WorkerRegistryClient workerRegistryClient) {
public DefaultWorkerTaskExecutor(@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull WorkerMessageSender workerMessageSender,
@NonNull TaskPluginManager taskPluginManager,
@Nullable StorageOperate storageOperate,
@NonNull WorkerRegistryClient workerRegistryClient) {
super(taskExecutionContext,
workerConfig,
workerMessageSender,

40
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerDelayTaskExecuteRunnableFactory.java → dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerTaskExecutorFactory.java
View File

@ -28,24 +28,23 @@ import javax.annotation.Nullable;
import lombok.NonNull;
public abstract class WorkerDelayTaskExecuteRunnableFactory<T extends WorkerDelayTaskExecuteRunnable>
public class DefaultWorkerTaskExecutorFactory
implements
WorkerTaskExecuteRunnableFactory<T> {
WorkerTaskExecutorFactory<DefaultWorkerTaskExecutor> {
protected final @NonNull TaskExecutionContext taskExecutionContext;
protected final @NonNull WorkerConfig workerConfig;
protected final @NonNull WorkerMessageSender workerMessageSender;
protected final @NonNull TaskPluginManager taskPluginManager;
protected final @Nullable StorageOperate storageOperate;
protected final @NonNull WorkerRegistryClient workerRegistryClient;
private final @NonNull TaskExecutionContext taskExecutionContext;
private final @NonNull WorkerConfig workerConfig;
private final @NonNull WorkerMessageSender workerMessageSender;
private final @NonNull TaskPluginManager taskPluginManager;
private final @Nullable StorageOperate storageOperate;
private final @NonNull WorkerRegistryClient workerRegistryClient;
protected WorkerDelayTaskExecuteRunnableFactory(
@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull WorkerMessageSender workerMessageSender,
@NonNull TaskPluginManager taskPluginManager,
@Nullable StorageOperate storageOperate,
@NonNull WorkerRegistryClient workerRegistryClient) {
public DefaultWorkerTaskExecutorFactory(@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull WorkerMessageSender workerMessageSender,
@NonNull TaskPluginManager taskPluginManager,
@Nullable StorageOperate storageOperate,
@NonNull WorkerRegistryClient workerRegistryClient) {
this.taskExecutionContext = taskExecutionContext;
this.workerConfig = workerConfig;
this.workerMessageSender = workerMessageSender;
@ -54,5 +53,14 @@ public abstract class WorkerDelayTaskExecuteRunnableFactory<T extends WorkerDela
this.workerRegistryClient = workerRegistryClient;
}
public abstract T createWorkerTaskExecuteRunnable();
@Override
public DefaultWorkerTaskExecutor createWorkerTaskExecutor() {
return new DefaultWorkerTaskExecutor(
taskExecutionContext,
workerConfig,
workerMessageSender,
taskPluginManager,
storageOperate,
workerRegistryClient);
}
}

4
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/GlobalTaskInstanceDispatchQueue.java → dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/GlobalTaskInstanceWaitingQueue.java
View File

@ -31,13 +31,13 @@ import org.springframework.stereotype.Component;
@Slf4j
@Component
public class GlobalTaskInstanceDispatchQueue {
public class GlobalTaskInstanceWaitingQueue {
private final WorkerConfig workerConfig;
private final BlockingQueue<TaskExecutionContext> blockingQueue;
public GlobalTaskInstanceDispatchQueue(WorkerConfig workerConfig) {
public GlobalTaskInstanceWaitingQueue(WorkerConfig workerConfig) {
this.workerConfig = workerConfig;
this.blockingQueue = new ArrayBlockingQueue<>(workerConfig.getExecThreads());
}

35
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/GlobalTaskInstanceDispatchQueueLooper.java → dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/GlobalTaskInstanceWaitingQueueLooper.java
View File

@ -18,12 +18,9 @@
package org.apache.dolphinscheduler.server.worker.runner;
import org.apache.dolphinscheduler.common.thread.BaseDaemonThread;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.extract.master.transportor.ITaskInstanceExecutionEvent;
import org.apache.dolphinscheduler.plugin.storage.api.StorageOperate;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.TaskPluginManager;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.registry.WorkerRegistryClient;
@ -36,10 +33,10 @@ import org.springframework.stereotype.Component;
@Slf4j
@Component
public class GlobalTaskInstanceDispatchQueueLooper extends BaseDaemonThread {
public class GlobalTaskInstanceWaitingQueueLooper extends BaseDaemonThread {
@Autowired
private GlobalTaskInstanceDispatchQueue globalTaskInstanceDispatchQueue;
private GlobalTaskInstanceWaitingQueue globalTaskInstanceWaitingQueue;
@Autowired
private WorkerConfig workerConfig;
@ -59,7 +56,7 @@ public class GlobalTaskInstanceDispatchQueueLooper extends BaseDaemonThread {
@Autowired
private WorkerRegistryClient workerRegistryClient;
protected GlobalTaskInstanceDispatchQueueLooper() {
protected GlobalTaskInstanceWaitingQueueLooper() {
super("GlobalTaskDispatchQueueLooper");
}
@ -72,39 +69,25 @@ public class GlobalTaskInstanceDispatchQueueLooper extends BaseDaemonThread {
public void run() {
while (true) {
try {
TaskExecutionContext taskExecutionContext = globalTaskInstanceDispatchQueue.take();
TaskExecutionContext taskExecutionContext = globalTaskInstanceWaitingQueue.take();
LogUtils.setTaskInstanceLogFullPathMDC(taskExecutionContext.getLogPath());
LogUtils.setTaskInstanceIdMDC(taskExecutionContext.getTaskInstanceId());
int delayTime = taskExecutionContext.getDelayTime();
if (delayTime > 0) {
// delay task process
long remainTime =
DateUtils.getRemainTime(
DateUtils.timeStampToDate(taskExecutionContext.getFirstSubmitTime()),
delayTime * 60L);
if (remainTime > 0) {
log.info("Current taskInstance is choose delay execution, delay time: {}s", remainTime);
taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.DELAY_EXECUTION);
// todo: use delay running event
workerMessageSender.sendMessage(taskExecutionContext,
ITaskInstanceExecutionEvent.TaskInstanceExecutionEventType.FINISH);
}
}
WorkerDelayTaskExecuteRunnable workerTaskExecuteRunnable = WorkerTaskExecuteRunnableFactoryBuilder
.createWorkerDelayTaskExecuteRunnableFactory(
WorkerTaskExecutor workerTaskExecutor = WorkerTaskExecutorFactoryBuilder
.createWorkerTaskExecutorFactory(
taskExecutionContext,
workerConfig,
workerMessageSender,
taskPluginManager,
storageOperate,
workerRegistryClient)
.createWorkerTaskExecuteRunnable();
if (workerManager.offer(workerTaskExecuteRunnable)) {
.createWorkerTaskExecutor();
if (workerManager.offer(workerTaskExecutor)) {
log.info("Success submit WorkerDelayTaskExecuteRunnable to WorkerManagerThread's waiting queue");
}
} catch (InterruptedException e) {
log.error("GlobalTaskDispatchQueueLooper interrupted");
Thread.currentThread().interrupt();
break;
} catch (Exception ex) {
log.error("GlobalTaskDispatchQueueLooper error", ex);

69
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerDelayTaskExecuteRunnable.java
View File

@ -1,69 +0,0 @@
/*
* 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.worker.runner;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.plugin.storage.api.StorageOperate;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.TaskPluginManager;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.registry.WorkerRegistryClient;
import org.apache.dolphinscheduler.server.worker.rpc.WorkerMessageSender;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import lombok.NonNull;
public abstract class WorkerDelayTaskExecuteRunnable extends WorkerTaskExecuteRunnable implements Delayed {
protected WorkerDelayTaskExecuteRunnable(@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull WorkerMessageSender workerMessageSender,
@NonNull TaskPluginManager taskPluginManager,
@Nullable StorageOperate storageOperate,
@NonNull WorkerRegistryClient workerRegistryClient) {
super(taskExecutionContext,
workerConfig,
workerMessageSender,
taskPluginManager,
storageOperate,
workerRegistryClient);
}
@Override
public long getDelay(TimeUnit unit) {
TaskExecutionContext taskExecutionContext = getTaskExecutionContext();
return unit.convert(
DateUtils.getRemainTime(
DateUtils.timeStampToDate(taskExecutionContext.getFirstSubmitTime()),
taskExecutionContext.getDelayTime() * 60L),
TimeUnit.SECONDS);
}
@Override
public int compareTo(Delayed o) {
if (o == null) {
return 1;
}
return Long.compare(this.getDelay(TimeUnit.MILLISECONDS), o.getDelay(TimeUnit.MILLISECONDS));
}
}

14
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerExecService.java
View File

@ -37,25 +37,19 @@ public class WorkerExecService {
private final ListeningExecutorService listeningExecutorService;
/**
* thread executor service
*/
private final ExecutorService execService;
/**
* running task
*/
private final ConcurrentHashMap<Integer, WorkerTaskExecuteRunnable> taskExecuteThreadMap;
private final ConcurrentHashMap<Integer, WorkerTaskExecutor> taskExecuteThreadMap;
public WorkerExecService(ExecutorService execService,
ConcurrentHashMap<Integer, WorkerTaskExecuteRunnable> taskExecuteThreadMap) {
ConcurrentHashMap<Integer, WorkerTaskExecutor> taskExecuteThreadMap) {
this.execService = execService;
this.listeningExecutorService = MoreExecutors.listeningDecorator(this.execService);
this.taskExecuteThreadMap = taskExecuteThreadMap;
WorkerServerMetrics.registerWorkerTaskTotalGauge(taskExecuteThreadMap::size);
}
public void submit(final WorkerTaskExecuteRunnable taskExecuteThread) {
public void submit(final WorkerTaskExecutor taskExecuteThread) {
taskExecuteThreadMap.put(taskExecuteThread.getTaskExecutionContext().getTaskInstanceId(), taskExecuteThread);
ListenableFuture future = this.listeningExecutorService.submit(taskExecuteThread);
FutureCallback futureCallback = new FutureCallback() {
@ -90,7 +84,7 @@ public class WorkerExecService {
return ((ThreadPoolExecutor) this.execService).getActiveCount();
}
public Map<Integer, WorkerTaskExecuteRunnable> getTaskExecuteThreadMap() {
public Map<Integer, WorkerTaskExecutor> getTaskExecuteThreadMap() {
return taskExecuteThreadMap;
}

23
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerManagerThread.java
View File

@ -25,8 +25,9 @@ import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContextCacheMana
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.metrics.WorkerServerMetrics;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.LinkedBlockingQueue;
import javax.annotation.Nullable;
@ -41,28 +42,22 @@ import org.springframework.stereotype.Component;
@Slf4j
public class WorkerManagerThread implements Runnable {
private final DelayQueue<WorkerDelayTaskExecuteRunnable> waitSubmitQueue;
private final BlockingQueue<WorkerTaskExecutor> waitSubmitQueue;
private final WorkerExecService workerExecService;
private final WorkerConfig workerConfig;
private final int workerExecThreads;
/**
* running task
*/
private final ConcurrentHashMap<Integer, WorkerTaskExecuteRunnable> taskExecuteThreadMap =
new ConcurrentHashMap<>();
private final ConcurrentHashMap<Integer, WorkerTaskExecutor> taskExecuteThreadMap = new ConcurrentHashMap<>();
public WorkerManagerThread(WorkerConfig workerConfig) {
this.workerConfig = workerConfig;
workerExecThreads = workerConfig.getExecThreads();
this.waitSubmitQueue = new DelayQueue<>();
this.waitSubmitQueue = new LinkedBlockingQueue<>();
workerExecService = new WorkerExecService(
ThreadUtils.newDaemonFixedThreadExecutor("Worker-Execute-Thread", workerConfig.getExecThreads()),
taskExecuteThreadMap);
}
public @Nullable WorkerTaskExecuteRunnable getTaskExecuteThread(Integer taskInstanceId) {
public @Nullable WorkerTaskExecutor getTaskExecuteThread(Integer taskInstanceId) {
return taskExecuteThreadMap.get(taskInstanceId);
}
@ -95,7 +90,7 @@ public class WorkerManagerThread implements Runnable {
.forEach(waitSubmitQueue::remove);
}
public boolean offer(WorkerDelayTaskExecuteRunnable workerDelayTaskExecuteRunnable) {
public boolean offer(WorkerTaskExecutor workerDelayTaskExecuteRunnable) {
return waitSubmitQueue.add(workerDelayTaskExecuteRunnable);
}
@ -122,8 +117,8 @@ public class WorkerManagerThread implements Runnable {
Thread.sleep(Constants.SLEEP_TIME_MILLIS);
}
if (this.getThreadPoolQueueSize() <= workerExecThreads) {
final WorkerDelayTaskExecuteRunnable workerDelayTaskExecuteRunnable = waitSubmitQueue.take();
workerExecService.submit(workerDelayTaskExecuteRunnable);
WorkerTaskExecutor workerTaskExecutor = waitSubmitQueue.take();
workerExecService.submit(workerTaskExecutor);
} else {
WorkerServerMetrics.incWorkerOverloadCount();
log.info("Exec queue is full, waiting submit queue {}, waiting exec queue size {}",

18
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecuteRunnable.java → dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutor.java
View File

@ -67,9 +67,9 @@ import org.slf4j.LoggerFactory;
import com.google.common.base.Strings;
public abstract class WorkerTaskExecuteRunnable implements Runnable {
public abstract class WorkerTaskExecutor implements Runnable {
protected static final Logger log = LoggerFactory.getLogger(WorkerTaskExecuteRunnable.class);
protected static final Logger log = LoggerFactory.getLogger(WorkerTaskExecutor.class);
protected final TaskExecutionContext taskExecutionContext;
protected final WorkerConfig workerConfig;
@ -80,13 +80,13 @@ public abstract class WorkerTaskExecuteRunnable implements Runnable {
protected @Nullable AbstractTask task;
protected WorkerTaskExecuteRunnable(
@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull WorkerMessageSender workerMessageSender,
@NonNull TaskPluginManager taskPluginManager,
@Nullable StorageOperate storageOperate,
@NonNull WorkerRegistryClient workerRegistryClient) {
protected WorkerTaskExecutor(
@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull WorkerMessageSender workerMessageSender,
@NonNull TaskPluginManager taskPluginManager,
@Nullable StorageOperate storageOperate,
@NonNull WorkerRegistryClient workerRegistryClient) {
this.taskExecutionContext = taskExecutionContext;
this.workerConfig = workerConfig;
this.workerMessageSender = workerMessageSender;

4
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecuteRunnableFactory.java → dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutorFactory.java
View File

@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.worker.runner;
public interface WorkerTaskExecuteRunnableFactory<T> {
public interface WorkerTaskExecutorFactory<T> {
T createWorkerTaskExecuteRunnable();
T createWorkerTaskExecutor();
}

16
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecuteRunnableFactoryBuilder.java → dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutorFactoryBuilder.java
View File

@ -30,15 +30,15 @@ import lombok.NonNull;
import lombok.experimental.UtilityClass;
@UtilityClass
public class WorkerTaskExecuteRunnableFactoryBuilder {
public class WorkerTaskExecutorFactoryBuilder {
public static WorkerDelayTaskExecuteRunnableFactory<?> createWorkerDelayTaskExecuteRunnableFactory(@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull WorkerMessageSender workerMessageSender,
@NonNull TaskPluginManager taskPluginManager,
@Nullable StorageOperate storageOperate,
@NonNull WorkerRegistryClient workerRegistryClient) {
return new DefaultWorkerDelayTaskExecuteRunnableFactory(taskExecutionContext,
public static WorkerTaskExecutorFactory<? extends WorkerTaskExecutor> createWorkerTaskExecutorFactory(@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull WorkerMessageSender workerMessageSender,
@NonNull TaskPluginManager taskPluginManager,
@Nullable StorageOperate storageOperate,
@NonNull WorkerRegistryClient workerRegistryClient) {
return new DefaultWorkerTaskExecutorFactory(taskExecutionContext,
workerConfig,
workerMessageSender,
taskPluginManager,

6
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceDispatchOperationFunction.java
View File

@ -24,7 +24,7 @@ import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContextCacheMana
import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.metrics.TaskMetrics;
import org.apache.dolphinscheduler.server.worker.runner.GlobalTaskInstanceDispatchQueue;
import org.apache.dolphinscheduler.server.worker.runner.GlobalTaskInstanceWaitingQueue;
import lombok.extern.slf4j.Slf4j;
@ -41,7 +41,7 @@ public class TaskInstanceDispatchOperationFunction
private WorkerConfig workerConfig;
@Autowired
private GlobalTaskInstanceDispatchQueue globalTaskInstanceDispatchQueue;
private GlobalTaskInstanceWaitingQueue globalTaskInstanceWaitingQueue;
@Override
public TaskInstanceDispatchResponse operate(TaskInstanceDispatchRequest taskInstanceDispatchRequest) {
@ -57,7 +57,7 @@ public class TaskInstanceDispatchOperationFunction
taskExecutionContext.getTaskInstanceId());
TaskMetrics.incrTaskTypeExecuteCount(taskExecutionContext.getTaskType());
if (!globalTaskInstanceDispatchQueue.addDispatchTask(taskExecutionContext)) {
if (!globalTaskInstanceWaitingQueue.addDispatchTask(taskExecutionContext)) {
log.error("Submit task: {} to wait queue error, current queue size: {} is full",
taskExecutionContext.getTaskName(), workerConfig.getExecThreads());
return TaskInstanceDispatchResponse.failed(taskExecutionContext.getTaskInstanceId(),

8
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceKillOperationFunction.java
View File

@ -29,7 +29,7 @@ import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
import org.apache.dolphinscheduler.plugin.task.api.utils.ProcessUtils;
import org.apache.dolphinscheduler.server.worker.message.MessageRetryRunner;
import org.apache.dolphinscheduler.server.worker.runner.WorkerManagerThread;
import org.apache.dolphinscheduler.server.worker.runner.WorkerTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.worker.runner.WorkerTaskExecutor;
import lombok.extern.slf4j.Slf4j;
@ -103,12 +103,12 @@ public class TaskInstanceKillOperationFunction
}
protected void cancelApplication(int taskInstanceId) {
WorkerTaskExecuteRunnable workerTaskExecuteRunnable = workerManager.getTaskExecuteThread(taskInstanceId);
if (workerTaskExecuteRunnable == null) {
WorkerTaskExecutor workerTaskExecutor = workerManager.getTaskExecuteThread(taskInstanceId);
if (workerTaskExecutor == null) {
log.warn("taskExecuteThread not found, taskInstanceId:{}", taskInstanceId);
return;
}
AbstractTask task = workerTaskExecuteRunnable.getTask();
AbstractTask task = workerTaskExecutor.getTask();
if (task == null) {
log.warn("task not found, taskInstanceId:{}", taskInstanceId);
return;

10
dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnableTest.java → dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerTaskExecutorTest.java
View File

@ -30,7 +30,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
public class DefaultWorkerDelayTaskExecuteRunnableTest {
public class DefaultWorkerTaskExecutorTest {
private TaskExecutionContext taskExecutionContext = Mockito.mock(TaskExecutionContext.class);
@ -54,7 +54,7 @@ public class DefaultWorkerDelayTaskExecuteRunnableTest {
.processDefineId(0)
.firstSubmitTime(System.currentTimeMillis())
.build();
WorkerTaskExecuteRunnable workerTaskExecuteRunnable = new DefaultWorkerDelayTaskExecuteRunnable(
WorkerTaskExecutor workerTaskExecutor = new DefaultWorkerTaskExecutor(
taskExecutionContext,
workerConfig,
workerMessageSender,
@ -62,7 +62,7 @@ public class DefaultWorkerDelayTaskExecuteRunnableTest {
storageOperate,
workerRegistryClient);
Assertions.assertAll(workerTaskExecuteRunnable::run);
Assertions.assertAll(workerTaskExecutor::run);
Assertions.assertEquals(TaskExecutionStatus.SUCCESS, taskExecutionContext.getCurrentExecutionStatus());
}
@ -78,7 +78,7 @@ public class DefaultWorkerDelayTaskExecuteRunnableTest {
.taskParams(
"{\"localParams\":[],\"resourceList\":[],\"type\":\"POSTGRESQL\",\"datasource\":null,\"sql\":\"select * from t_ds_user\",\"sqlType\":\"0\",\"preStatements\":[],\"postStatements\":[],\"segmentSeparator\":\"\",\"displayRows\":10,\"conditionResult\":\"null\",\"dependence\":\"null\",\"switchResult\":\"null\",\"waitStartTimeout\":null}")
.build();
WorkerTaskExecuteRunnable workerTaskExecuteRunnable = new DefaultWorkerDelayTaskExecuteRunnable(
WorkerTaskExecutor workerTaskExecutor = new DefaultWorkerTaskExecutor(
taskExecutionContext,
workerConfig,
workerMessageSender,
@ -86,7 +86,7 @@ public class DefaultWorkerDelayTaskExecuteRunnableTest {
storageOperate,
workerRegistryClient);
Assertions.assertAll(workerTaskExecuteRunnable::run);
Assertions.assertAll(workerTaskExecutor::run);
Assertions.assertEquals(TaskExecutionStatus.FAILURE, taskExecutionContext.getCurrentExecutionStatus());
}
}