This is an automated email from the ASF dual-hosted git repository.
liaoxin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 37a8ca8323a [enhance](job) terminate streaming task execute threads
promptly when idle (#58041)
37a8ca8323a is described below
commit 37a8ca8323a3f446caa6177b5482eca93386d7c1
Author: hui lai <[email protected]>
AuthorDate: Sun Nov 16 21:57:47 2025 +0800
[enhance](job) terminate streaming task execute threads promptly when idle
(#58041)
1. terminate streaming task execute threads promptly when idle.
2. set job_streaming_task_exec_thread_num default value from 10 to 100.
---
.../main/java/org/apache/doris/common/Config.java | 6 +++---
.../job/scheduler/StreamingTaskScheduler.java | 24 ++++++++++++++--------
2 files changed, 18 insertions(+), 12 deletions(-)
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 0695228e9bc..8c4f8881208 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -1988,10 +1988,10 @@ public class Config extends ConfigBase {
+ " greater than 0, otherwise it defaults to 3." })
public static int job_dictionary_task_consumer_thread_num = 3;
- @ConfField(masterOnly = true, description = {"用于执行 Streaming
任务的线程数,值应该大于0，否则默认为10",
+ @ConfField(masterOnly = true, description = {"用于执行 Streaming
任务的线程数,值应该大于0，否则默认为100",
"The number of threads used to execute Streaming Tasks, "
- + "the value should be greater than 0, if it is <=0,
default is 10."})
- public static int job_streaming_task_exec_thread_num = 10;
+ + "the value should be greater than 0, if it is <=0,
default is 100."})
+ public static int job_streaming_task_exec_thread_num = 100;
@ConfField(masterOnly = true, description = {"最大的 Streaming
作业数量,值应该大于0，否则默认为1024",
"The maximum number of Streaming jobs, "
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java
b/fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java
index 51e17f214c6..ea1cf7825ec 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java
@@ -40,15 +40,21 @@ import java.util.concurrent.TimeUnit;
@Log4j2
public class StreamingTaskScheduler extends MasterDaemon {
- private final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
- Config.job_streaming_task_exec_thread_num,
- Config.job_streaming_task_exec_thread_num,
- 0,
- TimeUnit.SECONDS,
- new ArrayBlockingQueue<>(Config.max_streaming_job_num),
- new CustomThreadFactory("streaming-task-execute"),
- new ThreadPoolExecutor.AbortPolicy()
- );
+ private final ThreadPoolExecutor threadPool;
+
+ {
+ threadPool = new ThreadPoolExecutor(
+ Config.job_streaming_task_exec_thread_num,
+ Config.job_streaming_task_exec_thread_num,
+ 60L,
+ TimeUnit.SECONDS,
+ new ArrayBlockingQueue<>(Config.max_streaming_job_num),
+ new CustomThreadFactory("streaming-task-execute"),
+ new ThreadPoolExecutor.AbortPolicy()
+ );
+ threadPool.allowCoreThreadTimeOut(true);
+ }
+
private final ScheduledThreadPoolExecutor delayScheduler
= new ScheduledThreadPoolExecutor(1, new
CustomThreadFactory("streaming-task-delay-scheduler"));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
