<!-- A task scheduler that will call @Scheduled methods -->
    <task:scheduler id="myScheduler" pool-size="10"/>
    <task:annotation-driven scheduler="myScheduler"/>

/**
 * Persistent entity that stores an async task.
 * 
 * @author Carlos Vara
 */
@Entity
@Table(name="TASK_QUEUE")
public class QueuedTaskHolder {
 
    // Getters -----------------------------------------------------------------
 
    @Id
    @MyAppId
    public String getId() {
        if ( this.id == null ) {
            this.setId(UUIDHelper.newUUID());
        }
        return this.id;
    }
 
    @NotNull
    @Past
    @Temporal(TemporalType.TIMESTAMP)
    public Calendar getCreationStamp() {
        return this.creationStamp;
    }
 
    @Temporal(TemporalType.TIMESTAMP)
    public Calendar getTriggerStamp() {
        return triggerStamp;
    }
 
    @Past
    @Temporal(TemporalType.TIMESTAMP)
    public Calendar getStartedStamp() {
        return this.startedStamp;
    }
 
    @Past
    @Temporal(TemporalType.TIMESTAMP))
    public Calendar getCompletedStamp() {
        return this.completedStamp;
    }
 
    @Lob
    @NotNull
    public byte[] getSerializedTask() {
        return this.serializedTask;
    }
 
    @Version
    protected int getVersion() {
        return this.version;
    }
 
 
    // Setters -----------------------------------------------------------------
 
    protected void setId(String id) {
        this.id = id;
    }
 
    public void setCreationStamp(Calendar creationStamp) {
        this.creationStamp = creationStamp;
    }
 
    public void setTriggerStamp(Calendar triggerStamp) {
        this.triggerStamp = triggerStamp;
    }
 
    public void setStartedStamp(Calendar startedStamp) {
        this.startedStamp = startedStamp;
    }
 
    public void setCompletedStamp(Calendar completedStamp) {
        this.completedStamp = completedStamp;
    }
 
    public void setSerializedTask(byte[] serializedTask) {
        this.serializedTask = serializedTask;
    }
 
    public void setVersion(int version) {
        this.version = version;
    }
 
 
    // Fields ------------------------------------------------------------------
 
    private String id;
    private Calendar creationStamp;
    private Calendar triggerStamp = null;
    private Calendar startedStamp = null;
    private Calendar completedStamp = null;
    private byte[] serializedTask;
    private int version;
 
 
    // Lifecycle events --------------------------------------------------------
 
    @SuppressWarnings("unused")
    @PrePersist
    private void onAbstractBaseEntityPrePersist() {
        this.ensureId();
        this.markCreation();
    }
 
    /**
     * Ensures that the entity has a unique UUID.
     */
    private void ensureId() {
        this.getId();
    }
 
    /**
     * Sets the creation stamp to now.
     */
    private void markCreation() {
        setCreationStamp(Calendar.getInstance(TimeZone.getTimeZone("Etc/UTC")));
    }
 
 
    @Override
    public String toString() {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy.MM.dd HH:mm:ss z");
        return new ToStringCreator(this).append("id", getId())
            .append("creationStamp", (getCreationStamp()!=null)?sdf.format(getCreationStamp().getTime()):null)
            .append("startedStamp", (getStartedStamp()!=null)?sdf.format(getStartedStamp().getTime()):null)
            .append("completedStamp", (getCompletedStamp()!=null)?sdf.format(getCompletedStamp().getTime()):null)
            .toString();
    }    
 
}

/**
 * DAO operations for the {@link QueuedTaskHolder} entities.
 * 
 * @author Carlos Vara
 */
public interface QueuedTaskHolderDao {
 
    /**
     * Adds a new task to the current persistence context. The task will be
     * persisted into the database at flush/commit.
     * 
     * @param queuedTask
     *            The task to be saved (enqueued).
     */
    void persist(QueuedTaskHolder queuedTask);
 
 
    /**
     * Finder that retrieves a task by its id.
     * 
     * @param taskId
     *            The id of the requested task.
     * @return The task with that id, or <code>null</code> if no such task
     *         exists.
     */
    QueuedTaskHolder findById(String taskId);
 
 
    /**
     * @return A task which is candidate for execution. The receiving thread
     *         will need to ensure a lock on it. <code>null</code> if no
     *         candidate task is available.
     */
    QueuedTaskHolder findNextTaskForExecution();
 
 
    /**
     * @return A task which has been in execution for too long without
     *         finishing. <code>null</code> if there aren't stalled tasks.
     */
    QueuedTaskHolder findRandomStalledTask();
 
}

