You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by ke...@apache.org on 2022/10/26 02:29:29 UTC
[dolphinscheduler] branch dev updated: Add task executor threads full policy config in worker (#12510)
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 76a9e6d8a6 Add task executor threads full policy config in worker (#12510)
76a9e6d8a6 is described below
commit 76a9e6d8a681e3d4cba7be8543c8a14e16e0afda
Author: Wenjun Ruan <we...@apache.org>
AuthorDate: Wed Oct 26 10:29:22 2022 +0800
Add task executor threads full policy config in worker (#12510)
---
docs/docs/en/architecture/configuration.md | 1 +
docs/docs/zh/architecture/configuration.md | 1 +
.../src/main/resources/application.yaml | 1 +
.../config/TaskExecuteThreadsFullPolicy.java | 24 ++++++++++++++++++++++
.../server/worker/config/WorkerConfig.java | 3 +++
.../server/worker/runner/WorkerManagerThread.java | 8 +++++++-
.../src/main/resources/application.yaml | 1 +
7 files changed, 38 insertions(+), 1 deletion(-)
diff --git a/docs/docs/en/architecture/configuration.md b/docs/docs/en/architecture/configuration.md
index 1a257dc92e..1399ff69be 100644
--- a/docs/docs/en/architecture/configuration.md
+++ b/docs/docs/en/architecture/configuration.md
@@ -295,6 +295,7 @@ Location: `worker-server/conf/application.yaml`
|worker.alert-listen-port|50052|the alert listen port of worker|
|worker.registry-disconnect-strategy.strategy|stop|Used when the worker disconnect from registry, default value: stop. Optional values include stop, waiting|
|worker.registry-disconnect-strategy.max-waiting-time|100s|Used when the worker disconnect from registry, and the disconnect strategy is waiting, this config means the worker will waiting to reconnect to registry in given times, and after the waiting times, if the worker still cannot connect to registry, will stop itself, if the value is 0s, will waitting infinitely |
+|worker.task-execute-threads-full-policy|REJECT|If REJECT, when the task waiting in the worker reaches exec-threads, it will reject the received task and the Master will redispatch it; If CONTINUE, it will put the task into the worker's execution queue and wait for a free thread to start execution|
### Alert Server related configuration
diff --git a/docs/docs/zh/architecture/configuration.md b/docs/docs/zh/architecture/configuration.md
index 691a253a50..eafcb1c8cb 100644
--- a/docs/docs/zh/architecture/configuration.md
+++ b/docs/docs/zh/architecture/configuration.md
@@ -289,6 +289,7 @@ common.properties配置文件目前主要是配置hadoop/s3/yarn相关的配置
|worker.alert-listen-port|50052|alert监听端口|
|worker.registry-disconnect-strategy.strategy|stop|当Worker与注册中心失联之后采取的策略, 默认值是: stop. 可选值包括： stop, waiting|
|worker.registry-disconnect-strategy.max-waiting-time|100s|当Worker与注册中心失联之后重连时间, 之后当strategy为waiting时，该值生效。 该值表示当Worker与注册中心失联时会在给定时间之内进行重连, 在给定时间之内重连失败将会停止自己，在重连时，Worker会丢弃kill正在执行的任务。值为0表示会无限期等待 |
+|worker.task-execute-threads-full-policy|REJECT|如果是 REJECT, 当Worker中等待队列中的任务数达到exec-threads时, Worker将会拒绝接下来新接收的任务，Master将会重新分发该任务; 如果是 CONTINUE, Worker将会接收任务，放入等待队列中等待空闲线程去执行该任务|
## Alert Server相关配置
diff --git a/dolphinscheduler-standalone-server/src/main/resources/application.yaml b/dolphinscheduler-standalone-server/src/main/resources/application.yaml
index 4fde80f141..2d9674cb6e 100644
--- a/dolphinscheduler-standalone-server/src/main/resources/application.yaml
+++ b/dolphinscheduler-standalone-server/src/main/resources/application.yaml
@@ -176,6 +176,7 @@ worker:
# alert server listen host
alert-listen-host: localhost
alert-listen-port: 50052
+ task-execute-threads-full-policy: REJECT
alert:
port: 50052
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/config/TaskExecuteThreadsFullPolicy.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/config/TaskExecuteThreadsFullPolicy.java
new file mode 100644
index 0000000000..437acc3e7f
--- /dev/null
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/config/TaskExecuteThreadsFullPolicy.java
@@ -0,0 +1,24 @@
+/*
+ * 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.config;
+
+public enum TaskExecuteThreadsFullPolicy {
+ CONTINUE,
+ REJECT,
+ ;
+}
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/config/WorkerConfig.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/config/WorkerConfig.java
index 73c2e5ca11..02728136de 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/config/WorkerConfig.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/config/WorkerConfig.java
@@ -60,6 +60,8 @@ public class WorkerConfig implements Validator {
private String workerAddress;
private String workerRegistryPath;
+ private TaskExecuteThreadsFullPolicy taskExecuteThreadsFullPolicy = TaskExecuteThreadsFullPolicy.REJECT;
+
@Override
public boolean supports(Class<?> clazz) {
return WorkerConfig.class.isAssignableFrom(clazz);
@@ -97,5 +99,6 @@ public class WorkerConfig implements Validator {
logger.info("Worker config: registryDisconnectStrategy -> {}", registryDisconnectStrategy);
logger.info("Worker config: workerAddress -> {}", registryDisconnectStrategy);
logger.info("Worker config: workerRegistryPath: {}", workerRegistryPath);
+ logger.info("Worker config: taskExecuteThreadsFullPolicy: {}", taskExecuteThreadsFullPolicy);
}
}
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerManagerThread.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerManagerThread.java
index c403d2e6ce..d30f5a53c1 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerManagerThread.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerManagerThread.java
@@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.lifecycle.ServerLifeCycleManager;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContextCacheManager;
+import org.apache.dolphinscheduler.server.worker.config.TaskExecuteThreadsFullPolicy;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.metrics.WorkerServerMetrics;
@@ -42,8 +43,8 @@ public class WorkerManagerThread implements Runnable {
private final Logger logger = LoggerFactory.getLogger(WorkerManagerThread.class);
private final DelayQueue<WorkerDelayTaskExecuteRunnable> waitSubmitQueue;
-
private final WorkerExecService workerExecService;
+ private final WorkerConfig workerConfig;
private final int workerExecThreads;
@@ -54,6 +55,7 @@ public class WorkerManagerThread implements Runnable {
new ConcurrentHashMap<>();
public WorkerManagerThread(WorkerConfig workerConfig) {
+ this.workerConfig = workerConfig;
workerExecThreads = workerConfig.getExecThreads();
this.waitSubmitQueue = new DelayQueue<>();
workerExecService = new WorkerExecService(
@@ -95,6 +97,10 @@ public class WorkerManagerThread implements Runnable {
}
public boolean offer(WorkerDelayTaskExecuteRunnable workerDelayTaskExecuteRunnable) {
+ if (workerConfig.getTaskExecuteThreadsFullPolicy() == TaskExecuteThreadsFullPolicy.CONTINUE) {
+ return waitSubmitQueue.offer(workerDelayTaskExecuteRunnable);
+ }
+
if (waitSubmitQueue.size() > workerExecThreads) {
logger.warn("Wait submit queue is full, will retry submit task later");
WorkerServerMetrics.incWorkerSubmitQueueIsFullCount();
diff --git a/dolphinscheduler-worker/src/main/resources/application.yaml b/dolphinscheduler-worker/src/main/resources/application.yaml
index d64f4fde2c..d895569490 100644
--- a/dolphinscheduler-worker/src/main/resources/application.yaml
+++ b/dolphinscheduler-worker/src/main/resources/application.yaml
@@ -76,6 +76,7 @@ worker:
strategy: waiting
# The max waiting time to reconnect to registry if you set the strategy to waiting
max-waiting-time: 100s
+ task-execute-threads-full-policy: REJECT
server:
port: 1235