Recap
In my previous post, I explained the benefits of using explicit optimistic locking. As we then discovered, there’s a very short time window in which a concurrent transaction can still commit a Product price change right before our current transaction gets committed.
This issue can be depicted as follows:
- Alice fetches a Product
- She then decides to order it
- The Product optimistic lock is acquired
- The Order is inserted in the current transaction database session
- The Product version is checked by the Hibernate explicit optimistic locking routine
- The price engine manages to commit the Product price change
- Alice transaction is committed without realizing the Product price has just changed
Replicating the issue
So we need a way to inject the Product price change in between the optimistic lock check and the order transaction commit.
After analyzing the Hibernate source code, we discover that the SessionImpl.beforeTransactionCompletion() method is calling the current configured Interceptor.beforeTransactionCompletion() callback, right after the internal actionQueue stage handler (where the explicit optimistic locked entity version is checked):
public void beforeTransactionCompletion(TransactionImplementor hibernateTransaction) {
LOG.trace( "before transaction completion" );
actionQueue.beforeTransactionCompletion();
try {
interceptor.beforeTransactionCompletion( hibernateTransaction );
}
catch (Throwable t) {
LOG.exceptionInBeforeTransactionCompletionInterceptor( t );
}
}Armed with this info, we can set-up a test to replicate our race condition:
private AtomicBoolean ready = new AtomicBoolean();
private final CountDownLatch endLatch = new CountDownLatch(1);
@Override
protected Interceptor interceptor() {
return new EmptyInterceptor() {
@Override
public void beforeTransactionCompletion(Transaction tx) {
if(ready.get()) {
LOGGER.info("Overwrite product price asynchronously");
executeNoWait(new Callable<Void>() {
@Override
public Void call() throws Exception {
Session _session = getSessionFactory().openSession();
_session.doWork(new Work() {
@Override
public void execute(Connection connection) throws SQLException {
try(PreparedStatement ps = connection.prepareStatement("UPDATE product set price = 14.49 WHERE id = 1")) {
ps.executeUpdate();
}
}
});
_session.close();
endLatch.countDown();
return null;
}
});
try {
LOGGER.info("Wait 500 ms for lock to be acquired!");
Thread.sleep(500);
} catch (InterruptedException e) {
throw new IllegalStateException(e);
}
}
}
};
}
@Test
public void testExplicitOptimisticLocking() throws InterruptedException {
try {
doInTransaction(new TransactionCallable<Void>() {
@Override
public Void execute(Session session) {
try {
final Product product = (Product) session.get(Product.class, 1L, new LockOptions(LockMode.OPTIMISTIC));
OrderLine orderLine = new OrderLine(product);
session.persist(orderLine);
lockUpgrade(session, product);
ready.set(true);
} catch (Exception e) {
throw new IllegalStateException(e);
}
return null;
}
});
} catch (OptimisticEntityLockException expected) {
LOGGER.info("Failure: ", expected);
}
endLatch.await();
}
protected void lockUpgrade(Session session, Product product) {}When running it, the test generates the following output:
#Alice selects a Product
DEBUG [main]: Query:{[select abstractlo0_.id as id1_1_0_, abstractlo0_.description as descript2_1_0_, abstractlo0_.price as price3_1_0_, abstractlo0_.version as version4_1_0_ from product abstractlo0_ where abstractlo0_.id=?][1]}
#Alice inserts an OrderLine
DEBUG [main]: Query:{[insert into order_line (id, product_id, unitPrice, version) values (default, ?, ?, ?)][1,12.99,0]}
#Alice transaction verifies the Product version
DEBUG [main]: Query:{[select version from product where id =?][1]}
#The price engine thread is started
INFO [main]: c.v.h.m.l.c.LockModeOptimisticRaceConditionTest - Overwrite product price asynchronously
#Alice thread sleeps for 500ms to give the price engine a chance to execute its transaction
INFO [main]: c.v.h.m.l.c.LockModeOptimisticRaceConditionTest - Wait 500 ms for lock to be acquired!
#The price engine changes the Product price
DEBUG [pool-1-thread-1]: Query:{[UPDATE product set price = 14.49 WHERE id = 1][]}
#Alice transaction is committed without realizing the Product price change
DEBUG [main]: o.h.e.t.i.j.JdbcTransaction - committed JDBC ConnectionSo, the race condition is real. It’s up to you to decide if your current application demands stronger data integrity requirements, but as rule of thumb, better safe than sorry.
Fixing the issue
To fix this issue, we just need to add a pessimistic lock request just before ending our transactional method.
@Override
protected void lockUpgrade(Session session, Product product) {
session.buildLockRequest(new LockOptions(LockMode.PESSIMISTIC_READ)).lock(product);
}The explicit shared lock will prevent concurrent writes on the entity we’ve previously locked optimistically. With this method, no other concurrent transaction can change the Product prior to releasing this lock (after the current transaction is committed or rolled back).
With the new pessimistic lock request in place, the previous test generates the following output:
#Alice selects a Product
DEBUG [main]: Query:{[select abstractlo0_.id as id1_1_0_, abstractlo0_.description as descript2_1_0_, abstractlo0_.price as price3_1_0_, abstractlo0_.version as version4_1_0_ from product abstractlo0_ where abstractlo0_.id=?][1]}
#Alice inserts an OrderLine
DEBUG [main]: Query:{[insert into order_line (id, product_id, unitPrice, version) values (default, ?, ?, ?)][1,12.99,0]}
#Alice applies an explicit physical lock on the Product entity
DEBUG [main]: Query:{[select id from product where id =? and version =? for update][1,0]}
#Alice transaction verifies the Product version
DEBUG [main]: Query:{[select version from product where id =?][1]}
#The price engine thread is started
INFO [main]: c.v.h.m.l.c.LockModeOptimisticRaceConditionTest - Overwrite product price asynchronously
#Alice thread sleeps for 500ms to give the price engine a chance to execute its transaction
INFO [main]: c.v.h.m.l.c.LockModeOptimisticRaceConditionTest - Wait 500 ms for lock to be acquired!
#The price engine cannot proceed because of the Product entity was locked exclusively, so Alice transaction is committed against the ordered Product price
DEBUG [main]: o.h.e.t.i.j.JdbcTransaction - committed JDBC Connection
#The physical lock is released and the price engine can change the Product price
DEBUG [pool-1-thread-1]: Query:{[UPDATE product set price = 14.49 WHERE id = 1][]}Even though we asked for a PESSIMISTIC_READ lock, HSQLDB can only execute a FOR UPDATE exclusive lock instead, equivalent to an explicit PESSIMISTIC_WRITE lock mode.
Conclusion
If you wonder why we use both optimistic and pessimistic locking for our current transaction, you must remember that optimistic locking is the only feasible concurrency control mechanism for multi-request conversations.
In our example, The Product entity is loaded by the first request, using a read-only transaction. The Product entity has an associated version, and this read-time entity snapshot is going to be locked optimistically during the write-time transaction.
The pessimistic lock is useful only during the write-time transaction, to prevent any concurrent update from occurring after the Product entity version check. So, both the logical lock and the physical lock are cooperating for ensuring the Order price data integrity.
While I was working on this blog post, the Java Champion Markus Eisele took me an interview about the Hibernate Master Class initiative. During the interview I tried to explain the current post examples, while emphasizing the true importance of knowing your tools beyond the reference documentation.
- Code available on GitHub.
| Reference: | How to fix optimistic locking race conditions with pessimistic locking from our JCG partner Vlad Mihalcea at the Vlad Mihalcea’s Blog blog. |
