You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@activemq.apache.org by "Thomas Themel (JIRA)" <ji...@apache.org> on 2013/06/19 12:57:19 UTC
[jira] [Updated] (AMQ-4589) Race condition in AjaxListener causes
lost messages
[ https://issues.apache.org/jira/browse/AMQ-4589?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Thomas Themel updated AMQ-4589:
-------------------------------
Attachment: AjaxListener-race.diff
Patch against activemq-parent 5.8.0
> Race condition in AjaxListener causes lost messages
> ---------------------------------------------------
>
> Key: AMQ-4589
> URL: https://issues.apache.org/jira/browse/AMQ-4589
> Project: ActiveMQ
> Issue Type: Bug
> Affects Versions: 5.8.0
> Environment: Jetty 9.0.3, Java 7, Windows
> Reporter: Thomas Themel
> Attachments: AjaxListener-race.diff
>
>
> Running activemq-web under a container that supports Servlet 3, there is a race condition when resuming continuations that leads to lost messages because the same continuation is resumed multiple times and the second time around the undelivered_message attribute is overwritten.
> Here's an example debug log of the issue under Jetty:
> {noformat}
> 2013-06-19 10:46:51,340 DEBUG o.a.a.web.MessageListenerServlet - doMessage timeout=25000
> 2013-06-19 10:46:51,350 DEBUG o.a.a.web.MessageListenerServlet - received null from ActiveMQMessageConsumer { value=ID:deapp0313-52345-1371228095770-3:126:1:1, started=true }
> 2013-06-19 10:46:51,350 DEBUG o.a.a.web.MessageListenerServlet - Suspending continuation org.eclipse.jetty.continuation.Servlet3Continuation@73d4a75a
> 2013-06-19 10:46:53,053 DEBUG org.apache.activemq.web.AjaxListener - message is ActiveMQTextMessage {commandId = 4419, responseRequired = true, messageId = ID:deapp0313-52345-1371228095770-1:1:1:1:4415, originalDestination = null, originalTransactionId = null, producerId = ID:deapp0313-52345-1371228095770-1:1:1:1, destination = topic://W2ETopic, transactionId = null, expiration = 0, timestamp = 1371631612606, arrival = 0, brokerInTime = 1371631612628, brokerOutTime = 1371631612629, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = org.apache.activemq.util.ByteSequence@1e5bc429, dataStructure = null, redeliveryCounter = 0, size = 0, properties = {instance=1796}, readOnlyProperties = true, readOnlyBody = true, droppable = false, text = {"instance":{"instanceId":1796,"changeNo":217...nput":true}}}
> 2013-06-19 10:46:53,062 DEBUG org.apache.activemq.web.AjaxListener - Resuming suspended continuation org.eclipse.jetty.continuation.Servlet3Continuation@73d4a75a
> 2013-06-19 10:46:53,063 DEBUG org.apache.activemq.web.AjaxListener - message for ActiveMQMessageConsumer { value=ID:deapp0313-52345-1371228095770-3:126:1:1, started=true } continuation=org.eclipse.jetty.continuation.Servlet3Continuation@73d4a75a
> 2013-06-19 10:46:53,064 DEBUG org.apache.activemq.web.AjaxListener - message is ActiveMQTextMessage {commandId = 4420, responseRequired = true, messageId = ID:deapp0313-52345-1371228095770-1:1:1:1:4416, originalDestination = null, originalTransactionId = null, producerId = ID:deapp0313-52345-1371228095770-1:1:1:1, destination = topic://W2ETopic, transactionId = null, expiration = 0, timestamp = 1371631612644, arrival = 0, brokerInTime = 1371631612644, brokerOutTime = 1371631613022, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = org.apache.activemq.util.ByteSequence@1aab6f51, dataStructure = null, redeliveryCounter = 0, size = 0, properties = {instance=1796}, readOnlyProperties = true, readOnlyBody = true, droppable = false, text = {"instance":{"instanceId":1796,"changeNo":217...nput":true}}}
> 2013-06-19 10:46:53,064 DEBUG org.apache.activemq.web.AjaxListener - Resuming suspended continuation org.eclipse.jetty.continuation.Servlet3Continuation@73d4a75a
> 2013-06-19 10:46:53,065 DEBUG o.a.a.web.MessageListenerServlet - GET client=org.apache.activemq.web.AjaxWebClient@7c00281 session=c5qseeifcc9l1n49e7ivdx3at clientId=w2e-gui-0.18586184200830758 uri=/amq query=timeout=120000&d=1371631611528&r=0.4084213834721595&clientId=w2e-gui-0.18586184200830758
> 2013-06-19 10:46:53,065 DEBUG o.a.a.web.MessageListenerServlet - doMessage timeout=25000
> 2013-06-19 10:46:53,065 DEBUG o.a.a.web.MessageListenerServlet - sending pre-existing message
> 2013-06-19 10:46:53,066 DEBUG o.a.a.web.MessageListenerServlet - Send 0 unconsumed messages
> 2013-06-19 10:46:53,065 ERROR org.apache.activemq.web.AjaxListener - Error receiving message java.lang.IllegalStateException: REDISPATCHED,resumed
> java.lang.IllegalStateException: REDISPATCHED,resumed
> at org.eclipse.jetty.server.HttpChannelState.dispatch(HttpChannelState.java:335) ~[jetty-server-9.0.3.v20130506.jar:9.0.3.v20130506]
> at org.eclipse.jetty.server.AsyncContextState.dispatch(AsyncContextState.java:109) ~[jetty-server-9.0.3.v20130506.jar:9.0.3.v20130506]
> at org.eclipse.jetty.continuation.Servlet3Continuation.resume(Servlet3Continuation.java:186) ~[jetty-continuation-9.0.3.v20130506.jar:9.0.3.v20130506]
> at org.apache.activemq.web.AjaxListener.onMessageAvailable(AjaxListener.java:74) ~[activemq-web-5.8.0.jar:5.8.0]
> at org.apache.activemq.ActiveMQMessageConsumer.dispatch(ActiveMQMessageConsumer.java:1343) [activemq-client-5.8.0.jar:5.8.0]
> at org.apache.activemq.ActiveMQSessionExecutor.dispatch(ActiveMQSessionExecutor.java:131) [activemq-client-5.8.0.jar:5.8.0]
> at org.apache.activemq.ActiveMQSessionExecutor.iterate(ActiveMQSessionExecutor.java:202) [activemq-client-5.8.0.jar:5.8.0]
> at org.apache.activemq.thread.PooledTaskRunner.runTask(PooledTaskRunner.java:129) [activemq-client-5.8.0.jar:5.8.0]
> at org.apache.activemq.thread.PooledTaskRunner$1.run(PooledTaskRunner.java:47) [activemq-client-5.8.0.jar:5.8.0]
> at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110) [na:1.7.0_07]
> at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603) [na:1.7.0_07]
> at java.lang.Thread.run(Thread.java:722) [na:1.7.0_07]
> 2013-06-19 10:46:53,068 DEBUG o.a.a.web.MessageListenerServlet - Continuation org.eclipse.jetty.continuation.Servlet3Continuation@73d4a75a completed.
> {noformat}
> The root of the race is that the code uses isSuspended() to check whether a continuation has been resumed, but resume() only resets isSuspended() after the resume request has been processed in the servlet engine. The race-free way to check whether is has been resumed is to check isResumed(), see attached patch.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira