TSK-1989: integrate execution of jobs in taskana core (#2087)

Co-authored-by: Mustapha Zorgati <15628173+mustaphazorgati@users.noreply.github.com>

common/taskana-common/src/main/java/pro/taskana/common/internal/configuration/DbSchemaCreator.java

@@ -52,9 +52,10 @@ public class DbSchemaCreator {
   /**
    * Run all db scripts.
    *
+   * @return true when schema was created, false when no schema created because already existing
    * @throws SQLException will be thrown if there will be some incorrect SQL statements invoked.
    */
-  public void run() throws SQLException {
+  public boolean run() throws SQLException {
     try (Connection connection = dataSource.getConnection()) {
       if (LOGGER.isDebugEnabled()) {
         LOGGER.debug(
@@ -72,6 +73,7 @@ public class DbSchemaCreator {
         BufferedReader reader =
             new BufferedReader(new InputStreamReader(resourceAsStream, StandardCharsets.UTF_8));
         runner.runScript(getSqlSchemaNameParsed(reader));
+        return true;
       }
     }
     if (LOGGER.isDebugEnabled()) {
@@ -80,6 +82,7 @@ public class DbSchemaCreator {
     if (!errorWriter.toString().trim().isEmpty()) {
       LOGGER.error(errorWriter.toString());
     }
+    return false;
   }
 
   public boolean isValidSchemaVersion(String expectedMinVersion) {

common/taskana-common/src/main/java/pro/taskana/common/internal/configuration/TaskanaConfigurationInitializer.java

@@ -45,12 +45,14 @@ public class TaskanaConfigurationInitializer {
 
   static {
     PROPERTY_INITIALIZER_BY_CLASS.put(Integer.class, new IntegerPropertyParser());
+    PROPERTY_INITIALIZER_BY_CLASS.put(Long.class, new LongPropertyParser());
     PROPERTY_INITIALIZER_BY_CLASS.put(Boolean.class, new BooleanPropertyParser());
     PROPERTY_INITIALIZER_BY_CLASS.put(String.class, new StringPropertyParser());
     PROPERTY_INITIALIZER_BY_CLASS.put(Duration.class, new DurationPropertyParser());
     PROPERTY_INITIALIZER_BY_CLASS.put(Instant.class, new InstantPropertyParser());
     PROPERTY_INITIALIZER_BY_CLASS.put(List.class, new ListPropertyParser());
     PROPERTY_INITIALIZER_BY_CLASS.put(Map.class, new MapPropertyParser());
+    PROPERTY_INITIALIZER_BY_CLASS.put(Enum.class, new EnumPropertyParser());
   }
 
   private TaskanaConfigurationInitializer() {
@@ -65,12 +67,15 @@ public class TaskanaConfigurationInitializer {
           .ifPresent(
               taskanaProperty -> {
                 Class<?> type = ReflectionUtil.wrap(field.getType());
-                PropertyParser<?> propertyParser =
+                PropertyParser<?> propertyParser;
-                    Optional.ofNullable(PROPERTY_INITIALIZER_BY_CLASS.get(type))
+                if (type.isEnum()) {
-                        .orElseThrow(
+                  propertyParser = PROPERTY_INITIALIZER_BY_CLASS.get(Enum.class);
-                            () ->
+                } else {
-                                new SystemException(
+                  propertyParser = PROPERTY_INITIALIZER_BY_CLASS.get(type);
-                                    String.format("Unknown configuration type '%s'", type)));
+                }
+                if (propertyParser == null) {
+                  throw new SystemException(String.format("Unknown configuration type '%s'", type));
+                }
                 propertyParser
                     .initialize(props, separator, field, taskanaProperty)
                     .ifPresent(value -> setFieldValue(instance, field, value));
@@ -275,7 +280,7 @@ public class TaskanaConfigurationInitializer {
         String separator,
         Field field,
         TaskanaProperty taskanaProperty) {
-      if (!List.class.isAssignableFrom(field.getType())) {
+      if (!List.class.isAssignableFrom(ReflectionUtil.wrap(field.getType()))) {
         throw new SystemException(
             String.format(
                 "Cannot initialize field '%s' because field type '%s' is not a List",
@@ -332,6 +337,12 @@ public class TaskanaConfigurationInitializer {
         String separator,
         Field field,
         TaskanaProperty taskanaProperty) {
+      if (!Instant.class.isAssignableFrom(ReflectionUtil.wrap(field.getType()))) {
+        throw new SystemException(
+            String.format(
+                "Cannot initialize field '%s' because field type '%s' is not an Instant",
+                field, field.getType()));
+      }
       return parseProperty(properties, taskanaProperty.value(), Instant::parse);
     }
   }
@@ -343,6 +354,12 @@ public class TaskanaConfigurationInitializer {
         String separator,
         Field field,
         TaskanaProperty taskanaProperty) {
+      if (!Duration.class.isAssignableFrom(ReflectionUtil.wrap(field.getType()))) {
+        throw new SystemException(
+            String.format(
+                "Cannot initialize field '%s' because field type '%s' is not a Duration",
+                field, field.getType()));
+      }
       return parseProperty(properties, taskanaProperty.value(), Duration::parse);
     }
   }
@@ -354,6 +371,12 @@ public class TaskanaConfigurationInitializer {
         String separator,
         Field field,
         TaskanaProperty taskanaProperty) {
+      if (!String.class.isAssignableFrom(ReflectionUtil.wrap(field.getType()))) {
+        throw new SystemException(
+            String.format(
+                "Cannot initialize field '%s' because field type '%s' is not a String",
+                field, field.getType()));
+      }
       return parseProperty(properties, taskanaProperty.value(), String::new);
     }
   }
@@ -365,10 +388,33 @@ public class TaskanaConfigurationInitializer {
         String separator,
         Field field,
         TaskanaProperty taskanaProperty) {
+      if (!Integer.class.isAssignableFrom(ReflectionUtil.wrap(field.getType()))) {
+        throw new SystemException(
+            String.format(
+                "Cannot initialize field '%s' because field type '%s' is not an Integer",
+                field, field.getType()));
+      }
       return parseProperty(properties, taskanaProperty.value(), Integer::parseInt);
     }
   }
 
+  static class LongPropertyParser implements PropertyParser<Long> {
+    @Override
+    public Optional<Long> initialize(
+        Map<String, String> properties,
+        String separator,
+        Field field,
+        TaskanaProperty taskanaProperty) {
+      if (!Long.class.isAssignableFrom(ReflectionUtil.wrap(field.getType()))) {
+        throw new SystemException(
+            String.format(
+                "Cannot initialize field '%s' because field type '%s' is not a Long",
+                field, field.getType()));
+      }
+      return parseProperty(properties, taskanaProperty.value(), Long::parseLong);
+    }
+  }
+
   static class BooleanPropertyParser implements PropertyParser<Boolean> {
     @Override
     public Optional<Boolean> initialize(
@@ -376,7 +422,48 @@ public class TaskanaConfigurationInitializer {
         String separator,
         Field field,
         TaskanaProperty taskanaProperty) {
+      if (!Boolean.class.isAssignableFrom(ReflectionUtil.wrap(field.getType()))) {
+        throw new SystemException(
+            String.format(
+                "Cannot initialize field '%s' because field type '%s' is not a Boolean",
+                field, field.getType()));
+      }
       return parseProperty(properties, taskanaProperty.value(), Boolean::parseBoolean);
     }
   }
+
+  static class EnumPropertyParser implements PropertyParser<Enum<?>> {
+    @Override
+    public Optional<Enum<?>> initialize(
+        Map<String, String> properties,
+        String separator,
+        Field field,
+        TaskanaProperty taskanaProperty) {
+      if (!field.getType().isEnum()) {
+        throw new SystemException(
+            String.format(
+                "Cannot initialize field '%s' because field type '%s' is not an Enum",
+                field, field.getType()));
+      }
+      return parseProperty(
+          properties,
+          taskanaProperty.value(),
+          string -> {
+            Map<String, ?> enumConstantsByLowerCasedName =
+                Arrays.stream(field.getType().getEnumConstants())
+                    .collect(
+                        Collectors.toMap(e -> e.toString().toLowerCase(), Function.identity()));
+            Object o = enumConstantsByLowerCasedName.get(string.toLowerCase());
+            if (o == null) {
+              throw new SystemException(
+                  String.format(
+                      "Invalid property value '%s': Valid values are '%s' or '%s",
+                      string,
+                      enumConstantsByLowerCasedName.keySet(),
+                      Arrays.toString(field.getType().getEnumConstants())));
+            }
+            return (Enum<?>) o;
+          });
+    }
+  }
 }

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

@@ -21,7 +21,6 @@ import pro.taskana.common.api.TimeInterval;
 import pro.taskana.common.api.exceptions.InvalidArgumentException;
 import pro.taskana.common.api.exceptions.MismatchedRoleException;
 import pro.taskana.common.api.exceptions.SystemException;
-import pro.taskana.common.internal.JobServiceImpl;
 import pro.taskana.common.internal.jobs.AbstractTaskanaJob;
 import pro.taskana.common.internal.transaction.TaskanaTransactionProvider;
 import pro.taskana.common.internal.util.CollectionUtil;
@@ -127,19 +126,6 @@ public class HistoryCleanupJob extends AbstractTaskanaJob {
     }
   }
 
-  /**
-   * Initializes the HistoryCleanupJob schedule. <br>
-   * All scheduled cleanup jobs are cancelled/deleted and a new one is scheduled.
-   *
-   * @param taskanaEngine the TASKANA engine.
-   */
-  public static void initializeSchedule(TaskanaEngine taskanaEngine) {
-    JobServiceImpl jobService = (JobServiceImpl) taskanaEngine.getJobService();
-    HistoryCleanupJob job = new HistoryCleanupJob(taskanaEngine, null, null);
-    jobService.deleteJobs(job.getType());
-    job.scheduleNextJob();
-  }
-
   @Override
   protected String getType() {
     return HistoryCleanupJob.class.getName();

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

@@ -21,6 +21,7 @@ import org.junit.jupiter.api.function.ThrowingConsumer;
 import pro.taskana.TaskanaConfiguration;
 import pro.taskana.classification.internal.jobs.ClassificationChangedJob;
 import pro.taskana.common.api.ScheduledJob;
+import pro.taskana.common.internal.jobs.AbstractTaskanaJob;
 import pro.taskana.common.internal.util.Pair;
 import pro.taskana.common.test.config.DataSourceGenerator;
 import pro.taskana.common.test.security.JaasExtension;
@@ -453,7 +454,7 @@ class HistoryCleanupJobAccTest extends AbstractAccTest {
                 scheduledJob -> scheduledJob.getType().equals(HistoryCleanupJob.class.getName()))
             .collect(Collectors.toList());
 
-    HistoryCleanupJob.initializeSchedule(taskanaEngine);
+    AbstractTaskanaJob.initializeSchedule(taskanaEngine, HistoryCleanupJob.class);
 
     jobsToRun = getJobMapper().findJobsToRun(Instant.now());
 

history/taskana-simplehistory-provider/src/test/resources/taskana.properties

@@ -25,4 +25,19 @@ taskana.workingtime.schedule.TUESDAY=00:00-00:00
 taskana.workingtime.schedule.WEDNESDAY=00:00-00:00
 taskana.workingtime.schedule.THURSDAY=00:00-00:00
 taskana.workingtime.schedule.FRIDAY=00:00-00:00
+# enable or disable the jobscheduler at all
+# set it to false and no jobs are running
+taskana.jobscheduler.enabled=false
+# wait time before the first job run in millis
+taskana.jobscheduler.initialstartdelay=100
+# sleeping time befor the next job runs
+taskana.jobscheduler.period=12
+# timeunit for the sleeping period
+# Possible values: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+taskana.jobscheduler.periodtimeunit=HOURS
+taskana.jobscheduler.enableTaskCleanupJob=true
+taskana.jobscheduler.enableTaskUpdatePriorityJob=true
+taskana.jobscheduler.enableWorkbasketCleanupJob=true
+taskana.jobscheduler.enableUserInfoRefreshJob=false
+taskana.jobscheduler.enableHistorieCleanupJob=true
 

lib/taskana-cdi/src/test/resources/taskana.properties

@@ -1,3 +1,18 @@
 datasource.jndi=java:jboss/datasources/TestDS
 taskana.domains=CDIDOMAIN
 taskana.classification.types=T1
+# enable or disable the jobscheduler at all
+# set it to false and no jobs are running
+taskana.jobscheduler.enabled=true
+# wait time before the first job run in millis
+taskana.jobscheduler.initialstartdelay=100
+# sleeping time befor the next job runs
+taskana.jobscheduler.period=1
+# timeunit for the sleeping period
+# Possible values: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+taskana.jobscheduler.periodtimeunit=HOURS
+taskana.jobscheduler.enableTaskCleanupJob=true
+taskana.jobscheduler.enableTaskUpdatePriorityJob=true
+taskana.jobscheduler.enableWorkbasketCleanupJob=true
+taskana.jobscheduler.enableUserInfoRefreshJob=false
+taskana.jobscheduler.enableHistorieCleanupJob=false

lib/taskana-core-test/src/test/java/acceptance/ArchitectureTest.java

@@ -68,6 +68,7 @@ import pro.taskana.common.api.exceptions.TaskanaRuntimeException;
 import pro.taskana.common.internal.InternalTaskanaEngine;
 import pro.taskana.common.internal.Interval;
 import pro.taskana.common.internal.TaskanaEngineImpl;
+import pro.taskana.common.internal.jobs.JobScheduler;
 import pro.taskana.common.internal.logging.LoggingAspect;
 import pro.taskana.common.internal.util.MapCreator;
 import pro.taskana.common.internal.workingtime.HolidaySchedule;
@@ -329,6 +330,8 @@ class ArchitectureTest {
                 .areNotAssignableTo(TaskanaEngine.class)
                 .and()
                 .areNotAssignableTo(InternalTaskanaEngine.class)
+                .and()
+                .areNotAssignableTo(JobScheduler.class)
                 .should()
                 .onlyDependOnClassesThat()
                 .resideOutsideOfPackage(rootPackage + "..")

lib/taskana-core-test/src/test/java/acceptance/TaskanaConfigurationTest.java

@@ -4,13 +4,16 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
 import java.lang.reflect.Field;
+import java.time.DayOfWeek;
 import java.time.Duration;
 import java.time.Instant;
+import java.time.LocalTime;
 import java.time.temporal.ChronoUnit;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Stream;
 import org.junit.jupiter.api.DynamicTest;
 import org.junit.jupiter.api.Test;
@@ -20,6 +23,7 @@ import org.junit.jupiter.api.function.ThrowingConsumer;
 import pro.taskana.TaskanaConfiguration;
 import pro.taskana.TaskanaConfiguration.Builder;
 import pro.taskana.common.api.CustomHoliday;
+import pro.taskana.common.api.LocalTimeInterval;
 import pro.taskana.common.api.TaskanaRole;
 import pro.taskana.common.internal.util.ReflectionUtil;
 import pro.taskana.testapi.extensions.TestContainerExtension;
@@ -101,8 +105,14 @@ class TaskanaConfigurationTest {
     Duration expectedUserRefreshJobRunEvery = Duration.ofDays(5);
     List<WorkbasketPermission> expectedMinimalPermissionsToAssignDomains =
         List.of(WorkbasketPermission.CUSTOM_2);
+    long expectedJobSchedulerInitialStartDelay = 15;
+    long expectedJobSchedulerPeriod = 10;
+    TimeUnit expectedJobSchedulerPeriodTimeUnit = TimeUnit.DAYS;
+    List<String> expectedJobSchedulerCustomJobs = List.of("Job_A", "Job_B");
 
     // when
+    Map<DayOfWeek, Set<LocalTimeInterval>> expectedWorkingTimeSchedule =
+        Map.of(DayOfWeek.MONDAY, Set.of(new LocalTimeInterval(LocalTime.MIN, LocalTime.NOON)));
     TaskanaConfiguration configuration =
         new Builder(TestContainerExtension.createDataSourceForH2(), false, "TASKANA")
             .domains(expectedDomains)
@@ -127,6 +137,17 @@ class TaskanaConfigurationTest {
             .userRefreshJobRunEvery(expectedUserRefreshJobRunEvery)
             .addAdditionalUserInfo(true)
             .minimalPermissionsToAssignDomains(expectedMinimalPermissionsToAssignDomains)
+            .jobSchedulerEnabled(false)
+            .jobSchedulerInitialStartDelay(expectedJobSchedulerInitialStartDelay)
+            .jobSchedulerPeriod(expectedJobSchedulerPeriod)
+            .jobSchedulerPeriodTimeUnit(expectedJobSchedulerPeriodTimeUnit)
+            .jobSchedulerEnableTaskCleanupJob(false)
+            .jobSchedulerEnableTaskUpdatePriorityJob(false)
+            .jobSchedulerEnableWorkbasketCleanupJob(false)
+            .jobSchedulerEnableUserInfoRefreshJob(false)
+            .jobSchedulerEnableHistorieCleanupJob(false)
+            .jobSchedulerCustomJobs(expectedJobSchedulerCustomJobs)
+            .workingTimeSchedule(expectedWorkingTimeSchedule)
             .build();
 
     // then
@@ -155,6 +176,19 @@ class TaskanaConfigurationTest {
     assertThat(configuration.isAddAdditionalUserInfo()).isTrue();
     assertThat(configuration.getMinimalPermissionsToAssignDomains())
         .isEqualTo(expectedMinimalPermissionsToAssignDomains);
+    assertThat(configuration.isJobSchedulerEnabled()).isFalse();
+    assertThat(configuration.getJobSchedulerInitialStartDelay())
+        .isEqualTo(expectedJobSchedulerInitialStartDelay);
+    assertThat(configuration.getJobSchedulerPeriod()).isEqualTo(expectedJobSchedulerPeriod);
+    assertThat(configuration.getJobSchedulerPeriodTimeUnit())
+        .isEqualTo(expectedJobSchedulerPeriodTimeUnit);
+    assertThat(configuration.isJobSchedulerEnableTaskCleanupJob()).isFalse();
+    assertThat(configuration.isJobSchedulerEnableTaskUpdatePriorityJob()).isFalse();
+    assertThat(configuration.isJobSchedulerEnableWorkbasketCleanupJob()).isFalse();
+    assertThat(configuration.isJobSchedulerEnableUserInfoRefreshJob()).isFalse();
+    assertThat(configuration.isJobSchedulerEnableHistorieCleanupJob()).isFalse();
+    assertThat(configuration.getJobSchedulerCustomJobs()).isEqualTo(expectedJobSchedulerCustomJobs);
+    assertThat(configuration.getWorkingTimeSchedule()).isEqualTo(expectedWorkingTimeSchedule);
   }
 
   @Test
@@ -184,6 +218,19 @@ class TaskanaConfigurationTest {
             .userRefreshJobRunEvery(Duration.ofDays(5))
             .addAdditionalUserInfo(true)
             .minimalPermissionsToAssignDomains(List.of(WorkbasketPermission.CUSTOM_2))
+            .jobSchedulerEnabled(false)
+            .jobSchedulerInitialStartDelay(10)
+            .jobSchedulerPeriod(15)
+            .jobSchedulerPeriodTimeUnit(TimeUnit.DAYS)
+            .jobSchedulerEnableTaskCleanupJob(false)
+            .jobSchedulerEnableTaskUpdatePriorityJob(false)
+            .jobSchedulerEnableWorkbasketCleanupJob(false)
+            .jobSchedulerEnableUserInfoRefreshJob(false)
+            .jobSchedulerEnableHistorieCleanupJob(false)
+            .jobSchedulerCustomJobs(List.of("Job_A", "Job_B"))
+            .workingTimeSchedule(
+                Map.of(
+                    DayOfWeek.MONDAY, Set.of(new LocalTimeInterval(LocalTime.MIN, LocalTime.NOON))))
             .build();
 
     TaskanaConfiguration copyConfiguration = new Builder(configuration).build();

lib/taskana-core-test/src/test/java/acceptance/common/TaskanaEngineExplizitTest.java

@@ -0,0 +1,38 @@
+package acceptance.common;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+
+import pro.taskana.TaskanaConfiguration;
+import pro.taskana.common.api.TaskanaEngine;
+import pro.taskana.common.api.TaskanaEngine.ConnectionManagementMode;
+import pro.taskana.common.internal.configuration.DB;
+import pro.taskana.common.internal.configuration.DbSchemaCreator;
+import pro.taskana.testapi.OracleSchemaHelper;
+import pro.taskana.testapi.extensions.TestContainerExtension;
+
+class TaskanaEngineExplizitTest {
+
+  @Test
+  void should_CreateTaskanaEnine_When_ExplizitModeIsActive() throws Exception {
+
+    String schemaName = TestContainerExtension.determineSchemaName();
+    if (DB.isOracle(TestContainerExtension.EXECUTION_DATABASE.dbProductId)) {
+      OracleSchemaHelper.initOracleSchema(TestContainerExtension.DATA_SOURCE, schemaName);
+    }
+
+    TaskanaConfiguration taskanaEngineConfiguration =
+        new TaskanaConfiguration.Builder(
+                TestContainerExtension.DATA_SOURCE, false, schemaName, true)
+            .initTaskanaProperties()
+            .build();
+
+    TaskanaEngine.buildTaskanaEngine(taskanaEngineConfiguration, ConnectionManagementMode.EXPLICIT);
+
+    DbSchemaCreator dsc =
+        new DbSchemaCreator(
+            taskanaEngineConfiguration.getDatasource(), taskanaEngineConfiguration.getSchemaName());
+    assertThat(dsc.isValidSchemaVersion(TaskanaEngine.MINIMAL_TASKANA_SCHEMA_VERSION)).isTrue();
+  }
+}

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

@@ -0,0 +1,21 @@
+package acceptance.jobs;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import pro.taskana.common.internal.jobs.Clock;
+
+public class FakeClock implements Clock {
+
+  List<ClockListener> listeners = new ArrayList<>();
+
+  @Override
+  public void register(ClockListener listener) {
+    listeners.add(listener);
+  }
+
+  @Override
+  public void start() {
+    listeners.forEach(ClockListener::timeElapsed);
+  }
+}

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

@@ -0,0 +1,177 @@
+package acceptance.jobs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.TestInstance.Lifecycle;
+
+import pro.taskana.TaskanaConfiguration;
+import pro.taskana.TaskanaConfiguration.Builder;
+import pro.taskana.classification.api.ClassificationService;
+import pro.taskana.classification.api.models.ClassificationSummary;
+import pro.taskana.common.api.ScheduledJob;
+import pro.taskana.common.api.TaskanaEngine;
+import pro.taskana.common.api.TaskanaEngine.ConnectionManagementMode;
+import pro.taskana.common.api.exceptions.SystemException;
+import pro.taskana.common.internal.JobMapper;
+import pro.taskana.common.internal.jobs.AbstractTaskanaJob;
+import pro.taskana.common.internal.jobs.JobScheduler;
+import pro.taskana.common.internal.transaction.TaskanaTransactionProvider;
+import pro.taskana.task.api.TaskService;
+import pro.taskana.task.api.TaskState;
+import pro.taskana.task.api.models.ObjectReference;
+import pro.taskana.task.api.models.TaskSummary;
+import pro.taskana.task.internal.jobs.TaskCleanupJob;
+import pro.taskana.testapi.DefaultTestEntities;
+import pro.taskana.testapi.TaskanaEngineConfigurationModifier;
+import pro.taskana.testapi.TaskanaInject;
+import pro.taskana.testapi.TaskanaIntegrationTest;
+import pro.taskana.testapi.builder.TaskBuilder;
+import pro.taskana.testapi.security.WithAccessId;
+import pro.taskana.workbasket.api.WorkbasketService;
+import pro.taskana.workbasket.api.models.WorkbasketSummary;
+
+@TaskanaIntegrationTest
+class JobSchedulerExecutionAccTest implements TaskanaEngineConfigurationModifier {
+  @TaskanaInject TaskanaConfiguration taskanaConfiguration;
+  @TaskanaInject TaskService taskService;
+  @TaskanaInject JobMapper jobMapper;
+  WorkbasketSummary workbasket;
+  ClassificationSummary classification;
+  ObjectReference primaryObjRef;
+
+  @Override
+  public Builder modify(Builder taskanaEngineConfigurationBuilder) {
+    return taskanaEngineConfigurationBuilder
+        .jobSchedulerEnableTaskCleanupJob(true)
+        .cleanupJobFirstRun(Instant.now().minus(10, ChronoUnit.MILLIS))
+        .cleanupJobRunEvery(Duration.ofMillis(1))
+        .cleanupJobMinimumAge(Duration.ofMillis(10));
+  }
+
+  @WithAccessId(user = "businessadmin")
+  @BeforeEach
+  void setup(WorkbasketService workbasketService, ClassificationService classificationService)
+      throws Exception {
+    workbasket =
+        DefaultTestEntities.defaultTestWorkbasket().buildAndStoreAsSummary(workbasketService);
+    classification =
+        DefaultTestEntities.defaultTestClassification()
+            .buildAndStoreAsSummary(classificationService);
+    primaryObjRef = DefaultTestEntities.defaultTestObjectReference().build();
+  }
+
+  @WithAccessId(user = "admin")
+  @Test
+  void should_ExecuteAJobSuccessfully() throws Exception {
+    Instant timeStampAnyJobIsOverdue = Instant.now().plus(10, ChronoUnit.DAYS);
+    TaskanaEngine taskanaEngine =
+        TaskanaEngine.buildTaskanaEngine(taskanaConfiguration, ConnectionManagementMode.EXPLICIT);
+    JobScheduler jobScheduler = new JobScheduler(taskanaEngine, new FakeClock());
+    TaskBuilder.newTask()
+        .workbasketSummary(workbasket)
+        .classificationSummary(classification)
+        .primaryObjRef(primaryObjRef)
+        .state(TaskState.COMPLETED)
+        .completed(Instant.now().minus(5, ChronoUnit.DAYS))
+        .buildAndStoreAsSummary(taskService);
+    final List<ScheduledJob> jobsToRun = jobMapper.findJobsToRun(timeStampAnyJobIsOverdue);
+
+    Thread.sleep(2); // to make sure that TaskCleanupJob is overdue
+    jobScheduler.start();
+
+    List<TaskSummary> existingTasks = taskService.createTaskQuery().list();
+    assertThat(existingTasks).isEmpty();
+    List<ScheduledJob> jobsToRunAfter = jobMapper.findJobsToRun(timeStampAnyJobIsOverdue);
+    assertThat(jobsToRunAfter).isNotEmpty().doesNotContainAnyElementsOf(jobsToRun);
+  }
+
+  public static class AlwaysFailJob extends AbstractTaskanaJob {
+
+    public AlwaysFailJob(
+        TaskanaEngine taskanaEngine, TaskanaTransactionProvider txProvider, ScheduledJob job) {
+      super(taskanaEngine, txProvider, job, true);
+    }
+
+    @Override
+    protected String getType() {
+      return AlwaysFailJob.class.getName();
+    }
+
+    @Override
+    protected void execute() {
+      throw new SystemException("I always fail. Muahhahaa!");
+    }
+  }
+
+  @Nested
+  @TestInstance(Lifecycle.PER_CLASS)
+  class AJobFails implements TaskanaEngineConfigurationModifier {
+
+    @TaskanaInject TaskanaConfiguration taskanaConfiguration;
+    @TaskanaInject TaskService taskService;
+    @TaskanaInject JobMapper jobMapper;
+    WorkbasketSummary workbasket;
+    ClassificationSummary classification;
+    ObjectReference primaryObjRef;
+
+    @Override
+    public Builder modify(Builder taskanaEngineConfigurationBuilder) {
+      return taskanaEngineConfigurationBuilder
+          .jobSchedulerEnableTaskCleanupJob(false)
+          .cleanupJobFirstRun(Instant.now().minus(10, ChronoUnit.MILLIS))
+          .cleanupJobRunEvery(Duration.ofMillis(1))
+          .cleanupJobMinimumAge(Duration.ofMillis(10))
+          .jobSchedulerCustomJobs(
+              List.of(AlwaysFailJob.class.getName(), TaskCleanupJob.class.getName()));
+    }
+
+    @WithAccessId(user = "businessadmin")
+    @BeforeEach
+    void setup(WorkbasketService workbasketService, ClassificationService classificationService)
+        throws Exception {
+      workbasket =
+          DefaultTestEntities.defaultTestWorkbasket().buildAndStoreAsSummary(workbasketService);
+      classification =
+          DefaultTestEntities.defaultTestClassification()
+              .buildAndStoreAsSummary(classificationService);
+      primaryObjRef = DefaultTestEntities.defaultTestObjectReference().build();
+    }
+
+    @WithAccessId(user = "admin")
+    @Test
+    void should_ContinueExecutingJobs_When_ASingeJobFails() throws Exception {
+      Instant timeStampAnyJobIsOverdue = Instant.now().plus(10, ChronoUnit.DAYS);
+      TaskanaEngine taskanaEngine =
+          TaskanaEngine.buildTaskanaEngine(taskanaConfiguration, ConnectionManagementMode.EXPLICIT);
+      JobScheduler jobScheduler = new JobScheduler(taskanaEngine, new FakeClock());
+      TaskBuilder.newTask()
+          .workbasketSummary(workbasket)
+          .classificationSummary(classification)
+          .primaryObjRef(primaryObjRef)
+          .state(TaskState.COMPLETED)
+          .completed(Instant.now().minus(5, ChronoUnit.DAYS))
+          .buildAndStoreAsSummary(taskService);
+      final List<ScheduledJob> jobsToRun = jobMapper.findJobsToRun(timeStampAnyJobIsOverdue);
+
+      Thread.sleep(2); // to make sure that TaskCleanupJob is overdue
+      jobScheduler.start();
+
+      List<TaskSummary> existingTasks = taskService.createTaskQuery().list();
+      assertThat(existingTasks).isEmpty();
+      List<ScheduledJob> jobsToRunAfter = jobMapper.findJobsToRun(timeStampAnyJobIsOverdue);
+      assertThat(jobsToRunAfter).isNotEmpty().doesNotContainAnyElementsOf(jobsToRun);
+      assertThat(jobsToRunAfter)
+          .filteredOn(job -> AlwaysFailJob.class.getName().equals(job.getType()))
+          .extracting(ScheduledJob::getRetryCount)
+          .containsExactly(2);
+    }
+  }
+}

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

@@ -0,0 +1,60 @@
+package acceptance.jobs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+
+import pro.taskana.TaskanaConfiguration;
+import pro.taskana.common.api.ScheduledJob;
+import pro.taskana.common.internal.JobMapper;
+import pro.taskana.task.internal.jobs.TaskCleanupJob;
+import pro.taskana.task.internal.jobs.TaskUpdatePriorityJob;
+import pro.taskana.testapi.TaskanaEngineConfigurationModifier;
+import pro.taskana.testapi.TaskanaInject;
+import pro.taskana.testapi.TaskanaIntegrationTest;
+import pro.taskana.workbasket.internal.jobs.WorkbasketCleanupJob;
+
+@TaskanaIntegrationTest
+class JobSchedulerInitAccTest implements TaskanaEngineConfigurationModifier {
+
+  @TaskanaInject JobMapper jobMapper;
+
+  Instant firstRun = Instant.now().minus(2, ChronoUnit.MINUTES).truncatedTo(ChronoUnit.MILLIS);
+  Duration runEvery = Duration.ofMinutes(5);
+
+  @Override
+  public TaskanaConfiguration.Builder modify(
+      TaskanaConfiguration.Builder taskanaEngineConfigurationBuilder) {
+    return taskanaEngineConfigurationBuilder
+        .cleanupJobRunEvery(runEvery)
+        .cleanupJobFirstRun(firstRun)
+        // config for TaskUpdatePriorityJob
+        .priorityJobActive(true)
+        .priorityJobRunEvery(runEvery)
+        .priorityJobFirstRun(firstRun)
+        .jobSchedulerEnabled(true)
+        .jobSchedulerInitialStartDelay(100)
+        .jobSchedulerPeriod(100)
+        .jobSchedulerPeriodTimeUnit(TimeUnit.SECONDS)
+        .jobSchedulerEnableTaskCleanupJob(true)
+        .jobSchedulerEnableTaskUpdatePriorityJob(true)
+        .jobSchedulerEnableWorkbasketCleanupJob(true);
+  }
+
+  @Test
+  void should_StartTheJobsImmediately_When_StartMethodIsCalled() throws Exception {
+    List<ScheduledJob> nextJobs = jobMapper.findJobsToRun(Instant.now().plus(runEvery));
+    assertThat(nextJobs).extracting(ScheduledJob::getDue).containsOnly(firstRun.plus(runEvery));
+    assertThat(nextJobs)
+        .extracting(ScheduledJob::getType)
+        .containsExactlyInAnyOrder(
+            TaskUpdatePriorityJob.class.getName(),
+            TaskCleanupJob.class.getName(),
+            WorkbasketCleanupJob.class.getName());
+  }
+}

lib/taskana-core-test/src/test/resources/taskana.properties

@@ -18,4 +18,18 @@ taskana.german.holidays.corpus-christi.enabled=false
 taskana.history.deletion.on.task.deletion.enabled=true
 taskana.validation.allowTimestampServiceLevelMismatch=false
 taskana.query.includeLongName=false
-
+# enable or disable the jobscheduler at all
+# set it to false and no jobs are running
+taskana.jobscheduler.enabled=false
+# wait time before the first job run in millis
+taskana.jobscheduler.initialstartdelay=100000
+# sleeping time befor the next job runs
+taskana.jobscheduler.period=12
+# timeunit for the sleeping period
+# Possible values: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+taskana.jobscheduler.periodtimeunit=HOURS
+taskana.jobscheduler.enableTaskCleanupJob=false
+taskana.jobscheduler.enableTaskUpdatePriorityJob=false
+taskana.jobscheduler.enableWorkbasketCleanupJob=false
+taskana.jobscheduler.enableUserInfoRefreshJob=false
+taskana.jobscheduler.enableHistorieCleanupJob=false

lib/taskana-core/src/main/java/pro/taskana/TaskanaConfiguration.java

@@ -23,6 +23,7 @@ import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Properties;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import javax.sql.DataSource;
 import org.slf4j.Logger;
@@ -65,6 +66,16 @@ public class TaskanaConfiguration {
   // region custom configuration
   private final Map<String, String> properties;
   private final Map<DayOfWeek, Set<LocalTimeInterval>> workingTimeSchedule;
+  private final boolean jobSchedulerEnabled;
+  private final long jobSchedulerInitialStartDelay;
+  private final long jobSchedulerPeriod;
+  private final TimeUnit jobSchedulerPeriodTimeUnit;
+  private final boolean jobSchedulerEnableTaskCleanupJob;
+  private final boolean jobSchedulerEnableTaskUpdatePriorityJob;
+  private final boolean jobSchedulerEnableWorkbasketCleanupJob;
+  private final boolean jobSchedulerEnableUserInfoRefreshJob;
+  private final boolean jobSchedulerEnableHistorieCleanupJob;
+  private final List<String> jobSchedulerCustomJobs;
 
   @TaskanaProperty("taskana.domains")
   private List<String> domains = new ArrayList<>();
@@ -96,6 +107,7 @@ public class TaskanaConfiguration {
   // TODO: validate this is positive
   @TaskanaProperty("taskana.jobs.cleanup.runEvery")
   private Duration cleanupJobRunEvery = Duration.ofDays(1);
+  // endregion
   // TODO: validate this is positive
   @TaskanaProperty("taskana.jobs.cleanup.minimumAge")
   private Duration cleanupJobMinimumAge = Duration.ofDays(14);
@@ -127,7 +139,6 @@ public class TaskanaConfiguration {
   // TODO: make Set
   @TaskanaProperty("taskana.user.minimalPermissionsToAssignDomains")
   private List<WorkbasketPermission> minimalPermissionsToAssignDomains = new ArrayList<>();
-  // endregion
 
   protected TaskanaConfiguration(Builder builder) {
     this.dataSource = builder.dataSource;
@@ -173,6 +184,16 @@ public class TaskanaConfiguration {
     this.userRefreshJobFirstRun = builder.userRefreshJobFirstRun;
     this.minimalPermissionsToAssignDomains =
         Collections.unmodifiableList(builder.minimalPermissionsToAssignDomains);
+    this.jobSchedulerEnabled = builder.jobSchedulerEnabled;
+    this.jobSchedulerInitialStartDelay = builder.jobSchedulerInitialStartDelay;
+    this.jobSchedulerPeriod = builder.jobSchedulerPeriod;
+    this.jobSchedulerPeriodTimeUnit = builder.jobSchedulerPeriodTimeUnit;
+    this.jobSchedulerEnableTaskCleanupJob = builder.jobSchedulerEnableTaskCleanupJob;
+    this.jobSchedulerEnableTaskUpdatePriorityJob = builder.jobSchedulerEnableTaskUpdatePriorityJob;
+    this.jobSchedulerEnableWorkbasketCleanupJob = builder.jobSchedulerEnableWorkbasketCleanupJob;
+    this.jobSchedulerEnableUserInfoRefreshJob = builder.jobSchedulerEnableUserInfoRefreshJob;
+    this.jobSchedulerEnableHistorieCleanupJob = builder.jobSchedulerEnableHistorieCleanupJob;
+    this.jobSchedulerCustomJobs = Collections.unmodifiableList(builder.jobSchedulerCustomJobs);
 
     if (LOGGER.isDebugEnabled()) {
       // TODO remove the reflection magic when introducing lombok toString magic :-)
@@ -332,6 +353,46 @@ public class TaskanaConfiguration {
     return schemaName;
   }
 
+  public boolean isJobSchedulerEnabled() {
+    return jobSchedulerEnabled;
+  }
+
+  public long getJobSchedulerInitialStartDelay() {
+    return jobSchedulerInitialStartDelay;
+  }
+
+  public long getJobSchedulerPeriod() {
+    return jobSchedulerPeriod;
+  }
+
+  public TimeUnit getJobSchedulerPeriodTimeUnit() {
+    return jobSchedulerPeriodTimeUnit;
+  }
+
+  public boolean isJobSchedulerEnableTaskCleanupJob() {
+    return jobSchedulerEnableTaskCleanupJob;
+  }
+
+  public boolean isJobSchedulerEnableTaskUpdatePriorityJob() {
+    return jobSchedulerEnableTaskUpdatePriorityJob;
+  }
+
+  public boolean isJobSchedulerEnableWorkbasketCleanupJob() {
+    return jobSchedulerEnableWorkbasketCleanupJob;
+  }
+
+  public boolean isJobSchedulerEnableUserInfoRefreshJob() {
+    return jobSchedulerEnableUserInfoRefreshJob;
+  }
+
+  public boolean isJobSchedulerEnableHistorieCleanupJob() {
+    return jobSchedulerEnableHistorieCleanupJob;
+  }
+
+  public List<String> getJobSchedulerCustomJobs() {
+    return jobSchedulerCustomJobs;
+  }
+
   /**
    * Helper method to determine whether all access ids (user Id and group ids) should be used in
    * lower case.
@@ -434,6 +495,36 @@ public class TaskanaConfiguration {
 
     @TaskanaProperty("taskana.jobs.user.refresh.firstRunAt")
     private Instant userRefreshJobFirstRun = Instant.parse("2018-01-01T23:00:00Z");
+
+    @TaskanaProperty("taskana.jobscheduler.enabled")
+    private boolean jobSchedulerEnabled = true;
+
+    @TaskanaProperty("taskana.jobscheduler.initialstartdelay")
+    private long jobSchedulerInitialStartDelay = 100;
+
+    @TaskanaProperty("taskana.jobscheduler.period")
+    private long jobSchedulerPeriod = 12;
+
+    @TaskanaProperty("taskana.jobscheduler.periodtimeunit")
+    private TimeUnit jobSchedulerPeriodTimeUnit = TimeUnit.HOURS;
+
+    @TaskanaProperty("taskana.jobscheduler.enableTaskCleanupJob")
+    private boolean jobSchedulerEnableTaskCleanupJob = true;
+
+    @TaskanaProperty("taskana.jobscheduler.enableTaskUpdatePriorityJob")
+    private boolean jobSchedulerEnableTaskUpdatePriorityJob = true;
+
+    @TaskanaProperty("taskana.jobscheduler.enableWorkbasketCleanupJob")
+    private boolean jobSchedulerEnableWorkbasketCleanupJob = true;
+
+    @TaskanaProperty("taskana.jobscheduler.enableUserInfoRefreshJob")
+    private boolean jobSchedulerEnableUserInfoRefreshJob = true;
+
+    @TaskanaProperty("taskana.jobscheduler.enableHistorieCleanupJob")
+    private boolean jobSchedulerEnableHistorieCleanupJob = true;
+
+    @TaskanaProperty("taskana.jobscheduler.customJobs")
+    private List<String> jobSchedulerCustomJobs = new ArrayList<>();
     // endregion
 
     // region user configuration
@@ -480,6 +571,17 @@ public class TaskanaConfiguration {
       this.userRefreshJobRunEvery = tec.getUserRefreshJobRunEvery();
       this.userRefreshJobFirstRun = tec.getUserRefreshJobFirstRun();
       this.minimalPermissionsToAssignDomains = tec.getMinimalPermissionsToAssignDomains();
+      this.jobSchedulerEnabled = tec.isJobSchedulerEnabled();
+      this.jobSchedulerInitialStartDelay = tec.getJobSchedulerInitialStartDelay();
+      this.jobSchedulerPeriod = tec.getJobSchedulerPeriod();
+      this.jobSchedulerPeriodTimeUnit = tec.getJobSchedulerPeriodTimeUnit();
+      this.jobSchedulerEnableTaskCleanupJob = tec.isJobSchedulerEnableTaskCleanupJob();
+      this.jobSchedulerEnableTaskUpdatePriorityJob =
+          tec.isJobSchedulerEnableTaskUpdatePriorityJob();
+      this.jobSchedulerEnableWorkbasketCleanupJob = tec.isJobSchedulerEnableWorkbasketCleanupJob();
+      this.jobSchedulerEnableUserInfoRefreshJob = tec.isJobSchedulerEnableUserInfoRefreshJob();
+      this.jobSchedulerEnableHistorieCleanupJob = tec.isJobSchedulerEnableHistorieCleanupJob();
+      this.jobSchedulerCustomJobs = tec.getJobSchedulerCustomJobs();
     }
 
     public Builder(DataSource dataSource, boolean useManagedTransactions, String schemaName) {
@@ -649,6 +751,65 @@ public class TaskanaConfiguration {
       return this;
     }
 
+    public Builder jobSchedulerEnabled(boolean jobSchedulerEnabled) {
+      this.jobSchedulerEnabled = jobSchedulerEnabled;
+      return this;
+    }
+
+    public Builder jobSchedulerInitialStartDelay(long jobSchedulerInitialStartDelay) {
+      this.jobSchedulerInitialStartDelay = jobSchedulerInitialStartDelay;
+      return this;
+    }
+
+    public Builder jobSchedulerPeriod(long jobSchedulerPeriod) {
+      this.jobSchedulerPeriod = jobSchedulerPeriod;
+      return this;
+    }
+
+    public Builder jobSchedulerPeriodTimeUnit(TimeUnit jobSchedulerPeriodTimeUnit) {
+      this.jobSchedulerPeriodTimeUnit = jobSchedulerPeriodTimeUnit;
+      return this;
+    }
+
+    public Builder jobSchedulerEnableTaskCleanupJob(boolean jobSchedulerEnableTaskCleanupJob) {
+      this.jobSchedulerEnableTaskCleanupJob = jobSchedulerEnableTaskCleanupJob;
+      return this;
+    }
+
+    public Builder jobSchedulerEnableTaskUpdatePriorityJob(
+        boolean jobSchedulerEnableTaskUpdatePriorityJob) {
+      this.jobSchedulerEnableTaskUpdatePriorityJob = jobSchedulerEnableTaskUpdatePriorityJob;
+      return this;
+    }
+
+    public Builder jobSchedulerEnableWorkbasketCleanupJob(
+        boolean jobSchedulerEnableWorkbasketCleanupJob) {
+      this.jobSchedulerEnableWorkbasketCleanupJob = jobSchedulerEnableWorkbasketCleanupJob;
+      return this;
+    }
+
+    public Builder jobSchedulerEnableUserInfoRefreshJob(
+        boolean jobSchedulerEnableUserInfoRefreshJob) {
+      this.jobSchedulerEnableUserInfoRefreshJob = jobSchedulerEnableUserInfoRefreshJob;
+      return this;
+    }
+
+    public Builder jobSchedulerEnableHistorieCleanupJob(
+        boolean jobSchedulerEnableHistorieCleanupJob) {
+      this.jobSchedulerEnableHistorieCleanupJob = jobSchedulerEnableHistorieCleanupJob;
+      return this;
+    }
+
+    public Builder jobSchedulerCustomJobs(List<String> jobSchedulerCustomJobs) {
+      this.jobSchedulerCustomJobs = jobSchedulerCustomJobs;
+      return this;
+    }
+
+    public Builder workingTimeSchedule(Map<DayOfWeek, Set<LocalTimeInterval>> workingTimeSchedule) {
+      this.workingTimeSchedule = workingTimeSchedule;
+      return this;
+    }
+
     public TaskanaConfiguration build() {
       return new TaskanaConfiguration(this);
     }

lib/taskana-core/src/main/java/pro/taskana/common/api/TaskanaEngine.java

@@ -17,6 +17,7 @@ import pro.taskana.workbasket.api.WorkbasketService;
 
 /** The TaskanaEngine represents an overall set of all needed services. */
 public interface TaskanaEngine {
+  String MINIMAL_TASKANA_SCHEMA_VERSION = "5.2.0";
 
   /**
    * Returns a {@linkplain TaskService} initialized with the current TaskanaEngine. {@linkplain

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

@@ -1,5 +1,7 @@
 package pro.taskana.common.internal;
 
+import static pro.taskana.common.api.TaskanaEngine.ConnectionManagementMode.EXPLICIT;
+
 import java.security.PrivilegedAction;
 import java.sql.Connection;
 import java.sql.SQLException;
@@ -43,6 +45,8 @@ import pro.taskana.common.api.security.CurrentUserContext;
 import pro.taskana.common.api.security.UserPrincipal;
 import pro.taskana.common.internal.configuration.DB;
 import pro.taskana.common.internal.configuration.DbSchemaCreator;
+import pro.taskana.common.internal.jobs.JobScheduler;
+import pro.taskana.common.internal.jobs.RealClock;
 import pro.taskana.common.internal.persistence.InstantTypeHandler;
 import pro.taskana.common.internal.persistence.MapTypeHandler;
 import pro.taskana.common.internal.persistence.StringTypeHandler;
@@ -83,7 +87,6 @@ import pro.taskana.workbasket.internal.WorkbasketServiceImpl;
 public class TaskanaEngineImpl implements TaskanaEngine {
 
   // must match the VERSION value in table
-  private static final String MINIMAL_TASKANA_SCHEMA_VERSION = "5.2.0";
   private static final Logger LOGGER = LoggerFactory.getLogger(TaskanaEngineImpl.class);
   private static final SessionStack SESSION_STACK = new SessionStack();
   protected final TaskanaConfiguration taskanaEngineConfiguration;
@@ -114,8 +117,14 @@ public class TaskanaEngineImpl implements TaskanaEngine {
         "initializing TASKANA with this configuration: {} and this mode: {}",
         taskanaEngineConfiguration,
         connectionManagementMode);
-    this.taskanaEngineConfiguration = taskanaEngineConfiguration;
+    if (connectionManagementMode == EXPLICIT) {
+      // at first we initialize Taskana DB with autocommit,
+      // at the end of constructor the mode is set
+      this.mode = ConnectionManagementMode.AUTOCOMMIT;
+    } else {
       this.mode = connectionManagementMode;
+    }
+    this.taskanaEngineConfiguration = taskanaEngineConfiguration;
     internalTaskanaEngineImpl = new InternalTaskanaEngineImpl();
     HolidaySchedule holidaySchedule =
         new HolidaySchedule(
@@ -143,6 +152,24 @@ public class TaskanaEngineImpl implements TaskanaEngine {
     afterRequestReviewManager = new AfterRequestReviewManager(this);
     beforeRequestChangesManager = new BeforeRequestChangesManager(this);
     afterRequestChangesManager = new AfterRequestChangesManager(this);
+
+    if (this.taskanaEngineConfiguration.isJobSchedulerEnabled()) {
+      TaskanaConfiguration tec =
+          new TaskanaConfiguration.Builder(this.taskanaEngineConfiguration)
+              .jobSchedulerEnabled(false)
+              .build();
+      TaskanaEngine taskanaEngine = TaskanaEngine.buildTaskanaEngine(tec, EXPLICIT);
+      RealClock clock =
+          new RealClock(
+              this.taskanaEngineConfiguration.getJobSchedulerInitialStartDelay(),
+              this.taskanaEngineConfiguration.getJobSchedulerPeriod(),
+              this.taskanaEngineConfiguration.getJobSchedulerPeriodTimeUnit());
+      JobScheduler jobScheduler = new JobScheduler(taskanaEngine, clock);
+      jobScheduler.start();
+    }
+
+    // don't remove, to reset possible explicit mode
+    this.mode = connectionManagementMode;
   }
 
   public static TaskanaEngine createTaskanaEngine(
@@ -194,6 +221,25 @@ public class TaskanaEngineImpl implements TaskanaEngine {
         sessionManager.getMapper(TaskMapper.class));
   }
 
+  public Connection getConnection() {
+    return connection;
+  }
+
+  @Override
+  public void setConnection(Connection connection) throws SQLException {
+    if (connection != null) {
+      this.connection = connection;
+      // disabling auto commit for passed connection in order to gain full control over the
+      // connection management
+      connection.setAutoCommit(false);
+      connection.setSchema(taskanaEngineConfiguration.getSchemaName());
+      mode = EXPLICIT;
+      sessionManager.startManagedSession(connection);
+    } else if (this.connection != null) {
+      closeConnection();
+    }
+  }
+
   // This should be part of the InternalTaskanaEngine. Unfortunately the jobs don't have access to
   // that engine.
   // Therefore, this getter exits and will be removed as soon as our jobs will be refactored.
@@ -234,9 +280,7 @@ public class TaskanaEngineImpl implements TaskanaEngine {
 
   @Override
   public void setConnectionManagementMode(ConnectionManagementMode mode) {
-    if (this.mode == ConnectionManagementMode.EXPLICIT
+    if (this.mode == EXPLICIT && connection != null && mode != EXPLICIT) {
-        && connection != null
-        && mode != ConnectionManagementMode.EXPLICIT) {
       if (sessionManager.isManagedSessionStarted()) {
         sessionManager.close();
       }
@@ -245,24 +289,9 @@ public class TaskanaEngineImpl implements TaskanaEngine {
     this.mode = mode;
   }
 
-  @Override
-  public void setConnection(Connection connection) throws SQLException {
-    if (connection != null) {
-      this.connection = connection;
-      // disabling auto commit for passed connection in order to gain full control over the
-      // connection management
-      connection.setAutoCommit(false);
-      connection.setSchema(taskanaEngineConfiguration.getSchemaName());
-      mode = ConnectionManagementMode.EXPLICIT;
-      sessionManager.startManagedSession(connection);
-    } else if (this.connection != null) {
-      closeConnection();
-    }
-  }
-
   @Override
   public void closeConnection() {
-    if (this.mode == ConnectionManagementMode.EXPLICIT) {
+    if (this.mode == EXPLICIT) {
       this.connection = null;
       if (sessionManager.isManagedSessionStarted()) {
         sessionManager.close();
@@ -397,21 +426,24 @@ public class TaskanaEngineImpl implements TaskanaEngine {
     return SqlSessionManager.newInstance(localSessionFactory);
   }
 
-  private void initializeDbSchema(TaskanaConfiguration taskanaEngineConfiguration)
+  private boolean initializeDbSchema(TaskanaConfiguration taskanaEngineConfiguration)
       throws SQLException {
     DbSchemaCreator dbSchemaCreator =
         new DbSchemaCreator(
             taskanaEngineConfiguration.getDatasource(), taskanaEngineConfiguration.getSchemaName());
-    dbSchemaCreator.run();
+    boolean schemaCreated = dbSchemaCreator.run();
 
+    if (!schemaCreated) {
       if (!dbSchemaCreator.isValidSchemaVersion(MINIMAL_TASKANA_SCHEMA_VERSION)) {
         throw new SystemException(
             "The Database Schema Version doesn't match the expected minimal version "
                 + MINIMAL_TASKANA_SCHEMA_VERSION);
       }
+    }
     ((ConfigurationServiceImpl) getConfigurationService())
         .checkSecureAccess(taskanaEngineConfiguration.isSecurityEnabled());
     ((ConfigurationServiceImpl) getConfigurationService()).setupDefaultCustomAttributes();
+    return schemaCreated;
   }
 
   /**
@@ -483,14 +515,14 @@ public class TaskanaEngineImpl implements TaskanaEngine {
                 + "to the database. No schema has been created.",
             e.getCause());
       }
-      if (mode != ConnectionManagementMode.EXPLICIT) {
+      if (mode != EXPLICIT) {
         SESSION_STACK.pushSessionToStack(sessionManager);
       }
     }
 
     @Override
     public void returnConnection() {
-      if (mode != ConnectionManagementMode.EXPLICIT) {
+      if (mode != EXPLICIT) {
         SESSION_STACK.popSessionFromStack();
         if (SESSION_STACK.getSessionStack().isEmpty()
             && sessionManager != null
@@ -520,10 +552,9 @@ public class TaskanaEngineImpl implements TaskanaEngine {
 
     @Override
     public void initSqlSession() {
-      if (mode == ConnectionManagementMode.EXPLICIT && connection == null) {
+      if (mode == EXPLICIT && connection == null) {
         throw new ConnectionNotSetException();
-      } else if (mode != ConnectionManagementMode.EXPLICIT
+      } else if (mode != EXPLICIT && !sessionManager.isManagedSessionStarted()) {
-          && !sessionManager.isManagedSessionStarted()) {
         sessionManager.startManagedSession();
       }
     }

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

@@ -1,12 +1,15 @@
 package pro.taskana.common.internal.jobs;
 
+import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.time.Duration;
 import java.time.Instant;
 
 import pro.taskana.common.api.ScheduledJob;
 import pro.taskana.common.api.TaskanaEngine;
+import pro.taskana.common.api.exceptions.SystemException;
 import pro.taskana.common.api.exceptions.TaskanaException;
+import pro.taskana.common.internal.JobServiceImpl;
 import pro.taskana.common.internal.TaskanaEngineImpl;
 import pro.taskana.common.internal.transaction.TaskanaTransactionProvider;
 
@@ -54,6 +57,61 @@ public abstract class AbstractTaskanaJob implements TaskanaJob {
     }
   }
 
+  /**
+   * Initializes the TaskCleanupJob schedule. <br>
+   * All scheduled cleanup jobs are cancelled/deleted and a new one is scheduled.
+   *
+   * @param taskanaEngine the TASKANA engine.
+   * @param jobClass the class of the job which should be scheduled
+   * @throws SystemException if the jobClass could not be scheduled.
+   */
+  public static void initializeSchedule(TaskanaEngine taskanaEngine, Class<?> jobClass) {
+    if (!AbstractTaskanaJob.class.isAssignableFrom(jobClass)) {
+      throw new SystemException(
+          String.format("Job '%s' is not a subclass of '%s'", jobClass, AbstractTaskanaJob.class));
+    }
+    Constructor<?> constructor;
+    try {
+      constructor =
+          jobClass.getConstructor(
+              TaskanaEngine.class, TaskanaTransactionProvider.class, ScheduledJob.class);
+    } catch (NoSuchMethodException e) {
+      throw new SystemException(
+          String.format(
+              "Job '%s' does not have a constructor matching (%s, %s, %s)",
+              jobClass, TaskanaEngine.class, TaskanaTransactionProvider.class, ScheduledJob.class));
+    }
+    AbstractTaskanaJob job;
+    try {
+      job = (AbstractTaskanaJob) constructor.newInstance(taskanaEngine, null, null);
+    } catch (InvocationTargetException e) {
+      throw new SystemException(
+          String.format(
+              "Required Constructor(%s, %s, %s) of job '%s' could not be invoked",
+              TaskanaEngine.class, TaskanaTransactionProvider.class, ScheduledJob.class, jobClass),
+          e);
+    } catch (InstantiationException e) {
+      throw new SystemException(
+          String.format(
+              "Required Constructor(%s, %s, %s) of job '%s' could not be initialized",
+              TaskanaEngine.class, TaskanaTransactionProvider.class, ScheduledJob.class, jobClass),
+          e);
+    } catch (IllegalAccessException e) {
+      throw new SystemException(
+          String.format(
+              "Required Constructor(%s, %s, %s) of job '%s' is not public",
+              TaskanaEngine.class, TaskanaTransactionProvider.class, ScheduledJob.class, jobClass),
+          e);
+    }
+    if (!job.async) {
+      throw new SystemException(
+          String.format("Job '%s' is not an async job. Please declare it as async", jobClass));
+    }
+    JobServiceImpl jobService = (JobServiceImpl) taskanaEngine.getJobService();
+    jobService.deleteJobs(job.getType());
+    job.scheduleNextJob();
+  }
+
   protected abstract String getType();
 
   protected abstract void execute() throws TaskanaException;

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

@@ -0,0 +1,14 @@
+package pro.taskana.common.internal.jobs;
+
+public interface Clock {
+  void register(ClockListener listener);
+
+  void start();
+
+  default void stop() {}
+
+  @FunctionalInterface
+  interface ClockListener {
+    void timeElapsed();
+  }
+}

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

@@ -9,7 +9,6 @@ import org.slf4j.LoggerFactory;
 
 import pro.taskana.common.api.ScheduledJob;
 import pro.taskana.common.api.TaskanaEngine;
-import pro.taskana.common.api.exceptions.SystemException;
 import pro.taskana.common.internal.JobServiceImpl;
 import pro.taskana.common.internal.transaction.TaskanaTransactionProvider;
 
@@ -42,16 +41,22 @@ public class JobRunner {
 
   private void runJobTransactionally(ScheduledJob scheduledJob) {
     TaskanaTransactionProvider.executeInTransactionIfPossible(
-        txProvider, () -> taskanaEngine.runAsAdmin(() -> runScheduledJob(scheduledJob)));
+        txProvider,
+        () -> {
+          Boolean successful = taskanaEngine.runAsAdmin(() -> runScheduledJob(scheduledJob));
+          if (successful) {
             jobService.deleteJob(scheduledJob);
           }
+        });
+  }
 
-  private void runScheduledJob(ScheduledJob scheduledJob) {
+  private boolean runScheduledJob(ScheduledJob scheduledJob) {
     try {
       AbstractTaskanaJob.createFromScheduledJob(taskanaEngine, txProvider, scheduledJob).run();
+      return true;
     } catch (Exception e) {
       LOGGER.error("Error running job: {} ", scheduledJob.getType(), e);
-      throw new SystemException(String.format("Error running job '%s'", scheduledJob.getType()), e);
+      return false;
     }
   }
 

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

@@ -0,0 +1,91 @@
+package pro.taskana.common.internal.jobs;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import pro.taskana.common.api.TaskanaEngine;
+import pro.taskana.common.api.exceptions.SystemException;
+import pro.taskana.task.internal.jobs.TaskCleanupJob;
+import pro.taskana.task.internal.jobs.TaskUpdatePriorityJob;
+import pro.taskana.workbasket.internal.jobs.WorkbasketCleanupJob;
+
+/**
+ * Schedules the {@linkplain JobRunner} based on given {@linkplain Clock} whith given {@linkplain
+ * TaskanaEngine}.
+ *
+ * <p>For running the jobs the {@linkplain PlainJavaTransactionProvider} is used.
+ */
+public class JobScheduler {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(JobScheduler.class);
+  private final TaskanaEngine taskanaEngine;
+
+  private final Clock clock;
+
+  private final PlainJavaTransactionProvider plainJavaTransactionProvider;
+
+  public JobScheduler(TaskanaEngine taskanaEngine, Clock clock) {
+    this.taskanaEngine = taskanaEngine;
+    this.clock = clock;
+    this.plainJavaTransactionProvider =
+        new PlainJavaTransactionProvider(
+            taskanaEngine, taskanaEngine.getConfiguration().getDatasource());
+    plainJavaTransactionProvider.executeInTransaction(
+        () -> {
+          if (taskanaEngine.getConfiguration().isJobSchedulerEnableTaskCleanupJob()) {
+            AbstractTaskanaJob.initializeSchedule(taskanaEngine, TaskCleanupJob.class);
+            LOGGER.info("Job '{}' enabled", TaskCleanupJob.class.getName());
+          }
+          if (taskanaEngine.getConfiguration().isJobSchedulerEnableTaskUpdatePriorityJob()) {
+            AbstractTaskanaJob.initializeSchedule(taskanaEngine, TaskUpdatePriorityJob.class);
+            LOGGER.info("Job '{}' enabled", TaskUpdatePriorityJob.class.getName());
+          }
+          if (taskanaEngine.getConfiguration().isJobSchedulerEnableWorkbasketCleanupJob()) {
+            AbstractTaskanaJob.initializeSchedule(taskanaEngine, WorkbasketCleanupJob.class);
+            LOGGER.info("Job '{}' enabled", WorkbasketCleanupJob.class.getName());
+          }
+          if (taskanaEngine.getConfiguration().isJobSchedulerEnableUserInfoRefreshJob()) {
+            initJobByClassName("pro.taskana.user.jobs.UserInfoRefreshJob");
+          }
+          if (taskanaEngine.getConfiguration().isJobSchedulerEnableHistorieCleanupJob()) {
+            initJobByClassName("pro.taskana.simplehistory.impl.jobs.HistoryCleanupJob");
+          }
+          taskanaEngine
+              .getConfiguration()
+              .getJobSchedulerCustomJobs()
+              .forEach(this::initJobByClassName);
+
+          return "Initialized Jobs successfully";
+        });
+    this.clock.register(this::runAsyncJobsAsAdmin);
+  }
+
+  public void start() {
+    clock.start();
+  }
+
+  public void stop() {
+    clock.stop();
+  }
+
+  private void initJobByClassName(String className) throws SystemException {
+    try {
+      Class<?> jobClass = Thread.currentThread().getContextClassLoader().loadClass(className);
+      AbstractTaskanaJob.initializeSchedule(taskanaEngine, jobClass);
+      LOGGER.info("Job '{}' enabled", className);
+    } catch (ClassNotFoundException e) {
+      throw new SystemException(String.format("Could not find class '%s'", className), e);
+    }
+  }
+
+  private void runAsyncJobsAsAdmin() {
+    taskanaEngine.runAsAdmin(
+        () -> {
+          JobRunner runner = new JobRunner(taskanaEngine);
+          runner.registerTransactionProvider(plainJavaTransactionProvider);
+          LOGGER.info("Running Jobs");
+          runner.runJobs();
+          return "Successful";
+        });
+  }
+}

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

@@ -8,6 +8,7 @@ import javax.sql.DataSource;
 import pro.taskana.common.api.TaskanaEngine;
 import pro.taskana.common.api.TaskanaEngine.ConnectionManagementMode;
 import pro.taskana.common.api.exceptions.SystemException;
+import pro.taskana.common.internal.TaskanaEngineImpl;
 import pro.taskana.common.internal.transaction.TaskanaTransactionProvider;
 
 public class PlainJavaTransactionProvider implements TaskanaTransactionProvider {
@@ -24,15 +25,18 @@ public class PlainJavaTransactionProvider implements TaskanaTransactionProvider
 
   @Override
   public <T> T executeInTransaction(Supplier<T> supplier) {
+    if (((TaskanaEngineImpl) taskanaEngine).getConnection() != null) {
+      return supplier.get();
+    }
     try (Connection connection = dataSource.getConnection()) {
       taskanaEngine.setConnection(connection);
       final T t = supplier.get();
       connection.commit();
+      taskanaEngine.closeConnection();
       return t;
     } catch (SQLException ex) {
       throw new SystemException("caught exception", ex);
     } finally {
-      taskanaEngine.closeConnection();
       taskanaEngine.setConnectionManagementMode(defaultConnectionManagementMode);
     }
   }

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

@@ -0,0 +1,44 @@
+package pro.taskana.common.internal.jobs;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public class RealClock implements Clock {
+
+  private final long initialStartDelay;
+  private final long period;
+  private final TimeUnit periodTimeUnit;
+  private final List<ClockListener> listeners = Collections.synchronizedList(new ArrayList<>());
+  private final ScheduledExecutorService timerService =
+      Executors.newSingleThreadScheduledExecutor();
+
+  public RealClock(long initialStartDelay, long period, TimeUnit periodTimeUnit) {
+    this.initialStartDelay = initialStartDelay;
+    this.period = period;
+    this.periodTimeUnit = periodTimeUnit;
+  }
+
+  @Override
+  public void register(ClockListener listener) {
+    listeners.add(listener);
+  }
+
+  @Override
+  public void start() {
+    timerService.scheduleAtFixedRate(
+        this::reportTimeElapse, initialStartDelay, period, periodTimeUnit);
+  }
+
+  @Override
+  public void stop() {
+    timerService.shutdown();
+  }
+
+  private void reportTimeElapse() {
+    listeners.forEach(ClockListener::timeElapsed);
+  }
+}

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

@@ -21,7 +21,6 @@ import pro.taskana.common.api.exceptions.InvalidArgumentException;
 import pro.taskana.common.api.exceptions.MismatchedRoleException;
 import pro.taskana.common.api.exceptions.SystemException;
 import pro.taskana.common.api.exceptions.TaskanaException;
-import pro.taskana.common.internal.JobServiceImpl;
 import pro.taskana.common.internal.jobs.AbstractTaskanaJob;
 import pro.taskana.common.internal.transaction.TaskanaTransactionProvider;
 import pro.taskana.common.internal.util.CollectionUtil;
@@ -66,19 +65,6 @@ public class TaskCleanupJob extends AbstractTaskanaJob {
     }
   }
 
-  /**
-   * Initializes the TaskCleanupJob schedule. <br>
-   * All scheduled cleanup jobs are cancelled/deleted and a new one is scheduled.
-   *
-   * @param taskanaEngine the TASKANA engine.
-   */
-  public static void initializeSchedule(TaskanaEngine taskanaEngine) {
-    JobServiceImpl jobService = (JobServiceImpl) taskanaEngine.getJobService();
-    TaskCleanupJob job = new TaskCleanupJob(taskanaEngine, null, null);
-    jobService.deleteJobs(job.getType());
-    job.scheduleNextJob();
-  }
-
   @Override
   protected String getType() {
     return TaskCleanupJob.class.getName();

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

@@ -8,7 +8,6 @@ import org.slf4j.LoggerFactory;
 import pro.taskana.common.api.ScheduledJob;
 import pro.taskana.common.api.TaskanaEngine;
 import pro.taskana.common.api.exceptions.SystemException;
-import pro.taskana.common.internal.JobServiceImpl;
 import pro.taskana.common.internal.jobs.AbstractTaskanaJob;
 import pro.taskana.common.internal.transaction.TaskanaTransactionProvider;
 import pro.taskana.task.internal.jobs.helper.TaskUpdatePriorityWorker;
@@ -61,19 +60,6 @@ public class TaskUpdatePriorityJob extends AbstractTaskanaJob {
     return batchSize;
   }
 
-  /**
-   * Initializes the TaskUpdatePriorityJob schedule. <br>
-   * All scheduled jobs are cancelled/deleted and a new one is scheduled.
-   *
-   * @param taskanaEngine the TASKANA engine.
-   */
-  public static void initializeSchedule(TaskanaEngine taskanaEngine) {
-    JobServiceImpl jobService = (JobServiceImpl) taskanaEngine.getJobService();
-    TaskUpdatePriorityJob job = new TaskUpdatePriorityJob(taskanaEngine);
-    jobService.deleteJobs(job.getType());
-    job.scheduleNextJob();
-  }
-
   @Override
   protected String getType() {
     return TaskUpdatePriorityJob.class.getName();

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

@@ -12,7 +12,6 @@ import pro.taskana.common.api.exceptions.InvalidArgumentException;
 import pro.taskana.common.api.exceptions.MismatchedRoleException;
 import pro.taskana.common.api.exceptions.SystemException;
 import pro.taskana.common.api.exceptions.TaskanaException;
-import pro.taskana.common.internal.JobServiceImpl;
 import pro.taskana.common.internal.jobs.AbstractTaskanaJob;
 import pro.taskana.common.internal.transaction.TaskanaTransactionProvider;
 import pro.taskana.common.internal.util.CollectionUtil;
@@ -50,19 +49,6 @@ public class WorkbasketCleanupJob extends AbstractTaskanaJob {
     }
   }
 
-  /**
-   * Initializes the WorkbasketCleanupJob schedule. <br>
-   * All scheduled cleanup jobs are cancelled/deleted and a new one is scheduled.
-   *
-   * @param taskanaEngine the taskana engine
-   */
-  public static void initializeSchedule(TaskanaEngine taskanaEngine) {
-    JobServiceImpl jobService = (JobServiceImpl) taskanaEngine.getJobService();
-    WorkbasketCleanupJob job = new WorkbasketCleanupJob(taskanaEngine, null, null);
-    jobService.deleteJobs(job.getType());
-    job.scheduleNextJob();
-  }
-
   @Override
   protected String getType() {
     return WorkbasketCleanupJob.class.getName();

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

@@ -10,6 +10,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.junit.jupiter.api.BeforeEach;
@@ -26,6 +27,7 @@ import pro.taskana.common.api.TaskanaEngine;
 import pro.taskana.common.api.TaskanaEngine.ConnectionManagementMode;
 import pro.taskana.common.internal.JobMapper;
 import pro.taskana.common.internal.JobServiceImpl;
+import pro.taskana.common.internal.jobs.AbstractTaskanaJob;
 import pro.taskana.common.internal.jobs.JobRunner;
 import pro.taskana.common.test.security.JaasExtension;
 import pro.taskana.common.test.security.WithAccessId;
@@ -144,7 +146,7 @@ class TaskCleanupJobAccTest extends AbstractAccTest {
             .filter(scheduledJob -> scheduledJob.getType().equals(TaskCleanupJob.class.getName()))
             .collect(Collectors.toList());
 
-    TaskCleanupJob.initializeSchedule(taskanaEngine);
+    AbstractTaskanaJob.initializeSchedule(taskanaEngine, TaskCleanupJob.class);
 
     jobsToRun = getJobMapper(taskanaEngine).findJobsToRun(Instant.now());
 
@@ -223,12 +225,16 @@ class TaskCleanupJobAccTest extends AbstractAccTest {
         new TaskanaConfiguration.Builder(AbstractAccTest.taskanaEngineConfiguration)
             .cleanupJobRunEvery(runEvery)
             .cleanupJobFirstRun(firstRun)
+            .jobSchedulerEnabled(true)
+            .jobSchedulerInitialStartDelay(0)
+            .jobSchedulerPeriod(1)
+            .jobSchedulerPeriodTimeUnit(TimeUnit.SECONDS)
+            .jobSchedulerEnableTaskCleanupJob(true)
             .build();
 
     TaskanaEngine taskanaEngine =
         TaskanaEngine.buildTaskanaEngine(
             taskanaEngineConfiguration, ConnectionManagementMode.AUTOCOMMIT);
-    TaskCleanupJob.initializeSchedule(taskanaEngine);
 
     List<ScheduledJob> nextJobs =
         getJobMapper(taskanaEngine).findJobsToRun(Instant.now().plus(runEvery));
@@ -246,7 +252,7 @@ class TaskCleanupJobAccTest extends AbstractAccTest {
             .build();
 
     TaskanaEngine taskanaEngine = TaskanaEngine.buildTaskanaEngine(taskanaEngineConfiguration);
-    TaskCleanupJob.initializeSchedule(taskanaEngine);
+    AbstractTaskanaJob.initializeSchedule(taskanaEngine, TaskCleanupJob.class);
 
     List<ScheduledJob> nextJobs = getJobMapper(taskanaEngine).findJobsToRun(Instant.now());
     assertThat(nextJobs).isEmpty();

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

@@ -19,6 +19,7 @@ import pro.taskana.common.api.ScheduledJob;
 import pro.taskana.common.api.TaskanaEngine;
 import pro.taskana.common.api.TaskanaEngine.ConnectionManagementMode;
 import pro.taskana.common.api.exceptions.SystemException;
+import pro.taskana.common.internal.jobs.AbstractTaskanaJob;
 import pro.taskana.common.test.security.JaasExtension;
 import pro.taskana.common.test.security.WithAccessId;
 import pro.taskana.task.api.TaskQueryColumnName;
@@ -109,11 +110,11 @@ class TaskUpdatePriorityJobAccTest extends AbstractAccTest {
         TaskanaEngine.buildTaskanaEngine(
             taskanaEngineConfiguration, ConnectionManagementMode.AUTOCOMMIT);
     // when
-    TaskUpdatePriorityJob.initializeSchedule(taskanaEngine);
+    AbstractTaskanaJob.initializeSchedule(taskanaEngine, TaskUpdatePriorityJob.class);
 
     // then
     assertThat(getJobMapper(taskanaEngine).findJobsToRun(someTimeInTheFuture))
-        .hasSizeGreaterThanOrEqualTo(1)
+        .isNotEmpty()
         .extracting(ScheduledJob::getType)
         .contains(TaskUpdatePriorityJob.class.getName());
   }

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

@@ -13,6 +13,7 @@ import org.junit.jupiter.api.extension.ExtendWith;
 import pro.taskana.classification.internal.jobs.ClassificationChangedJob;
 import pro.taskana.common.api.BaseQuery;
 import pro.taskana.common.api.ScheduledJob;
+import pro.taskana.common.internal.jobs.AbstractTaskanaJob;
 import pro.taskana.common.test.security.JaasExtension;
 import pro.taskana.common.test.security.WithAccessId;
 import pro.taskana.task.api.TaskState;
@@ -112,7 +113,7 @@ class WorkbasketCleanupJobAccTest extends AbstractAccTest {
                 scheduledJob -> scheduledJob.getType().equals(WorkbasketCleanupJob.class.getName()))
             .collect(Collectors.toList());
 
-    WorkbasketCleanupJob.initializeSchedule(taskanaEngine);
+    AbstractTaskanaJob.initializeSchedule(taskanaEngine, WorkbasketCleanupJob.class);
 
     jobsToRun = getJobMapper(taskanaEngine).findJobsToRun(Instant.now());
 

lib/taskana-core/src/test/java/acceptance/jobs/helper/SqlConnectionRunnerAccTest.java

@@ -28,12 +28,10 @@ class SqlConnectionRunnerAccTest extends AbstractAccTest {
     // when
     runner.runWithConnection(
         connection -> {
-          ResultSet resultSet;
           PreparedStatement preparedStatement =
               connection.prepareStatement("select * from TASK where ID = ?");
           preparedStatement.setString(1, taskId);
-          resultSet = preparedStatement.executeQuery();
+          ResultSet resultSet = preparedStatement.executeQuery();
-          // then
           assertThat(resultSet.next()).isTrue();
         });
   }

lib/taskana-core/src/test/resources/taskana.properties

@@ -24,3 +24,18 @@ taskana.workingtime.schedule.TUESDAY=00:00-00:00
 taskana.workingtime.schedule.WEDNESDAY=00:00-00:00
 taskana.workingtime.schedule.THURSDAY=00:00-00:00
 taskana.workingtime.schedule.FRIDAY=00:00-00:00
+# enable or disable the jobscheduler at all
+# set it to false and no jobs are running
+taskana.jobscheduler.enabled=false
+# wait time before the first job run in millis
+taskana.jobscheduler.initialstartdelay=100000
+# sleeping time befor the next job runs
+taskana.jobscheduler.period=12
+# timeunit for the sleeping period
+# Possible values: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+taskana.jobscheduler.periodtimeunit=HOURS
+taskana.jobscheduler.enableTaskCleanupJob=false
+taskana.jobscheduler.enableTaskUpdatePriorityJob=false
+taskana.jobscheduler.enableWorkbasketCleanupJob=false
+taskana.jobscheduler.enableUserInfoRefreshJob=false
+taskana.jobscheduler.enableHistorieCleanupJob=false

lib/taskana-spring-example/src/test/resources/taskana.properties

@@ -0,0 +1,35 @@
+taskana.roles.user=cn=ksc-users,cn=groups,OU=Test,O=TASKANA | teamlead-1 | teamlead-2 | user-1-1 | user-1-2 | user-2-1 | user-2-2 | user-b-1 | user-b-2
+taskana.roles.admin=admin | uid=admin,cn=users,OU=Test,O=TASKANA
+taskana.roles.businessadmin=businessadmin | cn=business-admins,cn=groups,OU=Test,O=TASKANA
+taskana.roles.monitor=monitor | cn=monitor-users,cn=groups,OU=Test,O=TASKANA
+taskana.roles.taskadmin=taskadmin
+taskana.domains=DOMAIN_A  ,  DOMAIN_B
+taskana.user.minimalPermissionsToAssignDomains=READ | OPEN
+taskana.classification.types=TASK , document
+taskana.classification.categories.task=EXTERNAL, manual, autoMAtic, Process
+taskana.classification.categories.document=EXTERNAL
+taskana.jobs.maxRetries=3
+taskana.jobs.batchSize=50
+taskana.jobs.cleanup.runEvery=P1D
+taskana.jobs.cleanup.firstRunAt=2018-07-25T08:00:00Z
+taskana.jobs.cleanup.minimumAge=P0D
+taskana.german.holidays.enabled=true
+taskana.german.holidays.corpus-christi.enabled=false
+taskana.history.deletion.on.task.deletion.enabled=true
+taskana.validation.allowTimestampServiceLevelMismatch=false
+taskana.query.includeLongName=false
+# enable or disable the jobscheduler at all
+# set it to false and no jobs are running
+taskana.jobscheduler.enabled=false
+# wait time before the first job run in millis
+taskana.jobscheduler.initialstartdelay=100000
+# sleeping time befor the next job runs
+taskana.jobscheduler.period=12
+# timeunit for the sleeping period
+# Possible values: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+taskana.jobscheduler.periodtimeunit=HOURS
+taskana.jobscheduler.enableTaskCleanupJob=false
+taskana.jobscheduler.enableTaskUpdatePriorityJob=false
+taskana.jobscheduler.enableWorkbasketCleanupJob=false
+taskana.jobscheduler.enableUserInfoRefreshJob=false
+taskana.jobscheduler.enableHistorieCleanupJob=false

lib/taskana-test-api/src/main/java/pro/taskana/testapi/extensions/TaskanaInitializationExtension.java

@@ -4,10 +4,12 @@ import static org.junit.platform.commons.support.AnnotationSupport.isAnnotated;
 import static pro.taskana.testapi.util.ExtensionCommunicator.getClassLevelStore;
 import static pro.taskana.testapi.util.ExtensionCommunicator.isTopLevelClass;
 
+import java.lang.reflect.Field;
 import java.util.HashMap;
 import java.util.Map;
 import javax.sql.DataSource;
 import org.apache.ibatis.session.SqlSession;
+import org.apache.ibatis.session.SqlSessionManager;
 import org.junit.jupiter.api.extension.ExtensionContext;
 import org.junit.jupiter.api.extension.ExtensionContext.Store;
 import org.junit.jupiter.api.extension.TestInstancePostProcessor;
@@ -27,6 +29,7 @@ import pro.taskana.common.api.security.CurrentUserContext;
 import pro.taskana.common.internal.ConfigurationMapper;
 import pro.taskana.common.internal.ConfigurationServiceImpl;
 import pro.taskana.common.internal.InternalTaskanaEngine;
+import pro.taskana.common.internal.JobMapper;
 import pro.taskana.common.internal.JobServiceImpl;
 import pro.taskana.common.internal.TaskanaEngineImpl;
 import pro.taskana.common.internal.security.CurrentUserContextImpl;
@@ -123,6 +126,7 @@ public class TaskanaInitializationExtension implements TestInstancePostProcessor
     UserService userService = taskanaEngine.getUserService();
     SqlSession sqlSession = taskanaEngineProxy.getSqlSession();
     WorkingTimeCalculator workingTimeCalculator = taskanaEngine.getWorkingTimeCalculator();
+    JobMapper jobMapper = getJobMapper(taskanaEngine);
     return Map.ofEntries(
         Map.entry(TaskanaConfiguration.class, taskanaEngine.getConfiguration()),
         Map.entry(TaskanaEngineImpl.class, taskanaEngine),
@@ -146,6 +150,18 @@ public class TaskanaInitializationExtension implements TestInstancePostProcessor
         Map.entry(WorkingTimeCalculatorImpl.class, workingTimeCalculator),
         Map.entry(ConfigurationMapper.class, sqlSession.getMapper(ConfigurationMapper.class)),
         Map.entry(UserService.class, userService),
-        Map.entry(UserServiceImpl.class, userService));
+        Map.entry(UserServiceImpl.class, userService),
+        Map.entry(JobMapper.class, jobMapper));
+  }
+
+  private static JobMapper getJobMapper(TaskanaEngine taskanaEngine)
+      throws NoSuchFieldException, IllegalAccessException {
+
+    Field sessionManagerField = TaskanaEngineImpl.class.getDeclaredField("sessionManager");
+    sessionManagerField.setAccessible(true);
+    SqlSessionManager sqlSessionManager =
+        (SqlSessionManager) sessionManagerField.get(taskanaEngine);
+
+    return sqlSessionManager.getMapper(JobMapper.class);
   }
 }

lib/taskana-test-api/src/main/java/pro/taskana/testapi/extensions/TestContainerExtension.java

@@ -89,12 +89,7 @@ public class TestContainerExtension implements InvocationInterceptor {
     return ds;
   }
 
-  private static void copyValue(String key, Store source, Store destination) {
+  public static String determineSchemaName() {
-    Object value = source.get(key);
-    destination.put(key, value);
-  }
-
-  private static String determineSchemaName() {
     String uniqueId = "A" + UUID.randomUUID().toString().replace("-", "_");
     if (EXECUTION_DATABASE == DB.ORACLE) {
       uniqueId = uniqueId.substring(0, 26);
@@ -104,6 +99,11 @@ public class TestContainerExtension implements InvocationInterceptor {
     return uniqueId;
   }
 
+  private static void copyValue(String key, Store source, Store destination) {
+    Object value = source.get(key);
+    destination.put(key, value);
+  }
+
   private static DB retrieveDatabaseFromEnv() {
     String property = System.getenv("DB");
     DB db;

lib/taskana-test-api/src/test/resources/taskana.properties

@@ -18,4 +18,18 @@ taskana.german.holidays.corpus-christi.enabled=false
 taskana.history.deletion.on.task.deletion.enabled=true
 taskana.validation.allowTimestampServiceLevelMismatch=false
 taskana.query.includeLongName=false
-
+# enable or disable the jobscheduler at all
+# set it to false and no jobs are running
+taskana.jobscheduler.enabled=false
+# wait time before the first job run in millis
+taskana.jobscheduler.initialstartdelay=100000
+# sleeping time befor the next job runs
+taskana.jobscheduler.period=12
+# timeunit for the sleeping period
+# Possible values: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+taskana.jobscheduler.periodtimeunit=HOURS
+taskana.jobscheduler.enableTaskCleanupJob=false
+taskana.jobscheduler.enableTaskUpdatePriorityJob=false
+taskana.jobscheduler.enableWorkbasketCleanupJob=false
+taskana.jobscheduler.enableUserInfoRefreshJob=false
+taskana.jobscheduler.enableHistorieCleanupJob=false

rest/taskana-rest-spring-example-boot/src/main/resources/application.properties

@@ -35,8 +35,6 @@ devMode=false
 enableCsrf=true
 ####### property that control if the database is cleaned and sample data is generated
 generateSampleData=true
-####### JobScheduler cron expression that specifies when the JobSchedler runs
-taskana.jobscheduler.async.cron=0 * * * * *
 ####### cache static resources properties
 spring.web.resources.cache.cachecontrol.cache-private=true
 ####### for upload of big workbasket- or classification-files

rest/taskana-rest-spring-example-boot/src/main/resources/taskana.properties

@@ -23,3 +23,18 @@ taskana.german.holidays.enabled=true
 taskana.german.holidays.corpus-christi.enabled=true
 taskana.historylogger.name=AUDIT
 taskana.routing.dmn=/dmn-table.dmn
+# enable or disable the jobscheduler at all
+# set it to false and no jobs are running
+taskana.jobscheduler.enabled=false
+# wait time before the first job run in millis
+taskana.jobscheduler.initialstartdelay=100
+# sleeping time befor the next job runs
+taskana.jobscheduler.period=12
+# timeunit for the sleeping period
+# Possible values: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+taskana.jobscheduler.periodtimeunit=HOURS
+taskana.jobscheduler.enableTaskCleanupJob=true
+taskana.jobscheduler.enableTaskUpdatePriorityJob=true
+taskana.jobscheduler.enableWorkbasketCleanupJob=true
+taskana.jobscheduler.enableUserInfoRefreshJob=true
+taskana.jobscheduler.enableHistorieCleanupJob=false

rest/taskana-rest-spring-example-common/src/main/java/pro/taskana/example/jobs/JobScheduler.java

@@ -1,67 +0,0 @@
-package pro.taskana.example.jobs;
-
-import java.lang.reflect.InvocationTargetException;
-import javax.annotation.PostConstruct;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.scheduling.annotation.Scheduled;
-import org.springframework.stereotype.Component;
-
-import pro.taskana.common.api.TaskanaEngine;
-import pro.taskana.common.internal.jobs.JobRunner;
-import pro.taskana.common.internal.transaction.TaskanaTransactionProvider;
-import pro.taskana.task.internal.jobs.TaskCleanupJob;
-import pro.taskana.user.jobs.UserInfoRefreshJob;
-import pro.taskana.workbasket.internal.jobs.WorkbasketCleanupJob;
-
-/** This class invokes the JobRunner periodically to schedule long running jobs. */
-@Component
-public class JobScheduler {
-
-  private static final Logger LOGGER = LoggerFactory.getLogger(JobScheduler.class);
-  private final TaskanaTransactionProvider springTransactionProvider;
-  private final TaskanaEngine taskanaEngine;
-
-  @Autowired
-  public JobScheduler(
-      TaskanaTransactionProvider springTransactionProvider, TaskanaEngine taskanaEngine) {
-    this.springTransactionProvider = springTransactionProvider;
-    this.taskanaEngine = taskanaEngine;
-  }
-
-  @PostConstruct
-  public void scheduleCleanupJob()
-      throws NoSuchMethodException, InvocationTargetException, IllegalAccessException,
-          ClassNotFoundException {
-    TaskCleanupJob.initializeSchedule(taskanaEngine);
-    WorkbasketCleanupJob.initializeSchedule(taskanaEngine);
-    UserInfoRefreshJob.initializeSchedule(taskanaEngine);
-
-    if (taskanaEngine.isHistoryEnabled()) {
-      Thread.currentThread()
-          .getContextClassLoader()
-          .loadClass("pro.taskana.simplehistory.impl.jobs.HistoryCleanupJob")
-          .getDeclaredMethod("initializeSchedule", TaskanaEngine.class)
-          .invoke(null, taskanaEngine);
-    }
-  }
-
-  @Scheduled(cron = "${taskana.jobscheduler.async.cron}")
-  public void triggerJobs() {
-    LOGGER.info("AsyncJobs started.");
-    runAsyncJobsAsAdmin();
-    LOGGER.info("AsyncJobs completed.");
-  }
-
-  private void runAsyncJobsAsAdmin() {
-    taskanaEngine.runAsAdmin(
-        () -> {
-          JobRunner runner = new JobRunner(taskanaEngine);
-          runner.registerTransactionProvider(springTransactionProvider);
-          LOGGER.info("Running Jobs");
-          runner.runJobs();
-          return "Successful";
-        });
-  }
-}

rest/taskana-rest-spring-example-common/src/test/java/pro/taskana/example/jobs/AsyncUpdateJobIntTest.java

@@ -1,163 +0,0 @@
-package pro.taskana.example.jobs;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static pro.taskana.rest.test.RestHelper.TEMPLATE;
-
-import java.time.Instant;
-import java.util.List;
-import org.junit.jupiter.api.Test;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.core.ParameterizedTypeReference;
-import org.springframework.hateoas.IanaLinkRelations;
-import org.springframework.http.HttpEntity;
-import org.springframework.http.HttpMethod;
-import org.springframework.http.ResponseEntity;
-
-import pro.taskana.classification.api.models.Classification;
-import pro.taskana.classification.rest.assembler.ClassificationRepresentationModelAssembler;
-import pro.taskana.classification.rest.models.ClassificationRepresentationModel;
-import pro.taskana.common.api.exceptions.InvalidArgumentException;
-import pro.taskana.common.rest.RestEndpoints;
-import pro.taskana.rest.test.RestHelper;
-import pro.taskana.rest.test.TaskanaSpringBootTest;
-import pro.taskana.task.api.models.Task;
-import pro.taskana.task.rest.assembler.TaskRepresentationModelAssembler;
-import pro.taskana.task.rest.models.TaskRepresentationModel;
-
-/** Test async updates. */
-@TaskanaSpringBootTest
-class AsyncUpdateJobIntTest {
-
-  private static final String CLASSIFICATION_ID = "CLI:100000000000000000000000000000000003";
-
-  private final ClassificationRepresentationModelAssembler
-      classificationRepresentationModelAssembler;
-  private final TaskRepresentationModelAssembler taskRepresentationModelAssembler;
-  private final JobScheduler jobScheduler;
-  private final RestHelper restHelper;
-
-  @Autowired
-  AsyncUpdateJobIntTest(
-      ClassificationRepresentationModelAssembler classificationRepresentationModelAssembler,
-      TaskRepresentationModelAssembler taskRepresentationModelAssembler,
-      JobScheduler jobScheduler,
-      RestHelper restHelper) {
-    this.classificationRepresentationModelAssembler = classificationRepresentationModelAssembler;
-    this.taskRepresentationModelAssembler = taskRepresentationModelAssembler;
-    this.jobScheduler = jobScheduler;
-    this.restHelper = restHelper;
-  }
-
-  @Test
-  void testUpdateClassificationPrioServiceLevel() throws InvalidArgumentException {
-
-    // 1st step: get old classification :
-    final Instant before = Instant.now();
-
-    ResponseEntity<ClassificationRepresentationModel> response =
-        TEMPLATE.exchange(
-            restHelper.toUrl(RestEndpoints.URL_CLASSIFICATIONS_ID, CLASSIFICATION_ID),
-            HttpMethod.GET,
-            new HttpEntity<>(RestHelper.generateHeadersForUser("teamlead-1")),
-            ParameterizedTypeReference.forType(ClassificationRepresentationModel.class));
-
-    assertThat(response.getBody()).isNotNull();
-    ClassificationRepresentationModel classification = response.getBody();
-    assertThat(classification.getLink(IanaLinkRelations.SELF)).isNotNull();
-
-    // 2nd step: modify classification and trigger update
-    classification.removeLinks();
-    classification.setServiceLevel("P5D");
-    classification.setPriority(1000);
-
-    TEMPLATE.exchange(
-        restHelper.toUrl(RestEndpoints.URL_CLASSIFICATIONS_ID, CLASSIFICATION_ID),
-        HttpMethod.PUT,
-        new HttpEntity<>(classification, RestHelper.generateHeadersForUser("teamlead-1")),
-        ParameterizedTypeReference.forType(ClassificationRepresentationModel.class));
-
-    // trigger jobs twice to refresh all entries. first entry on the first call and follow up on the
-    // seconds call
-    jobScheduler.triggerJobs();
-    jobScheduler.triggerJobs();
-
-    // verify the classification modified timestamp is after 'before'
-    ResponseEntity<ClassificationRepresentationModel> repeatedResponse =
-        TEMPLATE.exchange(
-            restHelper.toUrl(RestEndpoints.URL_CLASSIFICATIONS_ID, CLASSIFICATION_ID),
-            HttpMethod.GET,
-            new HttpEntity<>(RestHelper.generateHeadersForUser("teamlead-1")),
-            ParameterizedTypeReference.forType(ClassificationRepresentationModel.class));
-
-    assertThat(repeatedResponse.getBody()).isNotNull();
-
-    ClassificationRepresentationModel modifiedClassificationRepresentationModel =
-        repeatedResponse.getBody();
-    Classification modifiedClassification =
-        classificationRepresentationModelAssembler.toEntityModel(
-            modifiedClassificationRepresentationModel);
-
-    assertThat(before).isBefore(modifiedClassification.getModified());
-
-    List<String> affectedTasks =
-        List.of(
-            "TKI:000000000000000000000000000000000003",
-            "TKI:000000000000000000000000000000000004",
-            "TKI:000000000000000000000000000000000005",
-            "TKI:000000000000000000000000000000000006",
-            "TKI:000000000000000000000000000000000007",
-            "TKI:000000000000000000000000000000000008",
-            "TKI:000000000000000000000000000000000009",
-            "TKI:000000000000000000000000000000000010",
-            "TKI:000000000000000000000000000000000011",
-            "TKI:000000000000000000000000000000000012",
-            "TKI:000000000000000000000000000000000013",
-            "TKI:000000000000000000000000000000000014",
-            "TKI:000000000000000000000000000000000015",
-            "TKI:000000000000000000000000000000000016",
-            "TKI:000000000000000000000000000000000017",
-            "TKI:000000000000000000000000000000000018",
-            "TKI:000000000000000000000000000000000019",
-            "TKI:000000000000000000000000000000000020",
-            "TKI:000000000000000000000000000000000021",
-            "TKI:000000000000000000000000000000000022",
-            "TKI:000000000000000000000000000000000023",
-            "TKI:000000000000000000000000000000000024",
-            "TKI:000000000000000000000000000000000025",
-            "TKI:000000000000000000000000000000000026",
-            "TKI:000000000000000000000000000000000027",
-            "TKI:000000000000000000000000000000000028",
-            "TKI:000000000000000000000000000000000029",
-            "TKI:000000000000000000000000000000000030",
-            "TKI:000000000000000000000000000000000031",
-            "TKI:000000000000000000000000000000000032",
-            "TKI:000000000000000000000000000000000033",
-            "TKI:000000000000000000000000000000000034",
-            "TKI:000000000000000000000000000000000035",
-            "TKI:000000000000000000000000000000000100",
-            "TKI:000000000000000000000000000000000101",
-            "TKI:000000000000000000000000000000000102",
-            "TKI:000000000000000000000000000000000103");
-    for (String taskId : affectedTasks) {
-      verifyTaskIsModifiedAfterOrEquals(taskId, before);
-    }
-  }
-
-  private void verifyTaskIsModifiedAfterOrEquals(String taskId, Instant before)
-      throws InvalidArgumentException {
-
-    ResponseEntity<TaskRepresentationModel> taskResponse =
-        TEMPLATE.exchange(
-            restHelper.toUrl(RestEndpoints.URL_TASKS_ID, taskId),
-            HttpMethod.GET,
-            new HttpEntity<>(RestHelper.generateHeadersForUser("admin")),
-            ParameterizedTypeReference.forType(TaskRepresentationModel.class));
-
-    TaskRepresentationModel taskRepresentationModel = taskResponse.getBody();
-    assertThat(taskRepresentationModel).isNotNull();
-    Task task = taskRepresentationModelAssembler.toEntityModel(taskRepresentationModel);
-
-    Instant modified = task.getModified();
-    assertThat(before).as("Task " + task.getId() + " has not been refreshed.").isBefore(modified);
-  }
-}

rest/taskana-rest-spring-example-common/src/test/resources/taskana.properties

@@ -14,3 +14,18 @@ taskana.jobs.cleanup.runEvery=P1D
 taskana.jobs.cleanup.firstRunAt=2018-07-25T08:00:00Z
 taskana.jobs.cleanup.minimumAge=P14D
 taskana.german.holidays.enabled=true
+# enable or disable the jobscheduler at all
+# set it to false and no jobs are running
+taskana.jobscheduler.enabled=false
+# wait time before the first job run in millis
+taskana.jobscheduler.initialstartdelay=100
+# sleeping time befor the next job runs
+taskana.jobscheduler.period=12
+# timeunit for the sleeping period
+# Possible values: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+taskana.jobscheduler.periodtimeunit=HOURS
+taskana.jobscheduler.enableTaskCleanupJob=true
+taskana.jobscheduler.enableTaskUpdatePriorityJob=true
+taskana.jobscheduler.enableWorkbasketCleanupJob=true
+taskana.jobscheduler.enableUserInfoRefreshJob=true
+taskana.jobscheduler.enableHistorieCleanupJob=false

rest/taskana-rest-spring-example-wildfly/src/test/resources/taskana.properties

@@ -19,3 +19,18 @@ taskana.jobs.history.cleanup.minimumAge=P14D
 taskana.jobs.history.cleanup.runEvery=P1D
 taskana.german.holidays.enabled=true
 taskana.historylogger.name=AUDIT
+# enable or disable the jobscheduler at all
+# set it to false and no jobs are running
+taskana.jobscheduler.enabled=true
+# wait time before the first job run in millis
+taskana.jobscheduler.initialstartdelay=100
+# sleeping time befor the next job runs
+taskana.jobscheduler.period=12
+# timeunit for the sleeping period
+# Possible values: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+taskana.jobscheduler.periodtimeunit=HOURS
+taskana.jobscheduler.enableTaskCleanupJob=true
+taskana.jobscheduler.enableTaskUpdatePriorityJob=true
+taskana.jobscheduler.enableWorkbasketCleanupJob=true
+taskana.jobscheduler.enableUserInfoRefreshJob=true
+taskana.jobscheduler.enableHistorieCleanupJob=true

rest/taskana-rest-spring/src/main/java/pro/taskana/user/jobs/UserInfoRefreshJob.java

@@ -11,7 +11,6 @@ import pro.taskana.common.api.TaskanaEngine;
 import pro.taskana.common.api.exceptions.InvalidArgumentException;
 import pro.taskana.common.api.exceptions.MismatchedRoleException;
 import pro.taskana.common.api.exceptions.SystemException;
-import pro.taskana.common.internal.JobServiceImpl;
 import pro.taskana.common.internal.jobs.AbstractTaskanaJob;
 import pro.taskana.common.internal.transaction.TaskanaTransactionProvider;
 import pro.taskana.common.rest.ldap.LdapClient;
@@ -44,19 +43,6 @@ public class UserInfoRefreshJob extends AbstractTaskanaJob {
     refreshUserPostprocessorManager = new RefreshUserPostprocessorManager();
   }
 
-  /**
-   * Initializes the {@linkplain UserInfoRefreshJob} schedule. <br>
-   * All scheduled jobs are cancelled/deleted and a new one is scheduled.
-   *
-   * @param taskanaEngine the TASKANA engine.
-   */
-  public static void initializeSchedule(TaskanaEngine taskanaEngine) {
-    JobServiceImpl jobService = (JobServiceImpl) taskanaEngine.getJobService();
-    UserInfoRefreshJob job = new UserInfoRefreshJob(taskanaEngine);
-    jobService.deleteJobs(job.getType());
-    job.scheduleNextJob();
-  }
-
   @Override
   protected String getType() {
     return UserInfoRefreshJob.class.getName();
@@ -74,7 +60,7 @@ public class UserInfoRefreshJob extends AbstractTaskanaJob {
       List<User> users = ldapClient.searchUsersInUserRole();
       List<User> usersAfterProcessing =
           users.stream()
-              .map(user -> refreshUserPostprocessorManager.processUserAfterRefresh(user))
+              .map(refreshUserPostprocessorManager::processUserAfterRefresh)
               .collect(Collectors.toList());
       addExistingConfigurationDataToUsers(usersAfterProcessing);
       clearExistingUsersAndGroups();

rest/taskana-rest-spring/src/test/resources/mytaskana.properties

@@ -16,4 +16,18 @@ taskana.jobs.cleanup.firstRunAt=2018-07-25T08:00:00Z
 taskana.jobs.cleanup.minimumAge=P14D
 taskana.german.holidays.enabled=true
 taskana.history.deletion.on.task.deletion.enabled=true
-
+# enable or disable the jobscheduler at all
+# set it to false and no jobs are running
+taskana.jobscheduler.enabled=false
+# wait time before the first job run in millis
+taskana.jobscheduler.initialstartdelay=100
+# sleeping time befor the next job runs
+taskana.jobscheduler.period=12
+# timeunit for the sleeping period
+# Possible values: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+taskana.jobscheduler.periodtimeunit=HOURS
+taskana.jobscheduler.enableTaskCleanupJob=true
+taskana.jobscheduler.enableTaskUpdatePriorityJob=true
+taskana.jobscheduler.enableWorkbasketCleanupJob=true
+taskana.jobscheduler.enableUserInfoRefreshJob=true
+taskana.jobscheduler.enableHistorieCleanupJob=false

rest/taskana-rest-spring/src/test/resources/taskana.properties

@@ -15,4 +15,19 @@ taskana.jobs.cleanup.firstRunAt=2018-07-25T08:00:00Z
 taskana.jobs.cleanup.minimumAge=P14D
 taskana.german.holidays.enabled=true
 taskana.history.deletion.on.task.deletion.enabled=true
+# enable or disable the jobscheduler at all
+# set it to false and no jobs are running
+taskana.jobscheduler.enabled=false
+# wait time before the first job run in millis
+taskana.jobscheduler.initialstartdelay=100
+# sleeping time befor the next job runs
+taskana.jobscheduler.period=12
+# timeunit for the sleeping period
+# Possible values: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+taskana.jobscheduler.periodtimeunit=HOURS
+taskana.jobscheduler.enableTaskCleanupJob=true
+taskana.jobscheduler.enableTaskUpdatePriorityJob=true
+taskana.jobscheduler.enableWorkbasketCleanupJob=true
+taskana.jobscheduler.enableUserInfoRefreshJob=true
+taskana.jobscheduler.enableHistorieCleanupJob=false
 

routing/taskana-routing-rest/src/test/resources/taskana.properties

@@ -15,4 +15,18 @@ taskana.jobs.cleanup.firstRunAt=2018-07-25T08:00:00Z
 taskana.jobs.cleanup.minimumAge=P14D
 taskana.german.holidays.enabled=true
 taskana.history.deletion.on.task.deletion.enabled=true
-
+# enable or disable the jobscheduler at all
+# set it to false and no jobs are running
+taskana.jobscheduler.enabled=false
+# wait time before the first job run in millis
+taskana.jobscheduler.initialstartdelay=100
+# sleeping time befor the next job runs
+taskana.jobscheduler.period=12
+# timeunit for the sleeping period
+# Possible values: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+taskana.jobscheduler.periodtimeunit=HOURS
+taskana.jobscheduler.enableTaskCleanupJob=true
+taskana.jobscheduler.enableTaskUpdatePriorityJob=true
+taskana.jobscheduler.enableWorkbasketCleanupJob=true
+taskana.jobscheduler.enableUserInfoRefreshJob=false
+taskana.jobscheduler.enableHistorieCleanupJob=false

routing/taskana-spi-routing-dmn-router/src/test/resources/taskana.properties

@@ -21,3 +21,18 @@ taskana.german.holidays.enabled=true
 taskana.german.holidays.corpus-christi.enabled=true
 taskana.historylogger.name=AUDIT
 taskana.routing.dmn=/dmn-table.dmn
+# enable or disable the jobscheduler at all
+# set it to false and no jobs are running
+taskana.jobscheduler.enabled=false
+# wait time before the first job run in millis
+taskana.jobscheduler.initialstartdelay=100
+# sleeping time befor the next job runs
+taskana.jobscheduler.period=12
+# timeunit for the sleeping period
+# Possible values: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+taskana.jobscheduler.periodtimeunit=HOURS
+taskana.jobscheduler.enableTaskCleanupJob=true
+taskana.jobscheduler.enableTaskUpdatePriorityJob=true
+taskana.jobscheduler.enableWorkbasketCleanupJob=true
+taskana.jobscheduler.enableUserInfoRefreshJob=false
+taskana.jobscheduler.enableHistorieCleanupJob=false