enabled lock again

sterlp · sterlp · commit 6aca74e72d95 · 2025-05-31T18:49:30.000+02:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,41 +1,47 @@
 # Changelog
 
-## v1.6.7
+## v1.7.0
+
+-   Added support for a virtual thread factory
+-   Added support to configure a custom thread factory
 
 ### New features:
 
 ### Bug fixes:
-- abandoned triggers will now fire a failed trigger event
-- trigger history scheduler waits now in case of shut down for any history events to get saved
 
-## v1.6.6  - (2025-03-20)
+-   abandoned triggers will now fire a failed trigger event
+-   trigger history scheduler waits now in case of shut down for any history events to get saved
+
+## v1.6.6 - (2025-03-20)
 
 ### New features:
-- first metrics
+
+-   first metrics
 
 ### Bug fixes:
-- fixed history delete start of history
 
+-   fixed history delete start of history
 
-## v1.6.5  - (2025-03-19)
+## v1.6.5 - (2025-03-19)
 
 -   fixed correlation ID size
 -   cancel can now always be triggered from the UI
 -   showing slow used history
 
-## v1.6.4  - (2025-03-18)
+## v1.6.4 - (2025-03-18)
 
 -   in tests the task executor may finish a task before it returns it reference
 
-## v1.6.3  - (2025-03-18)
+## v1.6.3 - (2025-03-18)
 
 -   ensure the remove of a running trigger happens after it is added
 
-## v1.6.2  - (2025-03-18)
+## v1.6.2 - (2025-03-18)
 
 -   removed synchronized from TaskExecutorComponent
 