/**
 * JPA2 implementation of {@link QueuedTaskHolderDao}.
 * 
 * @author Carlos Vara
 */
@Repository
public class QueuedTaskHolderDaoJPA2 implements QueuedTaskHolderDao {
 
 
    // QueuedTaskDao methods ---------------------------------------------------
 
    @Override
    public void persist(QueuedTaskHolder queuedTask) {
        this.entityManager.persist(queuedTask);
    }
 
    @Override
    public QueuedTaskHolder findById(String taskId) {
        return this.entityManager.find(QueuedTaskHolder.class, taskId);
    }
 
    @Override
    public QueuedTaskHolder findNextTaskForExecution() {
 
        Calendar NOW = Calendar.getInstance();
 
        // select qt from QueuedTask where
        //      qt.startedStamp == null AND
        //      (qth.triggerStamp == null || qth.triggerStamp < NOW)
        // order by qth.version ASC, qt.creationStamp ASC
        CriteriaBuilder cb = this.entityManager.getCriteriaBuilder();
        CriteriaQuery<QueuedTaskHolder> cq = cb.createQuery(QueuedTaskHolder.class);
        Root<QueuedTaskHolder> qth = cq.from(QueuedTaskHolder.class);
        cq.select(qth)
            .where(cb.and(cb.isNull(qth.get(QueuedTaskHolder_.startedStamp)), 
                    cb.or(
                            cb.isNull(qth.get(QueuedTaskHolder_.triggerStamp)),
                            cb.lessThan(qth.get(QueuedTaskHolder_.triggerStamp), NOW))))
            .orderBy(cb.asc(qth.get(QueuedTaskHolder_.version)), cb.asc(qth.get(QueuedTaskHolder_.creationStamp)));
 
        List<QueuedTaskHolder> results = this.entityManager.createQuery(cq).setMaxResults(1).getResultList();
        if ( results.isEmpty() ) {
            return null;
        }
        else {
            return results.get();
        }
 
    }
 
    @Override
    public QueuedTaskHolder findRandomStalledTask() {
 
        Calendar TOO_LONG_AGO = Calendar.getInstance();
        TOO_LONG_AGO.add(Calendar.SECOND, -7200);
 
        // select qth from QueuedTask where 
        //      qth.startedStamp != null AND
        //      qth.startedStamp < TOO_LONG_AGO
        CriteriaBuilder cb = this.entityManager.getCriteriaBuilder();
        CriteriaQuery<QueuedTaskHolder> cq = cb.createQuery(QueuedTaskHolder.class);
        Root<QueuedTaskHolder> qth = cq.from(QueuedTaskHolder.class);
        cq.select(qth).where(
                cb.and(
                        cb.isNull(qth.get(QueuedTaskHolder_.completedStamp)),
                        cb.lessThan(qth.get(QueuedTaskHolder_.startedStamp), TOO_LONG_AGO)));
 
        List<QueuedTaskHolder> stalledTasks = this.entityManager.createQuery(cq).getResultList();
 
        if ( stalledTasks.isEmpty() ) {
            return null;
        }
        else {
            Random rand = new Random(System.currentTimeMillis());
            return stalledTasks.get(rand.nextInt(stalledTasks.size()));
        }
 
    }
 
 
    // Injected dependencies ---------------------------------------------------
 
    @PersistenceContext
    private EntityManager entityManager;
 
}

/**
 * A task executor with persistent task queueing.
 * 
 * @author Carlos Vara
 */
@Component("MyTaskExecutor")
public class MyTaskExecutor implements TaskExecutor {
 
    final static Logger logger = LoggerFactory.getLogger(MyTaskExecutor.class);
 
    @Autowired
    protected QueuedTaskHolderDao queuedTaskDao;
 
