TSK-1504: Next scheduled Job now based on due date of job before

history/taskana-simplehistory-provider/src/main/java/pro/taskana/simplehistory/impl/jobs/HistoryCleanupJob.java

@@ -47,12 +47,6 @@ public class HistoryCleanupJob extends AbstractTaskanaJob {
 
   private static final String TASKANA_JOB_HISTORY_BATCH_SIZE = "taskana.jobs.history.batchSize";
 
-  private static final String TASKANA_JOB_HISTORY_CLEANUP_RUN_EVERY =
-      "taskana.jobs.history.cleanup.runEvery";
-
-  private static final String TASKANA_JOB_HISTORY_CLEANUP_FIRST_RUN =
-      "taskana.jobs.history.cleanup.firstRunAt";
-
   private static final String TASKANA_JOB_HISTORY_CLEANUP_MINIMUM_AGE =
       "taskana.jobs.history.cleanup.minimumAge";
 
@@ -61,8 +55,6 @@ public class HistoryCleanupJob extends AbstractTaskanaJob {
   TaskanaHistoryEngineImpl taskanaHistoryEngine =
       TaskanaHistoryEngineImpl.createTaskanaEngine(taskanaEngineImpl);
 
-  private Instant firstRun = Instant.parse("2018-01-01T00:00:00Z");
-  private Duration runEvery = Duration.parse("P1D");
   private Duration minimumAge = Duration.parse("P14D");
   private int batchSize = 100;
 
@@ -268,20 +260,11 @@ public class HistoryCleanupJob extends AbstractTaskanaJob {
     LOGGER.debug("Entry to scheduleNextCleanupJob.");
     ScheduledJob job = new ScheduledJob();
     job.setType(Type.HISTORYCLEANUPJOB);
-    job.setDue(getNextDueForHistoryCleanupJob());
+    job.setDue(getNextDueForCleanupJob());
     taskanaEngineImpl.getJobService().createJob(job);
     LOGGER.debug("Exit from scheduleNextCleanupJob.");
   }
 
-  private Instant getNextDueForHistoryCleanupJob() {
-    Instant nextRunAt = firstRun;
-    while (nextRunAt.isBefore(Instant.now())) {
-      nextRunAt = nextRunAt.plus(runEvery);
-    }
-    LOGGER.info("Scheduling next run of the HistoryCleanupJob for {}", nextRunAt);
-    return nextRunAt;
-  }
-
   private void initJobParameters(Properties props) {
 
     String jobBatchSizeProperty = props.getProperty(TASKANA_JOB_HISTORY_BATCH_SIZE);
@@ -296,33 +279,6 @@ public class HistoryCleanupJob extends AbstractTaskanaJob {
       }
     }
 
-    String historyCleanupJobFirstRunProperty =
-        props.getProperty(TASKANA_JOB_HISTORY_CLEANUP_FIRST_RUN);
-    if (historyCleanupJobFirstRunProperty != null && !historyCleanupJobFirstRunProperty.isEmpty()) {
-      try {
-        firstRun = Instant.parse(historyCleanupJobFirstRunProperty);
-      } catch (Exception e) {
-        LOGGER.warn(
-            "Could not parse historyCleanupJobFirstRunProperty ({}). Using default."
-                + " Exception: {} ",
-            historyCleanupJobFirstRunProperty,
-            e.getMessage());
-      }
-    }
-
-    String historyCleanupJobRunEveryProperty =
-        props.getProperty(TASKANA_JOB_HISTORY_CLEANUP_RUN_EVERY);
-    if (historyCleanupJobRunEveryProperty != null && !historyCleanupJobRunEveryProperty.isEmpty()) {
-      try {
-        runEvery = Duration.parse(historyCleanupJobRunEveryProperty);
-      } catch (Exception e) {
-        LOGGER.warn(
-            "Could not parse historyCleanupJobRunEveryProperty ({}). Using default. Exception: {} ",
-            historyCleanupJobRunEveryProperty,
-            e.getMessage());
-      }
-    }
-
     String historyEventCleanupJobMinimumAgeProperty =
         props.getProperty(TASKANA_JOB_HISTORY_CLEANUP_MINIMUM_AGE);
     if (historyEventCleanupJobMinimumAgeProperty != null
@@ -339,7 +295,6 @@ public class HistoryCleanupJob extends AbstractTaskanaJob {
     }
 
     LOGGER.debug("Configured number of history events per transaction: {}", batchSize);
-    LOGGER.debug("HistoryCleanupJob configuration: first run at {}", firstRun);
     LOGGER.debug("HistoryCleanupJob configuration: runs every {}", runEvery);
     LOGGER.debug(
         "HistoryCleanupJob configuration: minimum age of history events to be cleanup up is {}",

history/taskana-simplehistory-provider/src/test/java/acceptance/jobs/HistoryCleanupJobAccTest.java

@@ -386,7 +386,7 @@ class HistoryCleanupJobAccTest extends AbstractAccTest {
       taskanaEngine.getJobService().createJob(job);
     }
 
-    List<ScheduledJob> jobsToRun = getJobMapper().findJobsToRun();
+    List<ScheduledJob> jobsToRun = getJobMapper().findJobsToRun(Instant.now());
 
     assertThat(jobsToRun).hasSize(30);
 
@@ -397,7 +397,7 @@ class HistoryCleanupJobAccTest extends AbstractAccTest {
 
     HistoryCleanupJob.initializeSchedule(taskanaEngine);
 
-    jobsToRun = getJobMapper().findJobsToRun();
+    jobsToRun = getJobMapper().findJobsToRun(Instant.now());
 
     assertThat(jobsToRun).doesNotContainAnyElementsOf(historyCleanupJobs);
   }

lib/taskana-core/src/main/java/pro/taskana/common/internal/JobMapper.java

@@ -1,5 +1,6 @@
 package pro.taskana.common.internal;
 
+import java.time.Instant;
 import java.util.List;
 import java.util.Map;
 import org.apache.ibatis.annotations.Delete;
@@ -38,7 +39,7 @@ public interface JobMapper {
   @Select(
       "<script> SELECT   JOB_ID, PRIORITY, CREATED, DUE, STATE, LOCKED_BY, LOCK_EXPIRES, TYPE, RETRY_COUNT, ARGUMENTS "
           + "FROM SCHEDULED_JOB "
-          + "WHERE STATE IN ( 'READY') AND (DUE is null OR DUE &lt; CURRENT_TIMESTAMP) AND (LOCK_EXPIRES is null OR LOCK_EXPIRES &lt; CURRENT_TIMESTAMP) AND RETRY_COUNT > 0 "
+          + "WHERE STATE IN ( 'READY') AND (DUE is null OR DUE &lt; #{now}) AND (LOCK_EXPIRES is null OR LOCK_EXPIRES &lt; CURRENT_TIMESTAMP) AND RETRY_COUNT > 0 "
           + "ORDER BY PRIORITY DESC "
           + "<if test=\"_databaseId == 'db2'\">with UR </if> "
           + "</script>")
@@ -59,7 +60,7 @@ public interface JobMapper {
             javaType = Map.class,
             typeHandler = MapTypeHandler.class)
       })
-  List<ScheduledJob> findJobsToRun();
+  List<ScheduledJob> findJobsToRun(Instant now);
 
   @Update(
       value =

lib/taskana-core/src/main/java/pro/taskana/common/internal/JobServiceImpl.java

@@ -73,7 +73,7 @@ public class JobServiceImpl implements JobService {
     List<ScheduledJob> availableJobs;
     try {
       taskanaEngineImpl.openConnection();
-      availableJobs = jobMapper.findJobsToRun();
+      availableJobs = jobMapper.findJobsToRun(Instant.now());
       LOGGER.debug("Found available jobs: {}", availableJobs);
     } finally {
       taskanaEngineImpl.returnConnection();

lib/taskana-core/src/main/java/pro/taskana/common/internal/jobs/AbstractTaskanaJob.java

@@ -1,6 +1,8 @@
 package pro.taskana.common.internal.jobs;
 
 import java.lang.reflect.InvocationTargetException;
+import java.time.Duration;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -13,6 +15,7 @@ import pro.taskana.common.internal.transaction.TaskanaTransactionProvider;
 /** Abstract base for all background jobs of TASKANA. */
 public abstract class AbstractTaskanaJob implements TaskanaJob {
 
+  protected final Duration runEvery;
   protected TaskanaEngineImpl taskanaEngineImpl;
   protected TaskanaTransactionProvider<Object> txProvider;
   protected ScheduledJob scheduledJob;
@@ -24,6 +27,7 @@ public abstract class AbstractTaskanaJob implements TaskanaJob {
     this.taskanaEngineImpl = (TaskanaEngineImpl) taskanaEngine;
     this.txProvider = txProvider;
     this.scheduledJob = job;
+    this.runEvery = taskanaEngineImpl.getConfiguration().getCleanupJobRunEvery();
   }
 
   public static TaskanaJob createFromScheduledJob(
@@ -54,4 +58,17 @@ public abstract class AbstractTaskanaJob implements TaskanaJob {
     }
     return result;
   }
+
+  protected Instant getNextDueForCleanupJob() {
+    Instant nextRun = Instant.now();
+    if (scheduledJob != null && scheduledJob.getDue() != null) {
+      nextRun = scheduledJob.getDue();
+    }
+
+    while (nextRun.isBefore(Instant.now())) {
+      nextRun = nextRun.plus(runEvery);
+    }
+
+    return nextRun;
+  }
 }

lib/taskana-core/src/main/java/pro/taskana/task/internal/jobs/TaskCleanupJob.java

@@ -33,8 +33,6 @@ public class TaskCleanupJob extends AbstractTaskanaJob {
   private static final SortDirection ASCENDING = SortDirection.ASCENDING;
 
   // Parameter
-  private final Instant firstRun;
-  private final Duration runEvery;
   private final Duration minimumAge;
   private final int batchSize;
   private final boolean allCompletedSameParentBusiness;
@@ -44,8 +42,6 @@ public class TaskCleanupJob extends AbstractTaskanaJob {
       TaskanaTransactionProvider<Object> txProvider,
       ScheduledJob scheduledJob) {
     super(taskanaEngine, txProvider, scheduledJob);
-    firstRun = taskanaEngine.getConfiguration().getCleanupJobFirstRun();
-    runEvery = taskanaEngine.getConfiguration().getCleanupJobRunEvery();
     minimumAge = taskanaEngine.getConfiguration().getCleanupJobMinimumAge();
     batchSize = taskanaEngine.getConfiguration().getMaxNumberOfUpdatesPerTransaction();
     allCompletedSameParentBusiness =
@@ -198,17 +194,8 @@ public class TaskCleanupJob extends AbstractTaskanaJob {
     LOGGER.debug("Entry to scheduleNextCleanupJob.");
     ScheduledJob job = new ScheduledJob();
     job.setType(ScheduledJob.Type.TASKCLEANUPJOB);
-    job.setDue(getNextDueForTaskCleanupJob());
+    job.setDue(getNextDueForCleanupJob());
     taskanaEngineImpl.getJobService().createJob(job);
     LOGGER.debug("Exit from scheduleNextCleanupJob.");
   }
-
-  private Instant getNextDueForTaskCleanupJob() {
-    Instant nextRunAt = firstRun;
-    while (nextRunAt.isBefore(Instant.now())) {
-      nextRunAt = nextRunAt.plus(runEvery);
-    }
-    LOGGER.info("Scheduling next run of the TaskCleanupJob for {}", nextRunAt);
-    return nextRunAt;
-  }
 }

lib/taskana-core/src/main/java/pro/taskana/workbasket/internal/jobs/WorkbasketCleanupJob.java

@@ -1,7 +1,5 @@
 package pro.taskana.workbasket.internal.jobs;
 
-import java.time.Duration;
-import java.time.Instant;
 import java.util.List;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -29,8 +27,6 @@ public class WorkbasketCleanupJob extends AbstractTaskanaJob {
   private static final Logger LOGGER = LoggerFactory.getLogger(WorkbasketCleanupJob.class);
 
   // Parameter
-  private final Instant firstRun;
-  private final Duration runEvery;
   private final int batchSize;
 
   public WorkbasketCleanupJob(
@@ -38,8 +34,6 @@ public class WorkbasketCleanupJob extends AbstractTaskanaJob {
       TaskanaTransactionProvider<Object> txProvider,
       ScheduledJob job) {
     super(taskanaEngine, txProvider, job);
-    firstRun = taskanaEngine.getConfiguration().getCleanupJobFirstRun();
-    runEvery = taskanaEngine.getConfiguration().getCleanupJobRunEvery();
     batchSize = taskanaEngine.getConfiguration().getMaxNumberOfUpdatesPerTransaction();
   }
 
@@ -128,17 +122,8 @@ public class WorkbasketCleanupJob extends AbstractTaskanaJob {
     LOGGER.debug("Entry to scheduleNextCleanupJob.");
     ScheduledJob job = new ScheduledJob();
     job.setType(ScheduledJob.Type.WORKBASKETCLEANUPJOB);
-    job.setDue(getNextDueForWorkbasketCleanupJob());
+    job.setDue(getNextDueForCleanupJob());
     taskanaEngineImpl.getJobService().createJob(job);
     LOGGER.debug("Exit from scheduleNextCleanupJob.");
   }
-
-  private Instant getNextDueForWorkbasketCleanupJob() {
-    Instant nextRunAt = firstRun;
-    while (nextRunAt.isBefore(Instant.now())) {
-      nextRunAt = nextRunAt.plus(runEvery);
-    }
-    LOGGER.info("Scheduling next run of the WorkbasketCleanupJob for {}", nextRunAt);
-    return nextRunAt;
-  }
 }

lib/taskana-core/src/test/java/acceptance/jobs/TaskCleanupJobAccTest.java

@@ -4,6 +4,8 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 import acceptance.AbstractAccTest;
 import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Iterator;
@@ -19,6 +21,9 @@ import org.junit.jupiter.api.function.ThrowingConsumer;
 
 import pro.taskana.common.api.ScheduledJob;
 import pro.taskana.common.api.ScheduledJob.Type;
+import pro.taskana.common.internal.JobMapper;
+import pro.taskana.common.internal.JobServiceImpl;
+import pro.taskana.common.internal.jobs.JobRunner;
 import pro.taskana.common.test.security.JaasExtension;
 import pro.taskana.common.test.security.WithAccessId;
 import pro.taskana.task.api.TaskService;
@@ -43,9 +48,9 @@ class TaskCleanupJobAccTest extends AbstractAccTest {
   @WithAccessId(user = "admin")
   @Test
   void should_CleanCompletedTasksUntilDate() throws Exception {
-    String id = createAndInsertTask(null);
+    String taskId = createAndInsertTask(null);
-    taskService.claim(id);
+    taskService.claim(taskId);
-    taskService.completeTask(id);
+    taskService.completeTask(taskId);
 
     long totalTasksCount = taskService.createTaskQuery().count();
     assertThat(totalTasksCount).isEqualTo(88);
@@ -88,14 +93,14 @@ class TaskCleanupJobAccTest extends AbstractAccTest {
   @WithAccessId(user = "admin")
   @Test
   void shouldNotCleanCompleteTasksAfterDefinedDay() throws Exception {
-    String id = createAndInsertTask(null);
+    String taskId = createAndInsertTask(null);
-    taskService.claim(id);
+    taskService.claim(taskId);
-    taskService.completeTask(id);
+    taskService.completeTask(taskId);
 
     TaskCleanupJob job = new TaskCleanupJob(taskanaEngine, null, null);
     job.run();
 
-    Task completedCreatedTask = taskService.getTask(id);
+    Task completedCreatedTask = taskService.getTask(taskId);
     assertThat(completedCreatedTask).isNotNull();
   }
 
@@ -113,7 +118,7 @@ class TaskCleanupJobAccTest extends AbstractAccTest {
       taskanaEngine.getJobService().createJob(job);
     }
 
-    List<ScheduledJob> jobsToRun = getJobMapper().findJobsToRun();
+    List<ScheduledJob> jobsToRun = getJobMapper().findJobsToRun(Instant.now());
 
     assertThat(jobsToRun).hasSize(30);
 
@@ -124,7 +129,7 @@ class TaskCleanupJobAccTest extends AbstractAccTest {
 
     TaskCleanupJob.initializeSchedule(taskanaEngine);
 
-    jobsToRun = getJobMapper().findJobsToRun();
+    jobsToRun = getJobMapper().findJobsToRun(Instant.now());
 
     assertThat(jobsToRun).doesNotContainAnyElementsOf(taskCleanupJobs);
   }
@@ -157,6 +162,35 @@ class TaskCleanupJobAccTest extends AbstractAccTest {
     return DynamicTest.stream(iterator, c -> "for parentBusinessProcessId = '" + c + "'", test);
   }
 
+  @WithAccessId(user = "admin")
+  @Test
+  void should_SetNextScheduledJobBasedOnDueDateOfPredecessor_When_RunningTaskCleanupJob()
+      throws Exception {
+    JobMapper jobMapper = getJobMapper();
+    List<ScheduledJob> jobsToRun = jobMapper.findJobsToRun(Instant.now());
+    assertThat(jobsToRun).isEmpty();
+
+    Instant firstDue = Instant.now().truncatedTo(ChronoUnit.MILLIS);
+    ScheduledJob scheduledJob = new ScheduledJob();
+    scheduledJob.setType(ScheduledJob.Type.TASKCLEANUPJOB);
+    scheduledJob.setDue(firstDue);
+
+    JobServiceImpl jobService = (JobServiceImpl) taskanaEngine.getJobService();
+    jobService.createJob(scheduledJob);
+    jobsToRun = jobMapper.findJobsToRun(Instant.now());
+
+    assertThat(jobsToRun).hasSize(1);
+    assertThat(jobsToRun.get(0).getDue()).isEqualTo(firstDue);
+
+    JobRunner runner = new JobRunner(taskanaEngine);
+    runner.runJobs();
+    Duration runEvery = taskanaEngineConfiguration.getCleanupJobRunEvery();
+    jobsToRun = jobMapper.findJobsToRun(Instant.now().plus(runEvery));
+
+    assertThat(jobsToRun).hasSize(1);
+    assertThat(jobsToRun).extracting(ScheduledJob::getDue).containsExactly(firstDue.plus(runEvery));
+  }
+
   private String createAndInsertTask(String parentBusinessProcessId) throws Exception {
     Task newTask = taskService.newTask("user-1-1", "DOMAIN_A");
     newTask.setClassificationKey("T2100");

lib/taskana-core/src/test/java/acceptance/jobs/WorkbasketCleanupJobAccTest.java

@@ -3,6 +3,7 @@ package acceptance.jobs;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import acceptance.AbstractAccTest;
+import java.time.Instant;
 import java.util.List;
 import java.util.stream.Collectors;
 import org.junit.jupiter.api.AfterEach;
@@ -102,7 +103,7 @@ class WorkbasketCleanupJobAccTest extends AbstractAccTest {
       taskanaEngine.getJobService().createJob(job);
     }
 
-    List<ScheduledJob> jobsToRun = getJobMapper().findJobsToRun();
+    List<ScheduledJob> jobsToRun = getJobMapper().findJobsToRun(Instant.now());
 
     assertThat(jobsToRun).hasSize(30);
 
@@ -113,7 +114,7 @@ class WorkbasketCleanupJobAccTest extends AbstractAccTest {
 
     WorkbasketCleanupJob.initializeSchedule(taskanaEngine);
 
-    jobsToRun = getJobMapper().findJobsToRun();
+    jobsToRun = getJobMapper().findJobsToRun(Instant.now());
 
     assertThat(jobsToRun).doesNotContainAnyElementsOf(workbasketCleanupJobs);
   }