-## v1.6.1  - (2025-03-18)
+## v1.6.1 - (2025-03-18)
+
 -   simpler RetryStrategy - as function
 -   showing last ping
 -   showing execution time or still running triggers
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerService.java b/core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerService.java
@@ -81,7 +81,7 @@ public void setMaxThreads(int value) {
 
     @PreDestroy
     public void stop() {
-        taskExecutor.close();
+        taskExecutor.shutdown();
         editSchedulerStatus.offline(name);
         runOrQueue.clear();
     }
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/component/TaskExecutorComponent.java b/core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/component/TaskExecutorComponent.java
@@ -1,6 +1,5 @@
 package org.sterl.spring.persistent_tasks.scheduler.component;
 
-import java.io.Closeable;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -35,7 +34,7 @@
  * </p>
  */
 @Slf4j
-public class TaskExecutorComponent implements Closeable {
+public class TaskExecutorComponent {
 
     private final String schedulerName;
     private final TriggerService triggerService;
@@ -49,16 +48,12 @@ public class TaskExecutorComponent implements Closeable {
     private final AtomicBoolean stopped = new AtomicBoolean(true);
     private final Lock lock = new ReentrantLock(true);
 
-    public TaskExecutorComponent(
-            String schedulerName,
-            TriggerService triggerService,
-            SchedulerThreadFactory threadFactory,
-            int maxThreads) {
+    public TaskExecutorComponent(String schedulerName, TriggerService triggerService,
+            SchedulerThreadFactory threadFactory, int maxThreads) {
         super();
         this.schedulerName = schedulerName;
         this.triggerService = triggerService;
         this.maxThreads.set(maxThreads);
-        this.start();
     }
 
     @NonNull
@@ -78,8 +73,8 @@ public Future<TriggerKey> submit(@Nullable TriggerEntity trigger) {
         if (trigger == null) {
             return CompletableFuture.completedFuture(null);
         }
+        lock.lock();
         assertStarted();
-        // lock.lock();
         try {
             var result = executor.submit(() -> runTrigger(trigger));
             runningTasks.put(trigger, result);
@@ -88,7 +83,7 @@ public Future<TriggerKey> submit(@Nullable TriggerEntity trigger) {
             runningTasks.remove(trigger);
             throw new RuntimeException("Failed to run " + trigger.getKey(), e);
         } finally {
-            // lock.unlock();
+            lock.unlock();
         }
 
     }
@@ -104,60 +99,72 @@ private TriggerKey runTrigger(TriggerEntity trigger) {
             triggerService.run(trigger);
             return trigger.getKey();
         } finally {
-            // lock.lock();
+            lock.lock();
             try {
                 if (runningTasks.remove(trigger) == null && runningTasks.size() > 0) {
                     var runningKeys = runningTasks.keySet().stream().map(TriggerEntity::key).toList();
                     log.error("Failed to remove trigger with {} - {}", trigger.key(), runningKeys);
                 }
             } finally {
-                // lock.unlock();
+                lock.unlock();
             }
         }
     }
 
     public void start() {
         if (stopped.compareAndExchange(true, false)) {
-            runningTasks.clear();
-            executor = new ThreadPoolExecutor(
-                    1, this.maxThreads.get(),
-                    60L, TimeUnit.SECONDS,
-                    new LinkedBlockingQueue<Runnable>());
-            log.info("Started {} with {} threads.", schedulerName, maxThreads.get());
-            stopped.set(false);
+            lock.lock();
+            try {
+                runningTasks.clear();
+                executor = new ThreadPoolExecutor(1, this.maxThreads.get(), 60L, TimeUnit.SECONDS,
+                        new LinkedBlockingQueue<Runnable>());
+                log.info("Started {} with {} threads.", schedulerName, maxThreads.get());
+                stopped.set(false);
+            } finally {
+                lock.unlock();
+            }
         }
     }
 
-    @Override
-    public void close() {
+    public void shutdown() {
         stopped.set(true);
         if (executor != null) {
-            executor.shutdown();
-            log.info("Shutdown {} with {} running tasks, waiting for {}.", schedulerName, runningTasks.size(),
-                    maxShutdownWaitTime);
-
-            if (runningTasks.size() > 0) {
-                try {
-                    executor.awaitTermination(maxShutdownWaitTime.getSeconds(), TimeUnit.SECONDS);
-                } catch (InterruptedException e) {
-                    Thread.currentThread().interrupt();
-                    log.warn("Failed to complete runnings tasks.", e.getCause() == null ? e : e.getCause());
-                    shutdownNow();
-                } finally {
-                    executor = null;
-                    runningTasks.clear();
+            lock.lock();
+            try {
+                executor.shutdown();
+                if (runningTasks.size() > 0) {
+                    log.info("Shutdown {} with {} running tasks, waiting for {}.", schedulerName, runningTasks.size(),
+                            maxShutdownWaitTime);
+                    try {
+                        executor.awaitTermination(maxShutdownWaitTime.getSeconds(), TimeUnit.SECONDS);
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                        log.warn("Failed to complete runnings tasks.", e.getCause() == null ? e : e.getCause());
+                        shutdownNow();
+                    }
+                } else {
+                    log.info("Shutdown {} with.", schedulerName);
                 }
+            } finally {
+                executor = null;
+                runningTasks.clear();
+                lock.unlock();
             }
         }
     }
 
     public void shutdownNow() {
         stopped.set(true);
         if (executor != null) {
-            executor.shutdownNow();
-            log.info("Force stop {} with {} running tasks", schedulerName, runningTasks.size());
-            runningTasks.clear();
-            executor = null;
+            lock.lock();
+            try {
+                executor.shutdownNow();
+                log.info("Force stop {} with {} running tasks", schedulerName, runningTasks.size());
+            } finally {
+                runningTasks.clear();
+                executor = null;
+                lock.unlock();
+            }
         }
     }
 
@@ -166,8 +173,7 @@ public int getFreeThreads() {
             return 0;
         }
         if (maxThreads.get() - runningTasks.size() < 0) {
-            log.warn("Already {} running tasks, more than threads {} in pool.",
-                    runningTasks.size(), maxThreads.get());
+            log.warn("Already {} running tasks, more than threads {} in pool.", runningTasks.size(), maxThreads.get());
         }
         return Math.max(maxThreads.get() - runningTasks.size(), 0);
     }
@@ -181,8 +187,7 @@ public Collection<Future<TriggerKey>> getRunningTasks() {
     }
 
     public List<TriggerEntity> getRunningTriggers() {
-        var doneAndNotRemovedFutures = this.runningTasks.entrySet().stream()
-                .filter(e -> e.getValue().isDone())
+        var doneAndNotRemovedFutures = this.runningTasks.entrySet().stream().filter(e -> e.getValue().isDone())
                 .toList();
 
         if (doneAndNotRemovedFutures.size() > 0) {

Original file line number	Diff line number	Diff line change
`@@ -81,7 +81,7 @@ public void setMaxThreads(int value) {`
`81`	`81`
`82`	`82`	`@PreDestroy`
`83`	`83`	`public void stop() {`
`84`		`- taskExecutor.close();`
	`84`	`+ taskExecutor.shutdown();`
`85`	`85`	`editSchedulerStatus.offline(name);`
`86`	`86`	`runOrQueue.clear();`
`87`	`87`	`}`