added timeout

Author: sterlp

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

@@ -142,8 +142,9 @@ public List<Future<TriggerKey>> executeTriggers() {
      * Triggers the execution of all pending triggers and wait for the result.
      */
     @SneakyThrows
-    public List<TriggerKey> executeTriggersAndWait() {
+    public List<TriggerKey> executeTriggersAndWait(Duration maxWaitTime) {
         final var result = new ArrayList<TriggerKey>();
+        final var timeOut = System.currentTimeMillis() + maxWaitTime.toMillis();
 
         List<Future<TriggerKey>> triggers;
         var isSomethingRunning = false;
@@ -162,6 +163,10 @@ public List<TriggerKey> executeTriggersAndWait() {
             if (isSomethingRunning) {
                 Thread.sleep(Duration.ofMillis(100));
             }
+            
+            if (System.currentTimeMillis() > timeOut) {
+                throw new RuntimeException("Timeout waiting for triggers after " + maxWaitTime);
+            }
 
         } while (!triggers.isEmpty() || isSomethingRunning);
 

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

@@ -2,6 +2,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.UUID;
 import java.util.concurrent.Callable;
@@ -40,8 +41,8 @@ public RetryStrategy retryStrategy() {
         var runTrigger = triggerService.queue(task.newTrigger().state("hallo").build());
 
         // WHEN
-        persistentTaskService.executeTriggersAndWait();
-        persistentTaskService.executeTriggersAndWait();
+        persistentTaskService.executeTriggersAndWait(Duration.ofSeconds(2));
+        persistentTaskService.executeTriggersAndWait(Duration.ofSeconds(1));
 
         // THEN
         assertThat(asserts.getCount("hallo")).isEqualTo(4);
@@ -67,7 +68,7 @@ void testLockTriggerInSchedulers() throws Exception {
             lockInvocations.add(() -> runNextTrigger());
         }
 
-        persistentTaskService.executeTriggersAndWait();
+        persistentTaskService.executeTriggersAndWait(Duration.ofSeconds(2));
 
         // THEN
         for (int i = 1; i <= 100; ++i) {

core/src/test/java/org/sterl/spring/persistent_tasks/history/HistoryServiceTest.java

@@ -2,11 +2,11 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.time.Duration;
 import java.time.OffsetDateTime;
 import java.util.Optional;
 import java.util.concurrent.TimeoutException;
 
-import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.domain.PageRequest;
@@ -35,7 +35,7 @@ void testReQueueTrigger() {
         // THEN
         assertThat(t).isPresent();
         // AND
-        persistentTaskService.executeTriggersAndWait();
+        persistentTaskService.executeTriggersAndWait(Duration.ofSeconds(2));
         asserts.assertValue(Task3.NAME + "::Hallo");
         // AND
         assertThat(subject.countTriggers(trigger.getKey())).isEqualTo(2);
@@ -46,7 +46,7 @@ void testTriggerHistory() throws TimeoutException, InterruptedException {
         // GIVEN
         final var trigger = Task3.ID.newUniqueTrigger("Hallo");
         triggerService.queue(trigger);
-        persistentTaskService.executeTriggersAndWait();
+        persistentTaskService.executeTriggersAndWait(Duration.ofSeconds(2));
         // WHEN
         var triggers = subject.findAllDetailsForKey(trigger.key(), PageRequest.of(0, 100)).getContent();
 

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

@@ -115,7 +115,7 @@ void testQueuedInFuture() throws TimeoutException, InterruptedException {
         subject.runOrQueue(triggerRequest);
 
         // WHEN
-        persistentTaskService.executeTriggersAndWait();
+        persistentTaskService.executeTriggersAndWait(Duration.ofSeconds(2));
         awaitRunningTasks();
 
         // THEN
@@ -132,7 +132,7 @@ void runSimpleTaskMultipleTimesTest() throws Exception {
         }
 
         // WHEN
-        persistentTaskService.executeTriggersAndWait();
+        persistentTaskService.executeTriggersAndWait(Duration.ofSeconds(2));
 
         // THEN
         for (int i = 1; i < 21; ++i) {

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

@@ -2,6 +2,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.time.Duration;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -222,7 +223,7 @@ void testRollbackAndRetry() throws Exception {
 
         // WHEN
         sendError.set(false);
-        var executed = persistentTaskService.executeTriggersAndWait();
+        var executed = persistentTaskService.executeTriggersAndWait(Duration.ofSeconds(2));
 
         // THEN
         assertThat(executed).hasSize(1);

core/src/test/java/org/sterl/spring/persistent_tasks/trigger/TriggerServiceTest.java

@@ -338,7 +338,7 @@ void testLockTrigger() throws Exception {
             }
 
             executor.invokeAll(lockInvocations);
-            persistentTaskService.executeTriggersAndWait();
+            persistentTaskService.executeTriggersAndWait(Duration.ofSeconds(2));
 
             // THEN
             for (int i = 1; i <= 100; ++i) {
@@ -359,7 +359,7 @@ void testQueuedInFuture() {
         subject.queue(triggerRequest);
 
         // WHEN
-        persistentTaskService.executeTriggersAndWait();
+        persistentTaskService.executeTriggersAndWait(Duration.ofSeconds(2));
 
         // THEN
         asserts.assertMissing(Task3.NAME + "::Hallo");

core/src/test/java/org/sterl/spring/persistent_tasks/trigger/api/TriggerResourceTest.java

@@ -174,7 +174,8 @@ void testUpdateRunAt() {
                 HttpMethod.POST, new HttpEntity<>(OffsetDateTime.now()), Trigger.class);
 
         // THEN
-        persistentTaskService.executeTriggersAndWait();
+        persistentTaskService.executeTriggersAndWait(Duration.ofSeconds(2));
+        
         asserts.assertValue(Task3.NAME + "::Hallo");
         asserts.assertMissing(Task3.NAME + "::Hallo2");
         assertThat(triggerService.countTriggers(TriggerStatus.WAITING)).isOne();