    @Autowired
    protected Serializer serializer;
 
 
    /**
     * Additional requirement: must be run inside a transaction.
     * Currently using MANDATORY as the app won't create tasks outside a
     * transaction.
     * 
     * @see org.springframework.core.task.TaskExecutor#execute(java.lang.Runnable)
     */
    @Override
    @Transactional(propagation=Propagation.MANDATORY)
    public void execute(Runnable task) {
 
        logger.debug("Trying to enqueue: {}", task);
 
        AbstractBaseTask abt; 
        try {
            abt = AbstractBaseTask.class.cast(task);
        } catch (ClassCastException e) {
            logger.error("Only runnables that extend AbstractBaseTask are accepted.");
            throw new IllegalArgumentException("Invalid task: " + task);
        }
 
        // Serialize the task
        QueuedTaskHolder newTask = new QueuedTaskHolder();
        byte[] serializedTask = this.serializer.serializeObject(abt);
        newTask.setTriggerStamp(abt.getTriggerStamp());
 
        logger.debug("New serialized task takes {} bytes", serializedTask.length);
 
        newTask.setSerializedTask(serializedTask);
 
        // Store it in the db
        this.queuedTaskDao.persist(newTask);
 
        // POST: Task has been enqueued
    }
 
 
    /**
     * Runs enqueued tasks.
     */
    @Scheduled(fixedRate=60l*1000l) // Every minute
    public void runner() {
 
        logger.debug("Started runner {}", Thread.currentThread().getName());
 
        QueuedTaskHolder lockedTask = null;
 
        // While there is work to do...
        while ( (lockedTask = tryLockTask()) != null ) {
 
            logger.debug("Obtained lock on {}", lockedTask);
 
            // Deserialize the task
            AbstractBaseTask runnableTask = this.serializer.deserializeAndCast(lockedTask.getSerializedTask());
            runnableTask.setQueuedTaskId(lockedTask.getId());
 
            // Run it
            runnableTask.run();
        }
 
        logger.debug("Finishing runner {}, nothing else to do.", Thread.currentThread().getName());
    }
 
 
    /**
     * The hypervisor re-queues for execution possible stalled tasks.
     */
    @Scheduled(fixedRate=60l*60l*1000l) // Every hour
    public void hypervisor() {
 
        logger.debug("Started hypervisor {}", Thread.currentThread().getName());
 
        // Reset stalled threads, one at a time to avoid too wide transactions
        while ( tryResetStalledTask() );
 
        logger.debug("Finishing hypervisor {}, nothing else to do.", Thread.currentThread().getName());
    }
 
 
    /**
     * Tries to ensure a lock on a task in order to execute it.
     * 
     * @return A locked task, or <code>null</code> if there is no task available
     *         or no lock could be obtained.
     */
    private QueuedTaskHolder tryLockTask() {
 
        int tries = 3;
 
        QueuedTaskHolder ret = null;
        while ( tries >  ) {
            try {
                ret = obtainLockedTask();
                return ret;
            } catch (OptimisticLockingFailureException e) {
                tries--;
            }
        }
 
        return null;
    }
 
    /**
     * Tries to reset a stalled task.
     * 
     * @return <code>true</code> if one task was successfully re-queued,
     *         <code>false</code> if no task was re-queued, either because there
     *         are no stalled tasks or because there was a conflict re-queueing
     *         it.
     */
    private boolean tryResetStalledTask() {
        int tries = 3;
 
        QueuedTaskHolder qt = null;
        while ( tries >  ) {
            try {
                qt = resetStalledTask();
                return qt != null;
            } catch (OptimisticLockingFailureException e) {
                tries--;
            }
        }
 
        return false;
    }
 
    /**
     * @return A locked task ready for execution, <code>null</code> if no ready
     *         task is available.
     * @throws OptimisticLockingFailureException
     *             If getting the lock fails.
     */
    @Transactional
    public QueuedTaskHolder obtainLockedTask() {
        QueuedTaskHolder qt = this.queuedTaskDao.findNextTaskForExecution();
        logger.debug("Next possible task for execution {}", qt);
        if ( qt != null ) {
            qt.setStartedStamp(Calendar.getInstance(TimeZone.getTimeZone("etc/UTC")));
        }
        return qt;
    }
 
 
    /**
     * Tries to reset a stalled task, returns null if no stalled task was reset.
     * 
     * @return The re-queued task, <code>null</code> if no stalled task is
     *         available.
     * @throws OptimisticLockingFailureException
     *             If the stalled task is modified by another thread during
     *             re-queueing.
     */
    @Transactional
    public QueuedTaskHolder resetStalledTask() {
        QueuedTaskHolder stalledTask = this.queuedTaskDao.findRandomStalledTask();
        logger.debug("Obtained this stalledTask {}", stalledTask);
        if ( stalledTask != null ) {
            stalledTask.setStartedStamp(null);
        }
        return stalledTask;
    }
 
}

/**
 * Superclass for all async tasks.
 * <ul>
 *  <li>Ensures that its associated queued task is marked as completed in the same tx.</li>
 *  <li>Marks the task as serializable.</li>
 * </ul>
 * 
 * @author Carlos Vara
 */
