adjusted test and fixed code

Author: sterlp

core/src/main/java/org/sterl/spring/persistent_tasks/PersistentTaskService.java

@@ -1,6 +1,7 @@
 package org.sterl.spring.persistent_tasks;
 
 import java.io.Serializable;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -139,6 +140,7 @@ public List<TriggerKey> executeTriggersAndWait() {
         final var result = new ArrayList<TriggerKey>();
 
         List<Future<TriggerKey>> triggers;
+        var isSomethingRunning = false;
         do {
             triggers = executeTriggers();
             for (Future<TriggerKey> future : triggers) {
@@ -149,12 +151,34 @@ public List<TriggerKey> executeTriggersAndWait() {
                     throw cause == null ? e : cause;
                 }
             }
-        } while (!triggers.isEmpty());
+            
+            isSomethingRunning = hasRunningTriggers();
+            if (isSomethingRunning) {
+                Thread.sleep(Duration.ofMillis(100));
+            }
+
+        } while (!triggers.isEmpty() || isSomethingRunning);
 
         return result;
     }
 
-    public List<TriggerData> findTriggerByCorrelationId(String correlationId) {
+    private boolean hasRunningTriggers() {
+        var running = this.schedulers.stream()
+                .map(s -> s.hasRunningTriggers())
+                .filter(r -> r == true)
+                .findAny();
+
+        return running.isPresent() && running.get() == true;
+    }
+
+    /**
+     * Returns all triggers for a correlationId sorted by the creation time.
+     * @param correlationId the id to search for
+     * @return the found {@link TriggerData} sorted by create time ASC
+     */
+    @Transactional(readOnly = true, timeout = 5)
+    public List<TriggerData> findAllTriggerByCorrelationId(String correlationId) {
+
         var running = triggerService.findTriggerByCorrelationId(correlationId)
                 .stream().map(TriggerEntity::getData)
                 .toList();
@@ -163,9 +187,10 @@ public List<TriggerData> findTriggerByCorrelationId(String correlationId) {
             .stream().map(TriggerHistoryLastStateEntity::getData)
             .toList();
 
+
         var result = new ArrayList<TriggerData>(running.size() + done.size());
-        result.addAll(running);
         result.addAll(done);
+        result.addAll(running);
         return result;
     }
 }

core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerService.java

@@ -118,6 +118,10 @@ public List<Future<TriggerKey>> triggerNextTasks(OffsetDateTime timeDue) {
 
             return taskExecutor.submit(result);
         } else {
+            log.debug("No free threads {}/{} right now to run jobs due for: {}",
+                    taskExecutor.getFreeThreads(),
+                    taskExecutor.getMaxThreads(),
+                    timeDue);
             pingRegistry();
             return Collections.emptyList();
         }
@@ -155,6 +159,7 @@ void checkIfTrigerIsRunning(TriggerAddedEvent addedTrigger) {
             log.debug("New triger added for imidiate execution {}", addedTrigger.key());
             taskExecutor.submit(toRun);
         }
+        // TODO implement a cleanup for old pending triggers which may never been triggered!
     }
 
     public SchedulerEntity getStatus() {
@@ -177,4 +182,8 @@ public List<TriggerEntity> rescheduleAbandonedTasks(OffsetDateTime timeout) {
     public List<SchedulerEntity> listAll() {
         return editSchedulerStatus.listAll();
     }
+    
+    public boolean hasRunningTriggers() {
+        return !this.taskExecutor.getRunningTriggers().isEmpty();
+    }
 }

core/src/main/java/org/sterl/spring/persistent_tasks/trigger/TriggerService.java

@@ -136,6 +136,13 @@ public Optional<TriggerEntity> cancel(TriggerKey key) {
         return editTrigger.cancelTask(key);
     }
 
+    public List<TriggerEntity> cancel(Collection<TriggerKey> key) {
+        return key.stream().map(editTrigger::cancelTask)
+           .filter(Optional::isPresent)
+           .map(Optional::get)
+           .toList();
+    }
+
     /**
      * Counts the trigger using the name only from the {@link TaskId}
      *

core/src/main/java/org/sterl/spring/persistent_tasks/trigger/model/TriggerEntity.java

@@ -115,6 +115,10 @@ public TriggerEntity withState(byte[] state) {
         return this;
     }
 
+    public boolean isWaiting() {
+        return data.getStatus() == TriggerStatus.WAITING;
+    }
+    
     public TriggerData copyData() {
         if (data == null) return null;
         return this.data.copy();

core/src/test/java/org/sterl/spring/persistent_tasks/TaskSchedulerServiceTest.java

@@ -100,7 +100,7 @@ void testChainedTasks() {
         asserts.awaitOrdered("234::chainTask1", "chainTask1::234::chainTask1");
         assertThat(correlationId).isEqualTo(correlationFound.get());
         // AND
-        var trigger= persistentTaskService.findTriggerByCorrelationId(correlationId);
+        var trigger= persistentTaskService.findAllTriggerByCorrelationId(correlationId);
         assertThat(trigger).hasSize(2);
         assertThat(trigger.get(0).getCorrelationId()).isEqualTo(correlationId);
         assertThat(trigger.get(1).getCorrelationId()).isEqualTo(correlationId);