This is an automated email from the ASF dual-hosted git repository.
kerwin pushed a commit to branch 3.1.1-prepare
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/3.1.1-prepare by this push:
new 75f6c416fb Add task executor threads full policy config in worker
(#12510)
75f6c416fb is described below
commit 75f6c416fb8bf2a1fddddeaf779957cc3917e65a
Author: Wenjun Ruan <[email protected]>
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 6a0438aebb..661481767d 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 184053b68a..cf57b32791 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 5b4c6fdac3..9a7b455956 100644
--- a/dolphinscheduler-standalone-server/src/main/resources/application.yaml
+++ b/dolphinscheduler-standalone-server/src/main/resources/application.yaml
@@ -175,6 +175,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 96f00a1f30..e9bf08b575 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 1e6a0ba6b7..f360dee177 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;
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;
import org.slf4j.Logger;
@@ -40,8 +41,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;
@@ -51,6 +52,7 @@ public class WorkerManagerThread implements Runnable {
private final ConcurrentHashMap<Integer, WorkerTaskExecuteRunnable>
taskExecuteThreadMap = new ConcurrentHashMap<>();
public WorkerManagerThread(WorkerConfig workerConfig) {
+ this.workerConfig = workerConfig;
workerExecThreads = workerConfig.getExecThreads();
this.waitSubmitQueue = new DelayQueue<>();
workerExecService = new WorkerExecService(
@@ -92,6 +94,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
