[#4] Allow job removal. Also implement clustering support for memory storage

Author: beikov

core/api/pom.xml

@@ -14,7 +14,7 @@
         <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>blaze-actor-core-api</artifactId>
-            <version>${version.blaze-notify-actor}</version>
+            <version>${version.blaze-actor}</version>
         </dependency>
         <!-- Test dependencies -->
 		<dependency>

core/api/src/main/java/com/blazebit/job/JobContext.java

@@ -30,6 +30,7 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -89,6 +90,21 @@ public interface JobContext extends ServiceProvider, ConfigurationSource {
      */
     <T extends JobInstance<?>> JobInstanceProcessor<?, T> getJobInstanceProcessor(T job);
 
+    /**
+     * Returns all partition keys.
+     *
+     * @return The list of all partition keys
+     */
+    Collection<PartitionKey> getPartitionKeys();
+
+    /**
+     * Returns the matching partition keys for the given job instance.
+     *
+     * @param jobInstance The job instance
+     * @return The list of matching partition keys
+     */
+    Collection<PartitionKey> getPartitionKeys(JobInstance<?> jobInstance);
+
     /**
      * Refreshes the job instance schedules for the given job instance.
      *
@@ -731,6 +747,7 @@ protected DefaultJobContext(TransactionSupport transactionSupport, JobManagerFac
                 Collection<PartitionKey> defaultTriggerPartitionKeys = this.partitionKeyProvider.getDefaultTriggerPartitionKeys();
                 if (partitionKeyEntries.isEmpty()) {
                     Collection<PartitionKey> instancePartitionKeys = this.partitionKeyProvider.getDefaultJobInstancePartitionKeys();
+
                     this.jobSchedulers = new HashMap<>(defaultTriggerPartitionKeys.size() + instancePartitionKeys.size());
                     for (PartitionKey instancePartitionKey : instancePartitionKeys) {
                         JobScheduler jobInstanceScheduler = jobSchedulerFactory.createJobScheduler(this, actorContext, DEFAULT_JOB_INSTANCE_ACTOR_NAME + "/" + instancePartitionKey, DEFAULT_JOB_INSTANCE_PROCESS_COUNT, instancePartitionKey);
@@ -818,7 +835,13 @@ public void refreshJobInstanceSchedules(JobInstance<?> jobInstance) {
                 }
             }
 
-            private List<PartitionKey> getPartitionKeys(JobInstance<?> jobInstance) {
+            @Override
+            public Collection<PartitionKey> getPartitionKeys() {
+                return Collections.unmodifiableSet(jobSchedulers.keySet());
+            }
+
+            @Override
+            public List<PartitionKey> getPartitionKeys(JobInstance<?> jobInstance) {
                 return jobInstanceClassToPartitionKeysMapping.computeIfAbsent(jobInstance.getClass(), (k) -> {
                     List<PartitionKey> v = new ArrayList<>(jobSchedulers.keySet().size());
                     for (PartitionKey partitionKey : jobSchedulers.keySet()) {

core/api/src/main/java/com/blazebit/job/JobInstanceState.java

@@ -43,6 +43,10 @@ public enum JobInstanceState {
     /**
      * A job instance that reached its maximum defer count and doesn't need further processing.
      */
-    DROPPED;
+    DROPPED,
+    /**
+     * A job instance with that state is going to be remove during an update.
+     */
+    REMOVED;
 
 }

core/api/src/main/java/com/blazebit/job/JobManager.java

@@ -18,6 +18,7 @@
 
 import java.time.Instant;
 import java.util.List;
+import java.util.Set;
 
 /**
  * A manager for adding, updating and querying job instances.
@@ -41,6 +42,23 @@ public interface JobManager {
      */
     void updateJobInstance(JobInstance<?> jobInstance);
 
+    /**
+     * Removes the given job instance.
+     *
+     * @param jobInstance The job instance to remove
+     */
+    void removeJobInstance(JobInstance<?> jobInstance);
+
+    /**
+     * Removes job instances that are in one of the given states and where the last execution time is lower than the given one.
+     *
+     * @param states The states for which job instances to remove
+     * @param executionTimeOlderThan Job instance with a last execution lower than this instant are removed. May be <code>null</code>
+     * @param partitionKey The partition key
+     * @return the number of removed job instances
+     */
+    int removeJobInstances(Set<JobInstanceState> states, Instant executionTimeOlderThan, PartitionKey partitionKey);
+
     /**
      * Returns a schedule time ordered list of job instances that need to be processed for the given partition.
      *

core/impl/pom.xml

@@ -19,12 +19,12 @@
         <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>blaze-actor-core-api</artifactId>
-            <version>${version.blaze-notify-actor}</version>
+            <version>${version.blaze-actor}</version>
         </dependency>
         <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>blaze-actor-core-impl</artifactId>
-            <version>${version.blaze-notify-actor}</version>
+            <version>${version.blaze-actor}</version>
             <scope>runtime</scope>
         </dependency>
         <!-- Test dependencies -->

jpa/model/src/main/java/com/blazebit/job/jpa/model/JpaPartitionKey.java

@@ -17,6 +17,7 @@
 package com.blazebit.job.jpa.model;
 
 import com.blazebit.job.JobInstance;
+import com.blazebit.job.JobInstanceState;
 import com.blazebit.job.PartitionKey;
 
 import java.util.ArrayList;
@@ -55,6 +56,13 @@ public interface JpaPartitionKey extends PartitionKey {
      */
     String getScheduleAttributeName();
 
+    /**
+     * Returns the attribute name of the last execution attribute of the entity type as given in {@link #getJobInstanceType()}.
+     *
+     * @return the attribute name of the last execution attribute
+     */
+    String getLastExecutionAttributeName();
+
     /**
      * Returns the attribute name of the partition key attribute of the entity type as given in {@link #getJobInstanceType()}.
      *
@@ -70,13 +78,21 @@ public interface JpaPartitionKey extends PartitionKey {
      */
     String getStatePredicate(String jobAlias);
 
+    /**
+     * Returns the expression for the state of a job.
+     *
+     * @param jobAlias The FROM clause alias for the job
+     * @return The state JPQL expression or an empty string
+     */
+    String getStateExpression(String jobAlias);
+
     /**
      * Returns the state value for ready jobs that must be bound in a query to the parameter name "readyState".
      * A <code>null</code> value means that no parameter should be bound.
      *
      * @return The ready state value of <code>null</code>
      */
-    Object getReadyStateValue();
+    Function<JobInstanceState, Object> getStateValueMappingFunction();
 
     /**
      * Returns the join fetches that should be applied to a query when fetching a job for this partition.
@@ -138,6 +154,14 @@ interface JpaPartitionKeyBuilder {
          */
         JpaPartitionKeyBuilder withScheduleAttributeName(String scheduleAttributeName);
 
+        /**
+         * Sets the given job last execution attribute name.
+         *
+         * @param lastExecutionAttributeName The job last execution attribute name
+         * @return this for chaining
+         */
+        JpaPartitionKeyBuilder withLastExecutionAttributeName(String lastExecutionAttributeName);
+
         /**
          * Sets the given job partition key attribute name.
          *
@@ -155,12 +179,12 @@ interface JpaPartitionKeyBuilder {
         JpaPartitionKeyBuilder withStateAttributeName(String stateAttributeName);
 
         /**
-         * Sets the given job ready state value.
+         * Sets the given state value mapping function.
          *
-         * @param readyStateValue The job ready state value
+         * @param stateValueMappingFunction The state value mapping function
          * @return this for chaining
          */
-        JpaPartitionKeyBuilder withReadyStateValue(Object readyStateValue);
+        JpaPartitionKeyBuilder withStateValueMappingFunction(Function<JobInstanceState, Object> stateValueMappingFunction);
 
         /**
          * Sets the given job attributes to fetch.
@@ -190,9 +214,10 @@ static JpaPartitionKeyBuilder builder() {
             Function<String, String> partitionPredicateProvider0;
             String idAttributeName0;
             String scheduleAttributeName0;
+            String lastExecutionAttributeName0;
             String partitionKeyAttributeName0;
             String stateAttributeName0;
-            Object readyStateValue0;
+            Function<JobInstanceState, Object> stateValueMappingFunction0;
             List<String> fetches0 = new ArrayList<>();
 
             @Override
@@ -225,6 +250,12 @@ public JpaPartitionKeyBuilder withScheduleAttributeName(String scheduleAttribute
                 return this;
             }
 
+            @Override
+            public JpaPartitionKeyBuilder withLastExecutionAttributeName(String lastExecutionAttributeName) {
+                this.lastExecutionAttributeName0 = lastExecutionAttributeName;
+                return this;
+            }
+
             @Override
             public JpaPartitionKeyBuilder withPartitionKeyAttributeName(String partitionKeyAttributeName) {
                 this.partitionKeyAttributeName0 = partitionKeyAttributeName;
@@ -238,8 +269,8 @@ public JpaPartitionKeyBuilder withStateAttributeName(String stateAttributeName)
             }
 
             @Override
-            public JpaPartitionKeyBuilder withReadyStateValue(Object readyStateValue) {
-                this.readyStateValue0 = readyStateValue;
+            public JpaPartitionKeyBuilder withStateValueMappingFunction(Function<JobInstanceState, Object> stateValueMappingFunction) {
+                this.stateValueMappingFunction0 = stateValueMappingFunction;
                 return this;
             }
 
@@ -257,9 +288,10 @@ public JpaPartitionKey build() {
                     private final Function<String, String> partitionPredicateProvider = partitionPredicateProvider0;
                     private final String idAttributeName = idAttributeName0;
                     private final String scheduleAttributeName = scheduleAttributeName0;
+                    private final String lastExecutionAttributeName = lastExecutionAttributeName0;
                     private final String partitionKeyAttributeName = partitionKeyAttributeName0;
                     private final String stateAttributeName = stateAttributeName0;
-                    private final Object readyStateValue = readyStateValue0;
+                    private final Function<JobInstanceState, Object> stateValueMappingFunction = stateValueMappingFunction0;
                     private final String[] fetches = fetches0.toArray(new String[fetches0.size()]);
 
                     @Override
@@ -282,6 +314,11 @@ public String getScheduleAttributeName() {
                         return scheduleAttributeName;
                     }
 
+                    @Override
+                    public String getLastExecutionAttributeName() {
+                        return lastExecutionAttributeName;
+                    }
+
                     @Override
                     public String getPartitionKeyAttributeName() {
                         return partitionKeyAttributeName;
@@ -293,8 +330,13 @@ public String getStatePredicate(String jobAlias) {
                     }
 
                     @Override
-                    public Object getReadyStateValue() {
-                        return readyStateValue;
+                    public String getStateExpression(String jobAlias) {
+                        return jobAlias + "." + stateAttributeName;
+                    }
+
+                    @Override
+                    public Function<JobInstanceState, Object> getStateValueMappingFunction() {
+                        return stateValueMappingFunction;
                     }
 
                     @Override

jpa/storage/src/main/java/com/blazebit/job/jpa/storage/JpaJobManager.java

@@ -33,6 +33,7 @@
 import com.blazebit.job.spi.TransactionSupport;
 
 import javax.persistence.EntityManager;
+import javax.persistence.Query;
 import javax.persistence.TypedQuery;
 import javax.persistence.metamodel.EntityType;
 import java.io.Serializable;
@@ -42,6 +43,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Function;
 
 /**
  * A JPA based implementation of the {@link JobManager} interface.
@@ -257,21 +259,21 @@ public List<JobInstance<?>> getJobInstancesToProcess(int partition, int partitio
         String partitionKeyAttributeName = jpaPartitionKey.getPartitionKeyAttributeName();
         String scheduleAttributeName = jpaPartitionKey.getScheduleAttributeName();
         String statePredicate = jpaPartitionKey.getStatePredicate("e");
-        Object readyStateValue = jpaPartitionKey.getReadyStateValue();
+        Function<JobInstanceState, Object> stateValueMappingFunction = jpaPartitionKey.getStateValueMappingFunction();
         String joinFetches = jpaPartitionKey.getJoinFetches("e");
         TypedQuery<? extends JobInstance<?>> typedQuery = entityManager.createQuery(
             "SELECT e FROM " + jobInstanceType.getName() + " e " +
                 joinFetches + " " +
-                "WHERE " + statePredicate + " " +
+                "WHERE e." + scheduleAttributeName + " <= :now " +
                 (partitionPredicate.isEmpty() ? "" : "AND " + partitionPredicate + " ") +
                 (partitionCount > 1 ? "AND MOD(e." + partitionKeyAttributeName + ", " + partitionCount + ") = " + partition + " " : "") +
-                "AND e." + scheduleAttributeName + " <= :now " +
+                (statePredicate == null || statePredicate.isEmpty() ? "" : "AND " + statePredicate + " ") +
                 "ORDER BY e." + scheduleAttributeName + " ASC, e." + idAttributeName + " ASC",
             jobInstanceType
         );
         typedQuery.setParameter("now", clock.instant());
-        if (readyStateValue != null) {
-            typedQuery.setParameter("readyState", readyStateValue);
+        if (stateValueMappingFunction != null) {
+            typedQuery.setParameter("readyState", stateValueMappingFunction);
         }
         List<JobInstance<?>> jobInstances = (List<JobInstance<?>>) (List) typedQuery
             // TODO: lockMode for update? advisory locks?
@@ -298,17 +300,18 @@ public Instant getNextSchedule(int partition, int partitionCount, PartitionKey p
         String partitionKeyAttributeName = jpaPartitionKey.getPartitionKeyAttributeName();
         String scheduleAttributeName = jpaPartitionKey.getScheduleAttributeName();
         String statePredicate = jpaPartitionKey.getStatePredicate("e");
-        Object readyStateValue = jpaPartitionKey.getReadyStateValue();
+        Function<JobInstanceState, Object> stateValueMappingFunction = jpaPartitionKey.getStateValueMappingFunction();
         TypedQuery<Instant> typedQuery = entityManager.createQuery(
             "SELECT e." + scheduleAttributeName + " FROM " + jobInstanceType.getName() + " e " +
-                "WHERE " + statePredicate + " " +
+                "WHERE 1=1 " +
+                (statePredicate == null || statePredicate.isEmpty() ? "" : "AND " + statePredicate + " ") +
                 (partitionPredicate.isEmpty() ? "" : "AND " + partitionPredicate + " ") +
                 (partitionCount > 1 ? "AND MOD(e." + partitionKeyAttributeName + ", " + partitionCount + ") = " + partition + " " : "") +
                 "ORDER BY e." + scheduleAttributeName + " ASC, e." + idAttributeName + " ASC",
             Instant.class
         );
-        if (readyStateValue != null) {
-            typedQuery.setParameter("readyState", readyStateValue);
+        if (stateValueMappingFunction != null) {
+            typedQuery.setParameter("readyState", stateValueMappingFunction);
         }
 
         List<Instant> nextSchedule = typedQuery.setMaxResults(1).getResultList();
@@ -325,8 +328,50 @@ public void updateJobInstance(JobInstance<?> jobInstance) {
         }
         if (!entityManager.contains(jobInstance)) {
             entityManager.merge(jobInstance);
+            if (jobInstance.getState() == JobInstanceState.REMOVED) {
+                entityManager.flush();
+                removeJobInstance(jobInstance);
+            }
+        } else if (jobInstance.getState() == JobInstanceState.REMOVED) {
+            removeJobInstance(jobInstance);
         }
 
         entityManager.flush();
     }
+
+    @Override
+    public void removeJobInstance(JobInstance<?> jobInstance) {
+        entityManager.remove(jobInstance);
+    }
+
+    @Override
+    public int removeJobInstances(Set<JobInstanceState> states, Instant executionTimeOlderThan, PartitionKey partitionKey) {
+        JpaPartitionKey jpaPartitionKey = (JpaPartitionKey) partitionKey;
+        String stateExpression = jpaPartitionKey.getStateExpression("i");
+        if (stateExpression != null && !stateExpression.isEmpty() && !states.isEmpty()) {
+            StringBuilder sb = new StringBuilder();
+            sb.append("DELETE FROM ").append(partitionKey.getJobInstanceType().getName()).append(" i ")
+                .append("WHERE ").append(stateExpression).append(" IN (");
+            int i = 0;
+            int size = states.size();
+            for (; i != size; i++) {
+                sb.append("param").append(i).append(',');
+            }
+            sb.setCharAt(sb.length() - 1, ')');
+            if (executionTimeOlderThan != null) {
+                sb.append(" AND i.").append(jpaPartitionKey.getLastExecutionAttributeName()).append(" < :lastExecution");
+            }
+            Query query = entityManager.createQuery(sb.toString());
+            i = 0;
+            for (JobInstanceState state : states) {
+                query.setParameter("param" + i, jpaPartitionKey.getStateValueMappingFunction().apply(state));
+                i++;
+            }
+            if (executionTimeOlderThan != null) {
+                query.setParameter("lastExecution", executionTimeOlderThan);
+            }
+            return query.executeUpdate();
+        }
+        return 0;
+    }
 }