Hi all,
I'm trying to setup a pure master slave configuration with a
persistent queue, but the moment I put some load on it (about 30 JMS
text msg/sec, no transactions), I get these kinds of errors in the
master:
2011-05-24 21:36:37,008 | ERROR | Slave Failed |
org.apache.activemq.broker.ft.MasterBroker | BrokerService[localhost]
Task-36
javax.jms.JMSException: Slave broker out of sync with master -
Message: ID:slavehost-35023-1306272591891-1:11:10878:1:1 on
queue://test does not exist among pending(200) for subscription:
ID:slavehost-35023-1306272591891-1:3:2:1
at
org.apache.activemq.broker.region.Queue.getMatchingMessage(Queue.java:2044)
at
org.apache.activemq.broker.region.Queue.processDispatchNotification(Queue.java:1981)
at
org.apache.activemq.broker.region.AbstractRegion.processDispatchNotificationViaDestination(AbstractRegion.java:447)
at
org.apache.activemq.broker.region.QueueRegion.processDispatchNotification(QueueRegion.java:91)
at
org.apache.activemq.broker.region.RegionBroker.processDispatchNotification(RegionBroker.java:698)
at
org.apache.activemq.broker.BrokerFilter.processDispatchNotification(BrokerFilter.java:205)
at
org.apache.activemq.broker.BrokerFilter.processDispatchNotification(BrokerFilter.java:205)
at
org.apache.activemq.broker.BrokerFilter.processDispatchNotification(BrokerFilter.java:205)
at
org.apache.activemq.broker.BrokerFilter.processDispatchNotification(BrokerFilter.java:205)
at
org.apache.activemq.broker.MutableBrokerFilter.processDispatchNotification(MutableBrokerFilter.java:211)
at
org.apache.activemq.broker.TransportConnection.processMessageDispatchNotification(TransportConnection.java:474)
at
org.apache.activemq.command.MessageDispatchNotification.visit(MessageDispatchNotification.java:77)
at
org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:306)
at
org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:179)
at
org.apache.activemq.transport.ResponseCorrelator.onCommand(ResponseCorrelator.java:116)
at
org.apache.activemq.transport.TransportFilter.onCommand(TransportFilter.java:69)
at
org.apache.activemq.transport.vm.VMTransport.iterate(VMTransport.java:218)
at
org.apache.activemq.thread.PooledTaskRunner.runTask(PooledTaskRunner.java:127)
at
org.apache.activemq.thread.PooledTaskRunner$1.run(PooledTaskRunner.java:48)
at
java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:619)
and
2011-05-24 21:32:28,349 | ERROR | Slave Failed |
org.apache.activemq.broker.ft.MasterBroker | ActiveMQ Transport:
tcp:///slavehost:35029
javax.jms.JMSException: Unmatched acknowledge: MessageAck {commandId =
51199, responseRequired = true, ackType = 2, consumerId =
ID:slavehost-35023-1306272591891-1:8:2:1, firstMessageId =
ID:slavehost-35023-1306272591891-1:12:2420:1:1, lastMessageId =
ID:slavehost-35023-1306272591891-1:12:2420:1:1, destination =
queue://test, transactionId =
TX:ID:slavehost-35023-1306272591891-1:8:332, messageCount = 1,
poisonCause = null};
Could not find Message-ID
ID:slavehost-35023-1306272591891-1:12:2420:1:1 in dispatched-list
(start of ack)
at
org.apache.activemq.broker.region.PrefetchSubscription.assertAckMatchesDispatched(PrefetchSubscription.java:439)
at
org.apache.activemq.broker.region.PrefetchSubscription.acknowledge(PrefetchSubscription.java:206)
at
org.apache.activemq.broker.region.AbstractRegion.acknowledge(AbstractRegion.java:383)
at
org.apache.activemq.broker.region.RegionBroker.acknowledge(RegionBroker.java:549)
at
org.apache.activemq.broker.BrokerFilter.acknowledge(BrokerFilter.java:77)
at
org.apache.activemq.broker.BrokerFilter.acknowledge(BrokerFilter.java:77)
at
org.apache.activemq.broker.BrokerFilter.acknowledge(BrokerFilter.java:77)
at
org.apache.activemq.broker.TransactionBroker.acknowledge(TransactionBroker.java:197)
at
org.apache.activemq.broker.MutableBrokerFilter.acknowledge(MutableBrokerFilter.java:87)
at
org.apache.activemq.broker.TransportConnection.processMessageAck(TransportConnection.java:465)
at org.apache.activemq.command.MessageAck.visit(MessageAck.java:229)
at
org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:306)
at
org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:179)
at
org.apache.activemq.transport.ResponseCorrelator.onCommand(ResponseCorrelator.java:116)
at
org.apache.activemq.transport.TransportFilter.onCommand(TransportFilter.java:69)
at
org.apache.activemq.transport.vm.VMTransport.iterate(VMTransport.java:218)
at
org.apache.activemq.thread.PooledTaskRunner.runTask(PooledTaskRunner.java:127)
at
org.apache.activemq.thread.PooledTaskRunner$1.run(PooledTaskRunner.java:48)
at
java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:619)
The EnqueueCount attribute of the queue in JMX shows 0 for the slave
which seems to indicate that no message made it to the slave. The
master config is as follows (slave is the same with only the
additional masterConnectorURI property set):
<beans
xmlns="http://www.springframework.org/schema/beans";
xmlns:amq="http://activemq.apache.org/schema/core";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd";>
<bean
class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="locations">
<value>file:${activemq.base}/conf/credentials.properties</value>
</property>
</bean>
<broker xmlns="http://activemq.apache.org/schema/core";
brokerName="localhost"
dataDirectory="${activemq.base}/data"
destroyApplicationContextOnStop="true"
useJmx="true"
dedicatedTaskRunner="false"
enableStatistics="true"
schedulerSupport="true">
<destinationPolicy>
<policyMap>
<policyEntries>
<policyEntry queue=">" producerFlowControl="false"
memoryLimit="1mb">
<pendingQueuePolicy>
<vmQueueCursor/>
</pendingQueuePolicy>
</policyEntry>
</policyEntries>
</policyMap>
</destinationPolicy>
<destinations>
<queue physicalName="test" />
</destinations>
<managementContext>
<managementContext createConnector="true"/>
</managementContext>
<persistenceAdapter>
<kahaDB directory="${activemq.base}/data/kahadb"/>
</persistenceAdapter>
<systemUsage>
<systemUsage>
<memoryUsage>
<memoryUsage limit="1 gb"/>
</memoryUsage>
<storeUsage>
<storeUsage limit="100 gb"/>
</storeUsage>
<tempUsage>
<tempUsage limit="10 gb"/>
</tempUsage>
</systemUsage>
</systemUsage>
<transportConnectors>
<transportConnector name="openwire" uri="tcp://0.0.0.0:35000"/>
<transportConnector name="stomp" uri="stomp://0.0.0.0:61613"/>
</transportConnectors>
</broker>
<import resource="jetty.xml"/>
</beans>
The brokers are run with these system properties
-Xms2048M -Xmx2048M -Dorg.apache.activemq.UseDedicatedTaskRunner=false
and I cleaned the data directories on both master and slave before
starting the test.
Any idea what could be wrong ?
cheers,
Tom