public abstract class AbstractBaseTask implements Runnable, Serializable {
 
    final static Logger logger = LoggerFactory.getLogger(AbstractBaseTask.class);
 
 
    // Common data -------------------------------------------------------------
 
    private transient String queuedTaskId;
    private transient QueuedTaskHolder qth;
    private transient Calendar triggerStamp;
 
 
    public void setQueuedTaskId(String queuedTaskId) {
        this.queuedTaskId = queuedTaskId;
    }
 
    public String getQueuedTaskId() {
        return queuedTaskId;
    }
 
    public void setTriggerStamp(Calendar triggerStamp) {
        this.triggerStamp = triggerStamp;
    }
 
    public Calendar getTriggerStamp() {
        return triggerStamp;
    }
 
 
    // Injected components -----------------------------------------------------
 
    @Autowired(required=true)
    protected transient QueuedTaskHolderDao queuedTaskHolderDao;
 
 
    // Lifecycle methods -------------------------------------------------------
 
    /**
     * Entrance point of the task.
     * <ul>
     *  <li>Ensures that the associated task in the queue exists.</li>
     *  <li>Marks the queued task as finished upon tx commit.</li>
     *  <li>In case of tx rollback, frees the task.</li>
     * </ul>
     * 
     * @see java.lang.Runnable#run()
     */
    @Override
    final public void run() {
 
        try {
            transactionalOps();
        } catch (RuntimeException e) {
            // Free the task, so it doesn't stall
            logger.warn("Exception forced task tx rollback: {}", e);
            freeTask();
        }
 
    }
 
    @Transactional
    private void transactionalOps() {
        doInTxBeforeTask();
        doTaskInTransaction();
        doInTxAfterTask();
    }
 
    @Transactional
    private void freeTask() {
        QueuedTaskHolder task = this.queuedTaskHolderDao.findById(this.queuedTaskId);
        task.setStartedStamp(null);
    }
 
 
    /**
     * Ensures that there is an associated task and that its state is valid.
     * Shouldn't be needed, just for extra security.
     */
    private void doInTxBeforeTask() {
        this.qth = this.queuedTaskHolderDao.findById(this.queuedTaskId);
        if ( this.qth == null ) {
            throw new IllegalArgumentException("Not executing: no associated task exists: " + this.getQueuedTaskId());
        }
        if ( this.qth.getStartedStamp() == null || this.qth.getCompletedStamp() != null ) {
            throw new IllegalArgumentException("Illegal queued task status: " + this.qth);
        }
    }
 
 
    /**
     * Method to be implemented by concrete tasks where their operations are
     * performed.
     */
    public abstract void doTaskInTransaction();
 
 
    /**
     * Marks the associated task as finished.
     */
    private void doInTxAfterTask() {
        this.qth.setCompletedStamp(Calendar.getInstance());
    }
 
 
    private static final long serialVersionUID = 1L;
}

/**
 * Logs the status of a User.
 * 
 * @author Carlos Vara
 */
@Configurable
public class ExampleTask extends AbstractBountyTask {
 
    final static Logger logger = LoggerFactory.getLogger(ExampleTask.class);
 
 
    // Injected components -----------------------------------------------------
 
    @Autowired
    private transient UserDao userDao;
 
 
    // Data --------------------------------------------------------------------
 
    private final String userId;
 
 
    public ExampleTask(String userId) {
        this.userId = userId;
    }
 
 
    /**
     * Logs the status of a user.
     */
    @Override
    public void doTaskInTransaction() {
 
        // Get the user
        User user = this.userDao.findBagById(this.userId);
        if ( user == null ) {
            logger.error("User {} doesn't exist in the system.", userId);
            return;
        }
 
        // Log the user status
        logger.info("User status: {}", user.getStatus());
    }
 
    private static final long serialVersionUID = 1L;
}

@Service
public class ExampleServiceImpl implements ExampleService {
 
    @Qualifier("BountyExecutor")
    @Autowired
    private TaskExecutor taskExecutor;
 
    @Transactional
    public void example() {
        // Task will execute ASAP
        this.taskExecutor.execute(new ExampleTask("1"));
        // Task won't execute until tomorrow
        ExampleTask et = new ExampleTask("2");
        Calendar tomorrow = Calendar.getInstance();
        tomorrow.add(Calendar.DAY, 1);
        et.setTriggerStamp(tomorrow);
        this.taskExecutor.execute(et);
    }
}