Periodic jobs when running multiple servers

Question

I'm planning to deploy an app using Play and have never used their "Jobs" before. My deploy will be large enough to require different Play servers load-balanced, but my computations won't be large enough to need hadoop/storm/others.

My question is, how do I handle this scenario in Play? If I set up a job in Play to run every minute, I don't want every single server to do the exact same thing at the same time.

I could only find this answer but I don't like any of those options.

So, is there any tools or best practices to coordinate jobs or do I have to do something from scratch?

Answer 1

You can use a table in your database to store a jobLock but you have to check/update this lock in a separate transactions (you have to use JPA.newEntityManager for this)

My JobLock class uses a LockMode enum

package enums;

public enum LockMode {
    FREE, ACQUIRED;
}

here is the JobLock class

package models;

import java.util.Date;
import java.util.List;

import javax.persistence.Entity;
import javax.persistence.EntityManager;
import javax.persistence.EnumType;
import javax.persistence.Enumerated;
import javax.persistence.Version;

import play.Logger;
import play.Play;
import play.data.validation.Required;
import play.db.jpa.JPA;
import play.db.jpa.Model;
import utils.Parser;
import enums.LockMode;
import exceptions.ServiceException;

/**
 * Technical class that allows to manage a lock in the database thus we can
 * synchronize multiple instances that thus cannot run the same job at the same
 * time
 * 
 * @author sebastien
 */
@Entity
public class JobLock extends Model {

    private static final Long MAX_ACQUISITION_DELAY = Parser.parseLong(Play.configuration.getProperty(
            "job.lock.acquisitiondelay", "10000"));

    @Required
    public String jobName;

    public Date acquisitionDate;

    @Required
    @Enumerated(EnumType.STRING)
    public LockMode lockMode;

    @Version
    public int version;

    // STATIC METHODS
    // ~~~~~~~~~~~~~~~

    /**
     * Acquire the lock for the type of job identified by the name parameter.
     * Acquisition of the lock is done on a separate transaction thus is
     * transaction is as small as possible and other instances will see the lock
     * acquisition sooner.
     * <p>
     * If we do not do that, the other instances will be blocked until the
     * instance that acquired the lock have finished is businees transaction
     * which could be long on a job.
     * </p>
     * 
     * @param name
     *            the name that identifies a job category, usually it is the job
     *            simple class name
     * @return the lock object if the acquisition is successfull, null otherwise
     */
    public static JobLock acquireLock(String name) {
        EntityManager em = JPA.newEntityManager();
        try {
            em.getTransaction().begin();
            List<JobLock> locks = em.createQuery("from JobLock where jobName=:name", JobLock.class)
                    .setParameter("name", name).setMaxResults(1).getResultList();
            JobLock lock = locks != null && !locks.isEmpty() ? locks.get(0) : null;
            if (lock == null) {
                lock = new JobLock();
                lock.jobName = name;
                lock.acquisitionDate = new Date();
                lock.lockMode = LockMode.ACQUIRED;
                em.persist(lock);
            } else {
                if (LockMode.ACQUIRED.equals(lock.lockMode)) {
                    if ((System.currentTimeMillis() - lock.acquisitionDate.getTime()) > MAX_ACQUISITION_DELAY) {
                        throw new ServiceException(String.format(
                                "Lock is held for too much time : there is a problem with job %s", name));
                    }
                    return null;
                }
                lock.lockMode = LockMode.ACQUIRED;
                lock.acquisitionDate = new Date();
                lock.willBeSaved = true;
            }
            em.flush();
            em.getTransaction().commit();
            return lock;
        } catch (Exception e) {
            // Do not log exception here because it is normal to have exception
            // in case of multi-node installation, this is the way to avoid
            // multiple job execution
            if (em.getTransaction().isActive()) {
                em.getTransaction().rollback();
            }
            // Maybe we have to inverse the test and to define which exception
            // is not problematic : exception that denotes concurrency in the
            // database are normal
            if (e instanceof ServiceException) {
                throw (ServiceException) e;
            } else {
                return null;
            }
        } finally {
            if (em.isOpen()) {
                em.close();
            }
        }
    }

