1.6.2 release

Author: sterlp

CHANGELOG.md

@@ -1,6 +1,10 @@
 # Changelog
 
-## v1.6.1
+## v1.6.2  - (2025-03-18)
+
+-   removed synchronized from TaskExecutorComponent
+
+## v1.6.1  - (2025-03-18)
 -   simpler RetryStrategy - as function
 -   showing last ping
 -   showing execution time or still running triggers

core/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.sterl.spring</groupId>
         <artifactId>spring-persistent-tasks-root</artifactId>
-        <version>1.6.2-SNAPSHOT</version>
+        <version>1.6.2</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/component/TaskExecutorComponent.java

@@ -74,27 +74,30 @@ public Future<TriggerKey> submit(@Nullable TriggerEntity trigger) {
             throw new IllegalStateException("Executor of " + schedulerName + " is already stopped");
         }
 
-        Future<TriggerKey> result;
-        synchronized (runningTasks) {
-            result = executor.submit(() -> runTrigger(trigger));
-            runningTasks.put(trigger, result);
+        try {
+            runningTasks.put(trigger, executor.submit(() -> runTrigger(trigger)));
+            return runningTasks.get(trigger);
+        } catch (Exception e) {
+            runningTasks.remove(trigger);
+            throw new RuntimeException("Failed to run " + trigger.getKey(), e);
         }
 
-        return result;
     }
 
     private TriggerKey runTrigger(TriggerEntity trigger) {
         try {
             triggerService.run(trigger);
             return trigger.getKey();
         } finally {
-            runningTasks.remove(trigger);
+            if (runningTasks.remove(trigger) == null) {
+                log.error("Failed to remove trigger with {}", trigger.key());
+            }
         }
     }
 
     public void start() {
-        if (stopped.compareAndExchange(true, false)) {
-            synchronized (runningTasks) {
+        synchronized (runningTasks) {
+            if (stopped.compareAndExchange(true, false)) {
                 runningTasks.clear();
                 executor = Executors.newFixedThreadPool(maxThreads.get());
                 log.info("Started {} with {} threads.", schedulerName, maxThreads.get());
@@ -105,43 +108,41 @@ public void start() {
     @Override
     public void close() {
         if (stopped.compareAndExchange(false, true)) {
+            ExecutorService executorRef;
             synchronized (runningTasks) {
-                doShutdown();
+                executorRef = executor;
+                executor = null;
             }
-        }
-    }
-
-    private void doShutdown() {
-        if (executor != null) {
-            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) {
-                    log.warn("Failed to complete runnings tasks.", e.getCause());
-                    shutdownNow();
-                } finally {
-                    executor = null;
-                    runningTasks.clear();
+            
+            if (executorRef != null) {
+                executorRef.shutdown();
+                if (runningTasks.size() > 0) {
+                    log.info("Shutdown {} with {} running tasks, waiting for {}.", schedulerName, runningTasks.size(),
+                            maxShutdownWaitTime);
+
+                    try {
+                        executorRef.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 {
+                        executorRef = null;
+                        runningTasks.clear();
+                    }
                 }
-            } else {
-                executor = null;
             }
         }
     }
 
     public void shutdownNow() {
-        if (stopped.compareAndExchange(false, true)) {
-            synchronized (runningTasks) {
-                if (executor != null) {
-                    executor.shutdownNow();
-                    log.info("Force stop {} with {} running tasks", schedulerName, runningTasks.size());
-                    runningTasks.clear();
-                    executor = null;
-                }
+        synchronized (runningTasks) {
+            if (executor != null) {
+                executor.shutdownNow();
+                log.info("Force stop {} with {} running tasks", schedulerName, runningTasks.size());
+                runningTasks.clear();
+                executor = null;
+                stopped.set(true);
             }
         }
     }
@@ -178,6 +179,6 @@ public int getMaxThreads() {
     }
 
     public boolean isRunning(TriggerEntity trigger) {
-        return runningTasks.contains(trigger);
+        return runningTasks.containsKey(trigger);
     }
 }

core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerServiceTest.java

@@ -4,6 +4,8 @@
 
 import java.time.Duration;
 import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -42,6 +44,31 @@ void schedulerShouldBeOnlineTest() {
         assertThat(status.getLastPing()).isBeforeOrEqualTo(time);
     }
 
+    @Test
+    void testFastTaskFreeThreadChound() throws Exception {
+        // GIVEN
+        var t = taskService.<String>replace("fastTask", s -> {});
+        for (int i = 0; i < 50; ++i) triggerService.queue(t.newUniqueTrigger(null));
+        // WHEN
+        List<Future<TriggerKey>> trigger;
+        while ( (trigger = subject.triggerNextTasks()).size() > 0) {
+            trigger.forEach(t1 -> {
+                try {
+                    t1.get();
+                } catch (Exception e) {
+                    throw new RuntimeException(e.getCause());
+                }
+            });
+        }
+        // THEN
+        assertThat(subject.getRunning().size()).isZero();
+        assertThat(subject.hasRunningTriggers()).isFalse();
+        // AND
+        subject.triggerNextTasks();
+        final SchedulerEntity scheduler = subject.getScheduler();
+        assertThat(scheduler.getRunningTasks()).isEqualTo(0);
+    }
+    
     @Test
     void testWillTriggerOnlyFreeThreadSize() throws Exception {
         // GIVEN

db/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.sterl.spring</groupId>
         <artifactId>spring-persistent-tasks-root</artifactId>
-        <version>1.6.2-SNAPSHOT</version>
+        <version>1.6.2</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

pom.xml

@@ -11,7 +11,7 @@
 
     <groupId>org.sterl.spring</groupId>
     <artifactId>spring-persistent-tasks-root</artifactId>
-    <version>1.6.2-SNAPSHOT</version>
+    <version>1.6.2</version>
     <packaging>pom</packaging>
 
     <inceptionYear>2024</inceptionYear>

test/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.sterl.spring</groupId>
         <artifactId>spring-persistent-tasks-root</artifactId>
-        <version>1.6.2-SNAPSHOT</version>
+        <version>1.6.2</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

ui/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.sterl.spring</groupId>
         <artifactId>spring-persistent-tasks-root</artifactId>
-        <version>1.6.2-SNAPSHOT</version>
+        <version>1.6.2</version>
         <relativePath>../pom.xml</relativePath>
     </parent>