How can I implement a proper counter bean with EJB 3.0?

Question

[EDIT] This question is "how do I do atomic changes to entity beans with EJB 3 and JPA 2.0". Should be simple, right?

I tried to fix my code based on the answers I got so far. I'm using JBoss 6.0.0M2 with Hypersonic (just download it and call run.bat).

My test case: Create 3 threads and call one of the testCounterMitLock*() 500 times in a loop. So a successful test should print "Anzahl eingetragene Zeilen: 1500" (3*500).

I tried:

        CounterTestVersion ct = manager.find(CounterTestVersion.class, 1);
        manager.lock(ct, LockModeType.WRITE);
        int wert = ct.getWert();

Obviously doesn't work because a different thread can change the value in the database before the lock is applied. So I try to fix that:

        CounterTestVersion ct = manager.find(CounterTestVersion.class, 1);
        manager.lock(ct, LockModeType.WRITE);
        manager.refresh (ct);
        int wert = ct.getWert();

The refresh() should give me the current value and the implicit query should also make sure the object gets locked now. No such luck. Let's try with JPA 2.0:

        CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.WRITE);
        int wert = ct.getWert();

That also doesn't work. Maybe the lock isn't enough?

        CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.PESSIMISTIC_WRITE);
        int wert = ct.getWert();

Uhm ... doesn't work either! One last desperate attempt:

        CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.PESSIMISTIC_WRITE);
        manager.flush();
        manager.refresh (ct);
        int wert = ct.getWert();

Okay ... can anyone explain why nothing works? I'm out of ideas.

[EDIT2] PS: To add insult to injury, this is the last output of the last running thread:

commit/rollback: 441/62

(441+62 = 503)...

Here is the complete code. First the bean:

package server.kap15;

import java.rmi.RemoteException;

import javax.ejb.*;
import javax.persistence.*;

@Stateful
public class CounterTestBean implements CounterTestRemote, SessionSynchronization {
    @PersistenceContext(unitName = "JavaEE")
    EntityManager manager;

    private int commit = 0;

    private int rollback = 0;

    public void initDatenbank() {
        manager.createNamedQuery("CounterTest.deleteAll").executeUpdate();
        manager.createNamedQuery("TestTabelle.deleteAll").executeUpdate();
        CounterTestVersion ct = new CounterTestVersion();
        ct.setNr(1);
        ct.setVersion(1);
        ct.setWert(1);
        manager.persist(ct);
    }

