Details
-
Bug
-
Status: Resolved
-
Major
-
Resolution: Fixed
-
5.9.1
-
None
-
None
Description
There's a potential for deadlock between transaction completion with optimizeDispatch=true and periodic message expiry, while checkpoint is in progress.
Contention is over pagedInPendingDispatchLock and checkpointLock.
The problem is that we're expiring messages under the pagedInPendingDispatchLock, so the solution is to change this and use the same pattern as we do with pagedInMessages expiry.
Using optimizedDispatch=false or expireMessagesPeriod=0 will workaround the issue.
Relevant thread traces
"ActiveMQ Journal Checkpoint Worker" daemon prio=10 tid=0x00007f0bb8b9d000 nid=0x169b waiting on condition [0x00007f0ba6e95000] java.lang.Thread.State: WAITING (parking) at sun.misc.Unsafe.park(Native Method) - parking to wait for <0x00000000e1c4c4d8> (a java.util.concurrent.locks.ReentrantReadWriteLock$NonfairSync) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186) at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:834) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:867) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1197) at java.util.concurrent.locks.ReentrantReadWriteLock$WriteLock.lock(ReentrantReadWriteLock.java:945) at org.apache.activemq.store.kahadb.MessageDatabase.checkpointUpdate(MessageDatabase.java:1366) at org.apache.activemq.store.kahadb.MessageDatabase.checkpointCleanup(MessageDatabase.java:840) at org.apache.activemq.store.kahadb.MessageDatabase$3.run(MessageDatabase.java:317) ActiveMQ Broker[localhost] Scheduler" daemon prio=10 tid=0x00007f0bbae73800 nid=0x1652 waiting on condition [0x00007f0ba7197000] java.lang.Thread.State: WAITING (parking) at sun.misc.Unsafe.park(Native Method) - parking to wait for <0x00000000e1c4c4d8> (a java.util.concurrent.locks.ReentrantReadWriteLock$NonfairSync) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186) at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:834) at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireShared(AbstractQueuedSynchronizer.java:964) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireShared(AbstractQueuedSynchronizer.java:1282) at java.util.concurrent.locks.ReentrantReadWriteLock$ReadLock.lock(ReentrantReadWriteLock.java:731) at org.apache.activemq.store.kahadb.MessageDatabase.store(MessageDatabase.java:886) at org.apache.activemq.store.kahadb.MessageDatabase.store(MessageDatabase.java:869) at org.apache.activemq.store.kahadb.KahaDBStore$KahaDBMessageStore.removeMessage(KahaDBStore.java:444) at org.apache.activemq.store.kahadb.KahaDBStore$KahaDBMessageStore.removeAsyncMessage(KahaDBStore.java:416) at org.apache.activemq.store.kahadb.KahaDBTransactionStore.removeAsyncMessage(KahaDBTransactionStore.java:495) at org.apache.activemq.store.kahadb.KahaDBTransactionStore$1.removeAsyncMessage(KahaDBTransactionStore.java:182) at org.apache.activemq.broker.region.Queue.acknowledge(Queue.java:850) at org.apache.activemq.broker.region.Queue.removeMessage(Queue.java:1616) at org.apache.activemq.broker.region.Queue.removeMessage(Queue.java:1608) at org.apache.activemq.broker.region.Queue.messageExpired(Queue.java:1674) at org.apache.activemq.broker.region.Queue.messageExpired(Queue.java:1664) at org.apache.activemq.broker.region.Queue.doBrowse(Queue.java:1037) at org.apache.activemq.broker.region.Queue.expireMessages(Queue.java:836) at org.apache.activemq.broker.region.Queue.access$100(Queue.java:98) at org.apache.activemq.broker.region.Queue$2.run(Queue.java:138) at org.apache.activemq.thread.SchedulerTimerTask.run(SchedulerTimerTask.java:33) at java.util.TimerThread.mainLoop(Timer.java:555) at java.util.TimerThread.run(Timer.java:505) "ActiveMQ NIO Worker 76408" daemon prio=10 tid=0x00007f0b98db3800 nid=0x5241 waiting on condition [0x00007f0ba8bb1000] java.lang.Thread.State: WAITING (parking) at sun.misc.Unsafe.park(Native Method) - parking to wait for <0x00000000e1c6ff98> (a java.util.concurrent.locks.ReentrantReadWriteLock$NonfairSync) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186) at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:834) at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireShared(AbstractQueuedSynchronizer.java:964) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireShared(AbstractQueuedSynchronizer.java:1282) at java.util.concurrent.locks.ReentrantReadWriteLock$ReadLock.lock(ReentrantReadWriteLock.java:731) at org.apache.activemq.broker.region.Queue.iterate(Queue.java:1475) - locked <0x00000000e1c70130> (a java.lang.Object) at org.apache.activemq.broker.region.Queue.wakeup(Queue.java:1709) at org.apache.activemq.broker.region.PrefetchSubscription$2.afterCommit(PrefetchSubscription.java:439) at org.apache.activemq.transaction.Transaction.fireAfterCommit(Transaction.java:117) at org.apache.activemq.transaction.Transaction.doPostCommit(Transaction.java:185) at org.apache.activemq.transaction.Transaction$2.call(Transaction.java:54) at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:334) at java.util.concurrent.FutureTask.run(FutureTask.java:166) at org.apache.activemq.store.kahadb.MessageDatabase.process(MessageDatabase.java:1132) at org.apache.activemq.store.kahadb.MessageDatabase$10.visit(MessageDatabase.java:1003) at org.apache.activemq.store.kahadb.data.KahaCommitCommand.visit(KahaCommitCommand.java:130) at org.apache.activemq.store.kahadb.MessageDatabase.process(MessageDatabase.java:985) at org.apache.activemq.store.kahadb.MessageDatabase.store(MessageDatabase.java:892) at org.apache.activemq.store.kahadb.MessageDatabase.store(MessageDatabase.java:869) at org.apache.activemq.store.kahadb.KahaDBTransactionStore.commit(KahaDBTransactionStore.java:294) at org.apache.activemq.transaction.XATransaction.storeCommit(XATransaction.java:85) at org.apache.activemq.transaction.XATransaction.commit(XATransaction.java:75) at org.apache.activemq.broker.TransactionBroker.commitTransaction(TransactionBroker.java:263) at org.apache.activemq.broker.BrokerFilter.commitTransaction(BrokerFilter.java:97) at org.apache.activemq.broker.BrokerFilter.commitTransaction(BrokerFilter.java:97) at org.apache.activemq.broker.BrokerFilter.commitTransaction(BrokerFilter.java:97) at org.apache.activemq.broker.BrokerFilter.commitTransaction(BrokerFilter.java:97) at org.apache.activemq.broker.MutableBrokerFilter.commitTransaction(MutableBrokerFilter.java:103) at org.apache.activemq.broker.TransportConnection.processCommitTransactionTwoPhase(TransportConnection.java:465) at org.apache.activemq.command.TransactionInfo.visit(TransactionInfo.java:102) at org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:329) at org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:184) at org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50) at org.apache.activemq.transport.WireFormatNegotiator.onCommand(WireFormatNegotiator.java:113) at org.apache.activemq.transport.AbstractInactivityMonitor.onCommand(AbstractInactivityMonitor.java:288) at org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:83) at org.apache.activemq.transport.nio.NIOTransport.serviceRead(NIOTransport.java:138) at org.apache.activemq.transport.nio.NIOTransport$1.onSelect(NIOTransport.java:69) at org.apache.activemq.transport.nio.SelectorSelection.onSelect(SelectorSelection.java:94) at org.apache.activemq.transport.nio.SelectorWorker$1.run(SelectorWorker.java:119) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:722)