    /**
     * Release the lock on the database thus another instance can take it. This
     * action change the {@link #lockMode} and set {@link #acquisitionDate} to
     * null. This is done in a separate transaction that can have visibility on
     * what happens on the database during the time of the business transaction
     * 
     * @param lock
     *            the lock to release
     * @return true if we managed to relase the lock and false otherwise
     */
    public static boolean releaseLock(JobLock lock) {
        EntityManager em = JPA.newEntityManager();

        if (lock == null || LockMode.FREE.equals(lock.lockMode)) {
            return false;
        }

        try {
            em.getTransaction().begin();
            lock = em.find(JobLock.class, lock.id);
            lock.lockMode = LockMode.FREE;
            lock.acquisitionDate = null;
            lock.willBeSaved = true;
            em.persist(lock);
            em.flush();
            em.getTransaction().commit();
            return true;
        } catch (Exception e) {
            if (em.getTransaction().isActive()) {
                em.getTransaction().rollback();
            }
            Logger.error(e, "Error during commit of lock release");
            return false;
        } finally {
            if (em.isOpen()) {
                em.close();
            }
        }
    }
}

and here is my LockAwareJob that uses this lock

package jobs;

import models.JobLock;
import notifiers.ExceptionMails;
import play.Logger;
import play.jobs.Job;

public abstract class LockAwareJob<V> extends Job<V> {

    @Override
    public final void doJob() throws Exception {
        String name = this.getClass().getSimpleName();
        try {
            JobLock lock = JobLock.acquireLock(name);
            if (lock != null) {
                Logger.info("Starting %s", name);
                try {
                    doJobWithLock(lock);
                } finally {
                    if (!JobLock.releaseLock(lock)) {
                        Logger.error("Lock acquired but cannot be released for %s", name);
                    }
                    Logger.info("End of %s", name);
                }
            } else {
                Logger.info("Another node is running %s : nothing to do", name);
            }
        } catch (Exception ex) {
            ExceptionMails.exception(ex, String.format("Error while executing job %s", name));
            throw ex;
        }
    }

    @Override
    public final V doJobWithResult() throws Exception {
        String name = this.getClass().getSimpleName();
        try {
            JobLock lock = JobLock.acquireLock(name);
            if (lock != null) {
                Logger.info("Starting %s", name);
                try {
                    return resultWithLock(lock);
                } finally {
                    if (!JobLock.releaseLock(lock)) {
                        Logger.error("Lock acquired but cannot be released for %s", name);
                    }
                    Logger.info("End of %s", name);
                }
            } else {
                Logger.info("Another node is running %s : nothing to do", name);
                return resultWithoutLock();
            }
        } catch (Exception ex) {
            ExceptionMails.exception(ex, String.format("Error while executing job %s", name));
            throw ex;
        }
    }

    public void doJobWithLock(JobLock lock) throws Exception {
    }

    public V resultWithLock(JobLock lock) throws Exception {
        doJobWithLock(lock);
        return null;
    }

    public V resultWithoutLock() throws Exception {
        return null;
    }
}

In my log4j.properties I add a special line to avoid having an error each time an instance failed acquiring the job lock

log4j.logger.org.hibernate.event.def.AbstractFlushingEventListener=FATAL

With this solution you can also use the JobLock id to store parameters associated with this job (last run date for example)

Answer 2

I would personally use one instance running jobs only for simplicity. Alternatively you could look at using Akka instead of Jobs if you want finer control over execution and better concurrency, parallel handling.

Answer 3

You can use a database flag as described here: Playframework concurrent jobs management by Pere Villega for two jobs.

But I think the solution from Guillaume Bort on Google Groups to use Memcache is the best one. There seems to be a module for Play 2: https://github.com/mumoshu/play2-memcached