I don't think doDispatch is Queue.java is correct in the trunk.
1. If the send call (or iterate, etc.) is called simultaneously, both might
page in a separate set of messages, as returned by pageIn. Since dispatch is
called immediately, both dispatches are running at the same time and the
messages get out of order.
I think this can be fixed by NOT having pageIn return any messages. Instead,
pageIn should read the messages. Then, it could lock ANOTHER LinkedList
(e.g., dispatchList). While holding that lock, add the messages that have
been read to dispatchList. Do not return anything from pageIn. Call
doDispatch with no parameters. doDispatch locks dispatchList and copies it
into another LinkedList that is a local variable. It then clears
dispatchList and unlocks it. Then, do the dispatching from that local
variable LinkedList.
2. queueMsgConext looks like it is being used in a non-threadsafe manner, if
doDispatch can be simultaneously invoked.
3. The round robin policy is throwing exceptions when there are no
subscribers. I think the try block just needs wrapped with a check if (count
!= 0). One way to trigger is as follows: start two consumers. Then start a
producer. Kill both consumers while the producer is still running.
Maybe I don't fully understand what's going on as I'm just getting started
looking at the code; if so, I apologize and will go back and be more
observant.
Exception Trace for Issue 3:
2006-11-16 22:32:46,765 [ception Handler] ERROR RoundRobinDispatchPolicy
- Caught error rotating consumers
java.lang.IndexOutOfBoundsException: Index: 0, Size: 0
at
edu.emory.mathcs.backport.java.util.concurrent.CopyOnWriteArrayList.remove(CopyOnWriteArrayList.java:346)
at
org.apache.activemq.broker.region.policy.RoundRobinDispatchPolicy.dispatch(RoundRobinDispatchPolicy.java:69)
at
org.apache.activemq.broker.region.Queue.removeSubscription(Queue.java:283)
at
org.apache.activemq.broker.region.AbstractRegion.removeConsumer(AbstractRegion.java:271)
at
org.apache.activemq.broker.region.RegionBroker.removeConsumer(RegionBroker.java:348)
at
org.apache.activemq.broker.BrokerFilter.removeConsumer(BrokerFilter.java:114)
at
org.apache.activemq.advisory.AdvisoryBroker.removeConsumer(AdvisoryBroker.java:210)
at
org.apache.activemq.broker.BrokerFilter.removeConsumer(BrokerFilter.java:114)
at
org.apache.activemq.broker.MutableBrokerFilter.removeConsumer(MutableBrokerFilter.java:124)
at
org.apache.activemq.broker.TransportConnection.processRemoveConsumer(TransportConnection.java:595)
at
org.apache.activemq.broker.TransportConnection.processRemoveSession(TransportConnection.java:632)
at
org.apache.activemq.broker.TransportConnection.processRemoveConnection(TransportConnection.java:713)
at
org.apache.activemq.broker.TransportConnection.stop(TransportConnection.java:887)
at
org.apache.activemq.broker.jmx.ManagedTransportConnection.stop(ManagedTransportConnection.java:74)
at
org.apache.activemq.util.ServiceSupport.dispose(ServiceSupport.java:40)
at
org.apache.activemq.broker.TransportConnection.serviceTransportException(TransportConnection.java:209)
at
org.apache.activemq.broker.TransportConnection.serviceException(TransportConnection.java:241)
at
org.apache.activemq.broker.TransportConnection$2.run(TransportConnection.java:224)
--
View this message in context:
http://www.nabble.com/Message-dispatch-error-tf2649298.html#a7393282
Sent from the ActiveMQ - Dev mailing list archive at Nabble.com.
