[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.

Author: sollhui

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, "

fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java

@@ -40,15 +40,21 @@
 
 @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"));