Browse Source

[Improvement][Master] Calculate the remainTime then we set the delay execution. (#15012)

* [#15007][fix][worker-server] fix bug task delay execution

* [#15007][fix][worker-server] fix bug task delay execution
    mvn spotless:apply

* [#15007][fix][worker-server] fix bug task delay execution
    worker fix  task delay logic

---------

Co-authored-by: 旺阳 <wang@lqwang.net>
3.2.1-prepare
dbac 1 year ago committed by GitHub
parent
commit
d683553324
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 28
      dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/LogicITaskInstanceDispatchOperationFunction.java
  2. 25
      dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/GlobalTaskInstanceDispatchQueueLooper.java

28
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/LogicITaskInstanceDispatchOperationFunction.java

@ -62,19 +62,23 @@ public class LogicITaskInstanceDispatchOperationFunction
LogUtils.setTaskInstanceLogFullPathMDC(taskExecutionContext.getLogPath()); LogUtils.setTaskInstanceLogFullPathMDC(taskExecutionContext.getLogPath());
MasterTaskExecutionContextHolder.putTaskExecutionContext(taskExecutionContext); MasterTaskExecutionContextHolder.putTaskExecutionContext(taskExecutionContext);
// 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(taskExecutionContext.getDelayTime()));
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());
}
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 = final MasterDelayTaskExecuteRunnable masterDelayTaskExecuteRunnable =
masterTaskExecuteRunnableFactoryBuilder masterTaskExecuteRunnableFactoryBuilder
.createWorkerDelayTaskExecuteRunnableFactory(taskExecutionContext.getTaskType()) .createWorkerDelayTaskExecuteRunnableFactory(taskExecutionContext.getTaskType())

25
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/GlobalTaskInstanceDispatchQueueLooper.java

@ -75,16 +75,21 @@ public class GlobalTaskInstanceDispatchQueueLooper extends BaseDaemonThread {
TaskExecutionContext taskExecutionContext = globalTaskInstanceDispatchQueue.take(); TaskExecutionContext taskExecutionContext = globalTaskInstanceDispatchQueue.take();
LogUtils.setTaskInstanceLogFullPathMDC(taskExecutionContext.getLogPath()); LogUtils.setTaskInstanceLogFullPathMDC(taskExecutionContext.getLogPath());
LogUtils.setTaskInstanceIdMDC(taskExecutionContext.getTaskInstanceId()); LogUtils.setTaskInstanceIdMDC(taskExecutionContext.getTaskInstanceId());
// delay task process
long remainTime = int delayTime = taskExecutionContext.getDelayTime();
DateUtils.getRemainTime(DateUtils.timeStampToDate(taskExecutionContext.getFirstSubmitTime()), if (delayTime > 0) {
taskExecutionContext.getDelayTime() * 60L); // delay task process
if (remainTime > 0) { long remainTime =
log.info("Current taskInstance is choose delay execution, delay time: {}s", remainTime); DateUtils.getRemainTime(
taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.DELAY_EXECUTION); DateUtils.timeStampToDate(taskExecutionContext.getFirstSubmitTime()),
// todo: use delay running event delayTime * 60L);
workerMessageSender.sendMessage(taskExecutionContext, if (remainTime > 0) {
ITaskInstanceExecutionEvent.TaskInstanceExecutionEventType.FINISH); 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 WorkerDelayTaskExecuteRunnable workerTaskExecuteRunnable = WorkerTaskExecuteRunnableFactoryBuilder
.createWorkerDelayTaskExecuteRunnableFactory( .createWorkerDelayTaskExecuteRunnableFactory(

Loading…
Cancel
Save