fix(core): max thread on worker is not respected

kestra-io · Mar 11, 2022 · 51508c2 · 51508c2
1 parent f31db14
commit 51508c2
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 15 deletions.
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
@@ -23,7 +23,7 @@ jobs:
   check:
     name: Check & Publish
     runs-on: ubuntu-latest
-    timeout-minutes: 30
+    timeout-minutes: 45
     env:
       SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
     steps:

diff --git a/core/src/main/java/io/kestra/core/runners/Worker.java b/core/src/main/java/io/kestra/core/runners/Worker.java
@@ -73,11 +73,7 @@ public Worker(ApplicationContext applicationContext, int thread) {
         this.metricRegistry = applicationContext.getBean(MetricRegistry.class);
 
         ExecutorsUtils executorsUtils = applicationContext.getBean(ExecutorsUtils.class);
-        this.executors = executorsUtils.maxCachedThreadPool(
-            Math.min(Runtime.getRuntime().availableProcessors(), thread),
-            thread,
-            "worker"
-        );
+        this.executors = executorsUtils.maxCachedThreadPool(thread,"worker");
     }
 
     @Override

diff --git a/core/src/main/java/io/kestra/core/utils/ExecutorsUtils.java b/core/src/main/java/io/kestra/core/utils/ExecutorsUtils.java
@@ -25,17 +25,21 @@ public ExecutorService cachedThreadPool(String name) {
         );
     }
 
-    public ExecutorService maxCachedThreadPool(int minThread, int maxThread, String name) {
+    public ExecutorService maxCachedThreadPool(int maxThread, String name) {
+        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
+            maxThread,
+            maxThread,
+            60L,
+            TimeUnit.SECONDS,
+            new LinkedBlockingQueue<>(),
+            threadFactoryBuilder.build(name + "_%d")
+        );
+
+        threadPoolExecutor.allowCoreThreadTimeOut(true);
+
         return this.wrap(
             name,
-            new ThreadPoolExecutor(
-                minThread,
-                maxThread,
-                60L,
-                TimeUnit.SECONDS,
-                new LinkedBlockingQueue<>(),
-                threadFactoryBuilder.build(name + "_%d")
-            )
+            threadPoolExecutor
         );
     }