    public boolean testCounterOhneLock() {
        try {
            CounterTest ct = manager.find(CounterTest.class, 1);
            int wert = ct.getWert();
            ct.setWert(wert + 1);
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (Throwable t) {
            return false;
        }
    }

    public boolean testCounterMitLock() {
        try {
            CounterTestVersion ct = manager.find(CounterTestVersion.class, 1);
            manager.lock(ct, LockModeType.WRITE);
            int wert = ct.getWert();
            ct.setWert(wert + 1);
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (Throwable t) {
            return false;
        }
    }

    public boolean testCounterMitLock2() {
        try {
            CounterTestVersion ct = manager.find(CounterTestVersion.class, 1);
            manager.lock(ct, LockModeType.WRITE);
            manager.refresh (ct);
            int wert = ct.getWert();
            ct.setWert(wert + 1);
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (Throwable t) {
            return false;
        }
    }

    public boolean testCounterMitLock3() {
        try {
            CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.WRITE);
            int wert = ct.getWert();
            ct.setWert(wert + 1);
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (Throwable t) {
            return false;
        }
    }

    public boolean testCounterMitLock4() {
        try {
            CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.PESSIMISTIC_WRITE);
            int wert = ct.getWert();
            ct.setWert(wert + 1);
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (Throwable t) {
            return false;
        }
    }

    public boolean testCounterMitLock5() {
        try {
            CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.PESSIMISTIC_WRITE);
            manager.flush();
            manager.refresh (ct);
            int wert = ct.getWert();
            ct.setWert(wert + 1);
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (Throwable t) {
            return false;
        }
    }

    public boolean testCounterMitVersion() {
        try {
            CounterTestVersion ctv = manager.find(CounterTestVersion.class, 1);
            int wert = ctv.getWert();
            ctv.setWert(wert + 1);
            manager.flush();
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (OptimisticLockException e) {
            System.out.println(">>> Versionskonflikt !");
            return false;
        } catch (Throwable t) {
            System.out.println(t.getMessage());
            return false;
        }
    }

    public long anzTestZeilen() {
        Query query = manager.createNamedQuery("TestTabelle.anzZeilen");
        Long anzahl = (Long) query.getSingleResult();
        return anzahl;
    }

    public void afterBegin() throws EJBException, RemoteException {
    }

    public void beforeCompletion() throws EJBException, RemoteException {
    }

    public void afterCompletion(boolean committed) throws EJBException,
    RemoteException {
        if (committed)
            commit++;
        else
            rollback++;
        System.out.println("commit/rollback: " + commit + "/" + rollback);
    }
}

The remote interface:

package server.kap15;

import javax.ejb.Remote;

@Remote
public interface CounterTestRemote {
    public void initDatenbank();

    public boolean testCounterOhneLock();

    public boolean testCounterMitLock();
    public boolean testCounterMitLock2();
    public boolean testCounterMitLock3();
    public boolean testCounterMitLock4();
    public boolean testCounterMitLock5();

    public boolean testCounterMitVersion();

    public long anzTestZeilen();
}

The persistence.xml:

<?xml version="1.0" encoding="UTF-8"?>
<persistence xmlns="http://java.sun.com/xml/ns/persistence"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://java.sun.com/xml/ns/persistence
        http://java.sun.com/xml/ns/persistence/persistence_1_0.xsd"
    version="1.0">
    <persistence-unit name="JavaEE">
        <jta-data-source>java:DefaultDS</jta-data-source>
    </persistence-unit>
</persistence>

The test client:

package client.kap15;

import java.util.Properties;
import javax.naming.*;
import javax.rmi.PortableRemoteObject;
import server.kap15.CounterTestRemote;

public class CounterTestMitLock extends Thread {
    CounterTestRemote ctr;

    public static void main(String[] args) {
        try
        {
            testMitLock();
            testMitLock2();
            testMitLock3();
            testMitLock4();
            testMitLock5();
        }
        catch (Exception e)
        {
            e.printStackTrace ();
        }
    }

    static int N = 3;
    static CounterThread[] ct = new CounterThread[N];

    private static void testMitLock () throws InterruptedException
    {
        System.out.println("--- Counter Test MIT Lock ----------------------");
        System.out.println("Testinstanzen erzeugen...");
        for (int i=0; i<N; i++)
            ct[i] = new CounterThreadMitLock();

        runTest ();
    }

    private static void testMitLock2 () throws InterruptedException
    {
        System.out.println("--- Counter Test MIT Lock2 ----------------------");
        System.out.println("Testinstanzen erzeugen...");
        for (int i=0; i<N; i++)
            ct[i] = new CounterThreadMitLock2();

        runTest ();
    }

    private static void testMitLock3 () throws InterruptedException
    {
        System.out.println("--- Counter Test MIT Lock3 ----------------------");
        System.out.println("Testinstanzen erzeugen...");
        for (int i=0; i<N; i++)
            ct[i] = new CounterThreadMitLock3();

        runTest ();
    }

    private static void testMitLock4 () throws InterruptedException
    {
        System.out.println("--- Counter Test MIT Lock4 ----------------------");
        System.out.println("Testinstanzen erzeugen...");
        for (int i=0; i<N; i++)
            ct[i] = new CounterThreadMitLock4();

        runTest ();
    }

    private static void testMitLock5 () throws InterruptedException
    {
        System.out.println("--- Counter Test MIT Lock5 ----------------------");
        System.out.println("Testinstanzen erzeugen...");
        for (int i=0; i<N; i++)
            ct[i] = new CounterThreadMitLock5();

        runTest ();
    }

    private static void runTest () throws InterruptedException
    {
        System.out.println("Datenbank initialisieren...");
        ct[0].ctr.initDatenbank();

        System.out.println("Test durchführen...");
        for (int i=0; i<N; i++)
            ct[i].start();

        System.out.println("Auf Ende warten...");
        for (int i=0; i<N; i++)
            ct[i].join();

        System.out.println("Anzahl eingetragene Zeilen: " + ct[0].ctr.anzTestZeilen());
    }

    private static CounterTestRemote verbinden() {
        try {
            Properties p = new Properties();
            p.put(Context.INITIAL_CONTEXT_FACTORY, "org.jnp.interfaces.NamingContextFactory");
            p.put(Context.URL_PKG_PREFIXES, "org.jboss.naming:org.jnp.interfaces");
            p.put(Context.PROVIDER_URL, "jnp://localhost:1099");
            Context ctx = new InitialContext(p);

            Object ref = ctx.lookup("CounterTestBean/remote");
            CounterTestRemote ctr = (CounterTestRemote) PortableRemoteObject.narrow(ref, CounterTestRemote.class);

            return ctr;
        } catch (NamingException e) {
            System.out.println("ERROR - NamingException!");
            System.exit(-1);
        }
        return null;
    }

    public abstract static class CounterThread extends Thread
    {
        protected CounterTestRemote ctr;

        public CounterThread ()
        {
            this.ctr = verbinden ();
        }

        public void run() {
            for (int i = 0; i < 500; i++)
                test ();
        }

        public abstract void test ();
    }

    public static class CounterThreadMitLock extends CounterThread
    {
        @Override
        public void test ()
        {
            this.ctr.testCounterMitLock();
        }

    }

    public static class CounterThreadMitLock2 extends CounterThread
    {
        @Override
        public void test ()
        {
            this.ctr.testCounterMitLock2();
        }

    }

    public static class CounterThreadMitLock3 extends CounterThread
    {
        @Override
        public void test ()
        {
            this.ctr.testCounterMitLock3();
        }

    }

    public static class CounterThreadMitLock4 extends CounterThread
    {
        @Override
        public void test ()
        {
            this.ctr.testCounterMitLock4();
        }

    }

    public static class CounterThreadMitLock5 extends CounterThread
    {
        @Override
        public void test ()
        {
            this.ctr.testCounterMitLock5();
        }

    }
}

Answer 1

You don't show what you do with the return value from testCounterWithLock. My guess is that you are getting Optimistic locking failures and the return value is false sometimes.

Optimstic locking is a reasonable model when clashes are likely to be infrequent in practice, and the caller can reasonbly redo the work. So if you are getting optimistic failures you could just retry.

Alternatively, use a pessimistic locking model, this locks the row in the database at the point you read. You can do that my adding a LockMode of Pessimistic to your call to find(). Use of pessimistic locking needs to be done with care, all too easy to get bad concurrency and/or deadlocks.

Answer 2

Even with LockModeType.READ or LockModeType.WRITE, JPA 1.0 does support only optimistic locking. The lock acquisitions can still be deferred to commit time, hence the problem you experienced.

From JPA 2.0 concurrency and locking:

PA 1.0 only supported Optimistic read or Optimistic write locking. JPA 2.0 supports Optimistic and Pessimistic locking

Other resources: EJB3 performance and Pessimist Locking with JPA

To have real pessimistic locking with JPA 1.0, you will need to rely on the database or implementation specific extension. E.g.:

JPA 2.0 (Something similar is possible with Hibernate API)

Account acc = em.find( Account.class, id, PESSIMISTIC );

JPA 1.0

Query query = em.createNativeQuery("SELECT * ... FOR UPDATE"); // works with most db
Account acc = (Account) query.getSingleResult();

At least, that's what I've finally use, because lock didn't worked as expected.

( Note: You could also implement a retry logic when an optimistic exception happens. But it's complicated because the transactions are managed by the app. server. You would need to use @TRANSACTION_NEW to pause the current transaction and start a new one, etc. ... too complicated I think! )

Answer 3

Since none of the locking modes worked, I tried ewernli's solution with a manual SELECT ... FOR UPDATE. That gave an interesting exception: "Unexpected token FOR". So I looked at the database.

JBoss is installed with Hypersonic 1.8 (HSQLDB) as the default which doesn't support row locking. Dear JBoss developers: A JPA implementation is supposed to throw an exception when a locking mode isn't supported.

So I added an Oracle datasource and changed my persistence.xml. After that, two tests work:

        CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.PESSIMISTIC_WRITE);
        int wert = ct.getWert();

and

    Query query = manager.createNativeQuery ("select * from COUNTER_TEST where NR = 1 for update", CounterTestVersion.class);
    CounterTestVersion ct = (CounterTestVersion)query.getSingleResult ();
    int wert = ct.getWert ()+1;

which is interesting. It should work with LockModeType.PESSIMISTIC_FORCE_INCREMENT, too. In this case, I see this error in the log:

ORA-00054: resource busy and acquire with NOWAIT specified

This happens in the call manager.find(). I can't see why the two behave differently in the load-phase. Maybe a bug in JBoss or Hibernate.