added a way to cancel and to fail a running trigger

Author: sterlp

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

@@ -44,27 +44,31 @@ public class PersistentTaskService {
     /**
      * Returns the last known {@link TriggerData} to a given key. First running triggers are checked.
      * Maybe out of the history event from a retry execution of the very same id.
-     * 
+     *
      * @param key the {@link TriggerKey} to look for
      * @return the {@link TriggerData} to the {@link TriggerKey}
      */
     public Optional<TriggerData> getLastTriggerData(TriggerKey key) {
         final Optional<TriggerEntity> trigger = triggerService.get(key);
         if (trigger.isEmpty()) {
             var history = historyService.findLastKnownStatus(key);
-            if (history.isPresent()) return Optional.ofNullable(history.get().getData());
+            if (history.isPresent()) {
+                return Optional.ofNullable(history.get().getData());
+            }
             return Optional.empty();
         } else {
             return Optional.ofNullable(trigger.get().getData());
         }
     }
-    
+
     public Optional<TriggerData> getLastDetailData(TriggerKey key) {
         var data = historyService.findAllDetailsForKey(key, Pageable.ofSize(1));
-        if (data.isEmpty()) return Optional.empty();
+        if (data.isEmpty()) {
+            return Optional.empty();
+        }
         return Optional.of(data.getContent().get(0).getData());
     }
-    
+
     @EventListener
     void queue(TriggerTaskCommand<? extends Serializable> event) {
         if (event.size() == 1) {
@@ -76,15 +80,17 @@ void queue(TriggerTaskCommand<? extends Serializable> event) {
 
     /**
      * Queues/updates the given triggers, if the {@link TriggerKey} is already present
-     * 
+     *
      * @param <T> the state type
      * @param triggers the triggers to add
      * @return the {@link TriggerKey}
      */
     @Transactional(timeout = 10)
     @NonNull
     public <T extends Serializable> List<TriggerKey> queue(Collection<AddTriggerRequest<T>> triggers) {
-        if (triggers == null || triggers.isEmpty()) return Collections.emptyList();
+        if (triggers == null || triggers.isEmpty()) {
+            return Collections.emptyList();
+        }
 
         return triggers.stream() //
             .map(t -> triggerService.queue(t)) //
@@ -93,7 +99,7 @@ public <T extends Serializable> List<TriggerKey> queue(Collection<AddTriggerRequ
     }
     /**
      * Queues/updates the given trigger, if the {@link TriggerKey} is already present.
-     * 
+     *
      * @param <T> the state type
      * @param trigger the trigger to add
      * @return the {@link TriggerKey}
@@ -107,7 +113,7 @@ public <T extends Serializable> TriggerKey queue(AddTriggerRequest<T> trigger) {
     /**
      * Runs the given trigger if a free threads are available
      * and the runAt time is not in the future.
-     * @return the reference to the {@link TriggerKey} 
+     * @return the reference to the {@link TriggerKey}
      */
     public <T extends Serializable> TriggerKey runOrQueue(
             AddTriggerRequest<T> triggerRequest) {
@@ -118,10 +124,10 @@ public <T extends Serializable> TriggerKey runOrQueue(
         }
         return triggerRequest.key();
     }
-    
+
     /**
      * Triggers the execution of all pending triggers.
-     * 
+     *
      * @return the reference to the {@link TriggerKey} of the running tasks
      */
     public List<Future<TriggerKey>> executeTriggers() {
@@ -131,7 +137,7 @@ public List<Future<TriggerKey>> executeTriggers() {
         }
         return result;
     }
-    
+
     /**
      * Triggers the execution of all pending triggers and wait for the result.
      */
@@ -151,7 +157,7 @@ public List<TriggerKey> executeTriggersAndWait() {
                     throw cause == null ? e : cause;
                 }
             }
-            
+
             isSomethingRunning = hasRunningTriggers();
             if (isSomethingRunning) {
                 Thread.sleep(Duration.ofMillis(100));
@@ -165,7 +171,7 @@ public List<TriggerKey> executeTriggersAndWait() {
     private boolean hasRunningTriggers() {
         var running = this.schedulers.stream()
                 .map(s -> s.hasRunningTriggers())
-                .filter(r -> r == true)
+                .filter(r -> r)
                 .findAny();
 
         return running.isPresent() && running.get() == true;
@@ -186,7 +192,7 @@ public List<TriggerData> findAllTriggerByCorrelationId(String correlationId) {
         var done = historyService.findTriggerByCorrelationId(correlationId)
             .stream().map(TriggerHistoryLastStateEntity::getData)
             .toList();
-        
+
 
         var result = new ArrayList<TriggerData>(running.size() + done.size());
         result.addAll(done);

core/src/main/java/org/sterl/spring/persistent_tasks/task/exception/CancelTaskException.java

@@ -0,0 +1,15 @@
+package org.sterl.spring.persistent_tasks.task.exception;
+
+/**
+ * Set the task to cancel and finish the execution
+ */
+public class CancelTaskException extends TaskException {
+    private static final long serialVersionUID = 1L;
+    public CancelTaskException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public CancelTaskException(String message) {
+        super(message);
+    }
+}

core/src/main/java/org/sterl/spring/persistent_tasks/task/exception/FailTaskNoRetryException.java

@@ -0,0 +1,15 @@
+package org.sterl.spring.persistent_tasks.task.exception;
+
+/**
+ * Set the task to failed and finish the execution.
+ */
+public class FailTaskNoRetryException extends TaskException {
+    private static final long serialVersionUID = 1L;
+    public FailTaskNoRetryException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public FailTaskNoRetryException(String message) {
+        super(message);
+    }
+}

core/src/main/java/org/sterl/spring/persistent_tasks/task/exception/TaskException.java

@@ -0,0 +1,13 @@
+package org.sterl.spring.persistent_tasks.task.exception;
+
+public abstract class TaskException extends RuntimeException {
+    private static final long serialVersionUID = 1L;
+    
+    public TaskException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public TaskException(String message) {
+        super(message);
+    }
+}

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

@@ -133,11 +133,11 @@ public <T extends Serializable> TriggerEntity queue(AddTriggerRequest<T> tigger)
      * If you changed your mind, cancel the persistentTask
      */
     public Optional<TriggerEntity> cancel(TriggerKey key) {
-        return editTrigger.cancelTask(key);
+        return editTrigger.cancelTask(key, null);
     }
 
     public List<TriggerEntity> cancel(Collection<TriggerKey> key) {
-        return key.stream().map(editTrigger::cancelTask)
+        return key.stream().map(t -> editTrigger.cancelTask(t, null))
            .filter(Optional::isPresent)
            .map(Optional::get)
            .toList();

core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/EditTriggerComponent.java

@@ -6,7 +6,6 @@
 import java.util.List;
 import java.util.Optional;
 
-import org.slf4j.event.Level;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.lang.NonNull;
 import org.springframework.stereotype.Component;
@@ -52,7 +51,7 @@ public Optional<TriggerEntity> completeTaskWithSuccess(TriggerKey key, Serializa
     }
 
     /**
-     * Sets success or error based on the fact if an exception is given or not.
+     * Sets error based on the fact if an exception is given or not.
      */
     public Optional<TriggerEntity> failTrigger(
             TriggerKey key, 
@@ -63,10 +62,6 @@ public Optional<TriggerEntity> failTrigger(
 
 
         result.ifPresent(t -> {
-            log.atLevel(retryAt == null ? Level.ERROR : Level.WARN)
-            .setCause(e)
-            .log("{} failed, retryAt={}",
-                    key, retryAt == null ? "no" : retryAt);
             t.complete(e);
             publisher.publishEvent(new TriggerFailedEvent(t.getId(), t.copyData(), state, e, retryAt));
 
@@ -76,6 +71,7 @@ public Optional<TriggerEntity> failTrigger(
                 t.runAt(retryAt);
             }
         });
+
         if (result.isEmpty()) {
             log.error("Trigger with key={} not found and may be at a wrong state!",
                     key, e);
@@ -84,17 +80,19 @@ public Optional<TriggerEntity> failTrigger(
         return result;
     }
 
-    public Optional<TriggerEntity> cancelTask(TriggerKey id) {
+    public Optional<TriggerEntity> cancelTask(TriggerKey id, Exception e) {
         return triggerRepository //
                 .findByKey(id) //
-                .map(t -> {
-                    t.cancel();
-                    publisher.publishEvent(new TriggerCanceledEvent(
-                            t.getId(), t.copyData(),
-                            stateSerializer.deserializeOrNull(t.getData().getState())));
-                    triggerRepository.delete(t);
-                    return t;
-                });
+                .map(t -> cancelTask(t, e));
+    }
+
+    private TriggerEntity cancelTask(TriggerEntity t, Exception e) {
+        t.cancel(e);
+        publisher.publishEvent(new TriggerCanceledEvent(
+                t.getId(), t.copyData(),
+                stateSerializer.deserializeOrNull(t.getData().getState())));
+        triggerRepository.delete(t);
+        return t;
     }
 
     public <T extends Serializable> TriggerEntity addTrigger(AddTriggerRequest<T> tigger) {

core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/RunTriggerComponent.java

@@ -9,6 +9,8 @@
 import org.springframework.transaction.annotation.Transactional;
 import org.sterl.spring.persistent_tasks.api.task.RunningTriggerContextHolder;
 import org.sterl.spring.persistent_tasks.task.TaskService;
+import org.sterl.spring.persistent_tasks.task.exception.CancelTaskException;
+import org.sterl.spring.persistent_tasks.task.exception.FailTaskNoRetryException;
 import org.sterl.spring.persistent_tasks.trigger.model.RunTaskWithStateCommand;
 import org.sterl.spring.persistent_tasks.trigger.model.TriggerEntity;
 
@@ -66,18 +68,21 @@ private Optional<TriggerEntity> failTaskAndState(RunTaskWithStateCommand runTask
         var task = runTaskWithStateCommand.task();
         Optional<TriggerEntity> result;
 
-        if (task != null 
-                && task.retryStrategy().shouldRetry(trigger.getData().getExecutionCount(), e)) {
+        if (e instanceof CancelTaskException) {
+            log.info("Cancel of a running trigger={} requested", trigger.getKey());
+            result = editTrigger.cancelTask(trigger.getKey(), e);
+        } else if (e instanceof FailTaskNoRetryException) {
+            log.warn("Fail no retry of a running trigger={} requested", trigger.getKey(), e);
+            result = editTrigger.failTrigger(trigger.getKey(), runTaskWithStateCommand.state(), e, null);
+        } else if (task == null 
+                || !task.retryStrategy().shouldRetry(trigger.getData().getExecutionCount(), e)) {
 
+            log.error("Failed trigger={}, no further retries!", trigger.getKey(), e);
+            result = editTrigger.failTrigger(trigger.getKey(), runTaskWithStateCommand.state(), e, null);
+        } else {
             final OffsetDateTime retryAt = task.retryStrategy().retryAt(trigger.getData().getExecutionCount(), e);
-
+            log.warn("Failed trigger={} with retryAt={}", trigger.getKey(), retryAt, e);
             result = editTrigger.failTrigger(trigger.getKey(), runTaskWithStateCommand.state(), e, retryAt);
-
-        } else {
-            log.error("{} failed, no more retries! {}", trigger.getKey(), 
-                    e == null ? "No exception given." : e.getMessage(), e);
-            
-            result = editTrigger.failTrigger(trigger.getKey(), runTaskWithStateCommand.state(), e, null);
         }
         return result;
     }

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

@@ -66,11 +66,18 @@ public TriggerKey getKey() {
         return data.getKey();
     }
 
-    public TriggerEntity cancel() {
+    public TriggerEntity cancel(Exception e) {
         this.data.setEnd(OffsetDateTime.now());
         this.data.setStatus(TriggerStatus.CANCELED);
-        this.data.setExceptionName("PersistentTask canceled");
-        this.data.setRunningDurationInMs(null);
+
+        if (e == null) {
+            this.data.setExceptionName("PersistentTask canceled");
+        } else {
+            data.setExceptionName(e.getClass().getName());
+            data.setLastException(ExceptionUtils.getStackTrace(e));
+        }
+
+        data.updateRunningDuration();
         return this;
     }
 

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

@@ -14,10 +14,10 @@
 import org.sterl.spring.persistent_tasks.AbstractSpringTest;
 import org.sterl.spring.persistent_tasks.api.RetryStrategy;
 import org.sterl.spring.persistent_tasks.api.TaskId.TriggerBuilder;
-import org.sterl.spring.persistent_tasks.api.task.PersistentTask;
-import org.sterl.spring.persistent_tasks.api.task.TransactionalTask;
 import org.sterl.spring.persistent_tasks.api.TriggerKey;
 import org.sterl.spring.persistent_tasks.api.TriggerStatus;
+import org.sterl.spring.persistent_tasks.api.task.PersistentTask;
+import org.sterl.spring.persistent_tasks.api.task.TransactionalTask;
 import org.sterl.spring.sample_app.person.PersonEntity;
 import org.sterl.spring.sample_app.person.PersonRepository;
 import org.sterl.test.Countdown;

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

@@ -22,6 +22,8 @@
 import org.sterl.spring.persistent_tasks.api.TriggerKey;
 import org.sterl.spring.persistent_tasks.api.TriggerStatus;
 import org.sterl.spring.persistent_tasks.history.repository.TriggerHistoryLastStateRepository;
+import org.sterl.spring.persistent_tasks.task.exception.CancelTaskException;
+import org.sterl.spring.persistent_tasks.task.exception.FailTaskNoRetryException;
 import org.sterl.spring.persistent_tasks.task.repository.TaskRepository;
 import org.sterl.spring.persistent_tasks.trigger.component.StateSerializer.DeSerializationFailedException;
 import org.sterl.spring.persistent_tasks.trigger.event.TriggerAddedEvent;
@@ -420,4 +422,46 @@ void testBadStateNoRetry() {
         assertThat(events.stream(TriggerSuccessEvent.class).count()).isZero();
         assertThat(events.stream(TriggerFailedEvent.class).count()).isOne();
     }
+    
+    @Test
+    void tesCancelRunningTrigger() {
+        // GIVEN
+        TaskId<String> taskId = taskService.replace("foo-cancel", c -> {
+            throw new CancelTaskException(c);
+        });
+        var key1 = subject.queue(taskId.newTrigger().build()).getKey();
+
+        // WHEN
+        assertThat(runNextTrigger()).isPresent();
+        assertThat(runNextTrigger()).isEmpty();
+
+        // THEN
+        assertThat(historyService.findLastKnownStatus(key1).get().status()).isEqualTo(TriggerStatus.CANCELED);
+        
+        // AND
+        assertThat(events.stream(TriggerCanceledEvent.class).count()).isOne();
+        assertThat(events.stream(TriggerFailedEvent.class).count()).isZero();
+        assertThat(events.stream(TriggerSuccessEvent.class).count()).isZero();
+    }
+    
+    @Test
+    void tesFailRunningTriggerNoRetry() {
+        // GIVEN
+        TaskId<String> taskId = taskService.replace("foo-fail", c -> {
+            throw new FailTaskNoRetryException(c);
+        });
+        var key1 = subject.queue(taskId.newTrigger().build()).getKey();
+
+        // WHEN
+        assertThat(runNextTrigger()).isPresent();
+        assertThat(runNextTrigger()).isEmpty();
+
+        // THEN
+        assertThat(historyService.findLastKnownStatus(key1).get().status()).isEqualTo(TriggerStatus.FAILED);
+        
+        // AND
+        assertThat(events.stream(TriggerFailedEvent.class).count()).isOne();
+        assertThat(events.stream(TriggerCanceledEvent.class).count()).isZero();
+        assertThat(events.stream(TriggerSuccessEvent.class).count()).isZero();
+    }
 }