*Problem:*
I have a clustered activemq failover setup, via zookeeper and leveldb
(configurations are below). My java application (lets call it pipeline)
talks to this cluster via apache camel. Pipeline has both producer and
consumer endpoints. (About 10 different queue destinations).
In a non-clustered setup, where activemq is running local to pipeline,
things are fine. But in a clustered activemq setup, the activemq master
functions fine for a while, after which it stops responding to "send
messages". I used a sample java application too to connect to this cluster
after it has reached this "hung" state. The connection is successful, but
"send message" times out with following exception:
javax.jms.JMSException:
org.apache.activemq.transport.RequestTimedOutIOException
at
org.apache.activemq.util.JMSExceptionSupport.create(JMSExceptionSupport.java:62)
at
org.apache.activemq.ActiveMQConnection.doSyncSendPacket(ActiveMQConnection.java:1431)
at
org.apache.activemq.ActiveMQConnection.close(ActiveMQConnection.java:675)
I suspect, after a while the master is not able to replicate the incoming
messages across the slave nodes.
The behaviour is intermittent and mostly happens after a good number of
messages (unable to determine the exact threshold) have been sent to this
cluster via the application. Kind of messages? Small in size, containing a
small string or a flag at max.
Also, activemq logs are of no help. But i have provided relevant parts of
activemq config and thread dump (master) below.
*Following are details pertaining to my setup: *
Zookeeper version: 3.3.6
ActiveMQ version: 5.9.0
Topology:
(each box below is a physical machine)
| zookeeper1 | --- | zookeeper2 | -- | zookeeper3 |
| activemq1 | | activemq2 | | activemq3 |
*Activemq.xml (relevant parts) (same - except the host names - for all the
three instances of activemq):*
<persistenceAdapter>
<replicatedLevelDB
directory="${activemq.data}"
replicas="3"
bind="tcp://0.0.0.0:0"
zkAddress="ccles1.ccllocal.com:2181,ccles2.ccllocal.com:2181,ccles3.ccllocal.com:2181"
zkPath="/activemq/leveldb-stores"
hostname="ccles3.ccllocal.com"
/>
</persistenceAdapter>
*Thread dump from master (important parts):*
Thread 1460: (state = BLOCKED)
-
org.apache.activemq.leveldb.LevelDBStore.createQueueMessageStore(org.apache.activemq.command.ActiveMQQueue)
@bci=4, line=571 (Interpreted frame)
-
org.apache.activemq.leveldb.replicated.ProxyLevelDBStore.createQueueMessageStore(org.apache.activemq.command.ActiveMQQueue)
@bci=5, line=96 (Interpreted frame)
-
org.apache.activemq.broker.region.DestinationFactoryImpl.createDestination(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ActiveMQDestination,
org.apache.activemq.broker.region.DestinationStatistics) @bci=63, line=84
(Interpreted frame)
-
org.apache.activemq.broker.region.AbstractRegion.createDestination(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ActiveMQDestination) @bci=10, line=512
(Interpreted frame)
-
org.apache.activemq.broker.jmx.ManagedQueueRegion.createDestination(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ActiveMQDestination) @bci=3, line=56
(Interpreted frame)
-
org.apache.activemq.broker.region.AbstractRegion.addDestination(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ActiveMQDestination, boolean) @bci=62, line=134
(Interpreted frame)
-
org.apache.activemq.broker.region.RegionBroker.addDestination(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ActiveMQDestination, boolean) @bci=167, line=322
(Interpreted frame)
-
org.apache.activemq.broker.BrokerFilter.addDestination(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ActiveMQDestination, boolean) @bci=7, line=167
(Interpreted frame)
-
org.apache.activemq.advisory.AdvisoryBroker.addDestination(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ActiveMQDestination, boolean) @bci=4, line=185
(Interpreted frame)
-
org.apache.activemq.broker.BrokerFilter.addDestination(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ActiveMQDestination, boolean) @bci=7, line=167
(Interpreted frame)
-
org.apache.activemq.broker.BrokerFilter.addDestination(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ActiveMQDestination, boolean) @bci=7, line=167
(Interpreted frame)
-
org.apache.activemq.broker.MutableBrokerFilter.addDestination(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ActiveMQDestination, boolean) @bci=7, line=172
(Interpreted frame)
-
org.apache.activemq.broker.region.RegionBroker.addProducer(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ProducerInfo) @bci=29, line=372 (Interpreted
frame)
-
org.apache.activemq.broker.jmx.ManagedRegionBroker.addProducer(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ProducerInfo) @bci=3, line=253 (Interpreted
frame)
-
org.apache.activemq.broker.BrokerFilter.addProducer(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ProducerInfo) @bci=6, line=102 (Interpreted
frame)
-
org.apache.activemq.advisory.AdvisoryBroker.addProducer(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ProducerInfo) @bci=3, line=173 (Interpreted
frame)
-
org.apache.activemq.broker.CompositeDestinationBroker.addProducer(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ProducerInfo) @bci=76, line=56 (Interpreted
frame)
-
org.apache.activemq.broker.BrokerFilter.addProducer(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ProducerInfo) @bci=6, line=102 (Interpreted
frame)
-
org.apache.activemq.broker.MutableBrokerFilter.addProducer(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.ProducerInfo) @bci=6, line=107 (Interpreted
frame)
-
org.apache.activemq.broker.TransportConnection.processAddProducer(org.apache.activemq.command.ProducerInfo)
@bci=195, line=534 (Interpreted frame)
-
org.apache.activemq.command.ProducerInfo.visit(org.apache.activemq.state.CommandVisitor)
@bci=2, line=105 (Interpreted frame)
-
org.apache.activemq.broker.TransportConnection.service(org.apache.activemq.command.Command)
@bci=41, line=292 (Interpreted frame)
-
org.apache.activemq.broker.TransportConnection$1.onCommand(java.lang.Object)
@bci=60, line=149 (Interpreted frame)
- org.apache.activemq.transport.MutexTransport.onCommand(java.lang.Object)
@bci=52, line=50 (Interpreted frame)
-
org.apache.activemq.transport.WireFormatNegotiator.onCommand(java.lang.Object)
@bci=29, line=113 (Interpreted frame)
-
org.apache.activemq.transport.AbstractInactivityMonitor.onCommand(java.lang.Object)
@bci=156, line=270 (Interpreted frame)
-
org.apache.activemq.transport.TransportSupport.doConsume(java.lang.Object)
@bci=16, line=83 (Interpreted frame)
- org.apache.activemq.transport.tcp.TcpTransport.doRun() @bci=7, line=214
(Interpreted frame)
- org.apache.activemq.transport.tcp.TcpTransport.run() @bci=47, line=196
(Interpreted frame)
- java.lang.Thread.run() @bci=11, line=744 (Interpreted frame)
Thread 13929: (state = BLOCKED)
- sun.misc.Unsafe.park(boolean, long) @bci=0 (Compiled frame; information
may be imprecise)
- java.util.concurrent.locks.LockSupport.park(java.lang.Object) @bci=14,
line=186 (Interpreted frame)
-
java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt()
@bci=1, line=834 (Interpreted frame)
-
java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(int)
@bci=72, line=994 (Interpreted frame)
-
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(int)
@bci=24, line=1303 (Interpreted frame)
- java.util.concurrent.CountDownLatch.await() @bci=5, line=236 (Interpreted
frame)
-
org.apache.activemq.leveldb.LevelDBStore.commit(org.apache.activemq.command.TransactionId,
boolean, java.lang.Runnable, java.lang.Runnable) @bci=105, line=473
(Interpreted frame)
- org.apache.activemq.transaction.LocalTransaction.commit(boolean)
@bci=140, line=70 (Interpreted frame)
-
org.apache.activemq.broker.TransactionBroker.commitTransaction(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.TransactionId, boolean) @bci=12, line=253
(Interpreted frame)
-
org.apache.activemq.broker.MutableBrokerFilter.commitTransaction(org.apache.activemq.broker.ConnectionContext,
org.apache.activemq.command.TransactionId, boolean) @bci=7, line=112
(Interpreted frame)
-
org.apache.activemq.broker.TransportConnection.processCommitTransactionOnePhase(org.apache.activemq.command.TransactionInfo)
@bci=39, line=424 (Interpreted frame)
-
org.apache.activemq.command.TransactionInfo.visit(org.apache.activemq.state.CommandVisitor)
@bci=78, line=100 (Interpreted frame)
-
org.apache.activemq.broker.TransportConnection.service(org.apache.activemq.command.Command)
@bci=41, line=292 (Interpreted frame)
-
org.apache.activemq.broker.TransportConnection$1.onCommand(java.lang.Object)
@bci=60, line=149 (Interpreted frame)
- org.apache.activemq.transport.MutexTransport.onCommand(java.lang.Object)
@bci=52, line=50 (Interpreted frame)
-
org.apache.activemq.transport.WireFormatNegotiator.onCommand(java.lang.Object)
@bci=29, line=113 (Compiled frame)
-
org.apache.activemq.transport.AbstractInactivityMonitor.onCommand(java.lang.Object)
@bci=156, line=270 (Compiled frame)
-
org.apache.activemq.transport.TransportSupport.doConsume(java.lang.Object)
@bci=16, line=83 (Compiled frame)
- org.apache.activemq.transport.tcp.TcpTransport.doRun() @bci=7, line=214
(Compiled frame)
- org.apache.activemq.transport.tcp.TcpTransport.run() @bci=47, line=196
(Compiled frame)
- java.lang.Thread.run() @bci=11, line=744 (Interpreted frame)
Thread 10888: (state = BLOCKED)
- org.apache.activemq.leveldb.LevelDBStore.getTopicGCPositions() @bci=4,
line=768 (Interpreted frame)
- org.apache.activemq.leveldb.DBManager$$anonfun$pollGc$1.apply$mcV$sp()
@bci=17, line=640 (Interpreted frame)
- org.fusesource.hawtdispatch.package$$anon$4.run() @bci=4, line=357
(Interpreted frame)
- org.fusesource.hawtdispatch.internal.SerialDispatchQueue.run() @bci=299,
line=100 (Compiled frame)
- org.fusesource.hawtdispatch.internal.pool.SimpleThread.run() @bci=82,
line=77 (Interpreted frame)
PS: repost, because the earlier was posted w/o me being subscribed
--
View this message in context:
http://activemq.2283324.n4.nabble.com/Clustered-Activemq-replicated-LevelDB-via-zookeeper-apache-camel-causes-activemq-to-hang-tp4681427.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.
