activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Christian Posta <christian.po...@gmail.com>
Subject Re: persistent message missing to a durable subscriber when it reconnects(restarts)
Date Tue, 26 Mar 2013 14:05:46 GMT
Can you give it a try on a recent snapshot?
https://repository.apache.org/content/repositories/snapshots/org/apache/activemq/apache-activemq/5.9-SNAPSHOT/



On Mon, Mar 25, 2013 at 11:39 PM, Jaewoong Choi <jaewoong.choi@ymail.com>wrote:

> Hi,
>
> I encountered a problem that a persistence message is not dispatched but
> missing to a durable subscriber, especially when the subscriber
> reconnects(restarts).  Here's the test environment, the scenario and the
> problem.
>
> *test env*
> ActiveMQ 5.7.0/jdk 1.6
> All configurations are not touched to be default but only persistent is
> disabled: BrokerService#setPersistent(false)
>
> *test purpose*
> test how persistent messages are dispatched properly to a durable
> subscriber
> even when the durable subscriber has intermittent connection situation.
>
> *test scenario*
> 1. A durable subscriber is created to a topic.
> 2. A publisher is created and publishes 30 (m1,m2 .. m30) messages to a
> topic.
> 3. The durable subscriber reconnects/resume every after consuming 2
> messages.  So, it's supposed to reconnect 15 times until it completes
> consuming all 30 messages.
>
> *problem*
> The durable subscriber is missing a message sporadically (at step 3 above).
> For example, it receives m6 after m4 (m5 is missing).
>
> Per the trace level logs from org.apache.activemq.broker.region.cursors
> package (see below), looks like the deactivating thread (by
> javax.jms.MessageConsumer#close) doesn't move pending messages back to
> pending list properly but does only inflight messages (at
> DurableTopicSubscription#deactivate), especially when
> DurableTopicSubscription#dispatchPending is called by publishing thread
> simultaneously.
>
> *trace log*
> -----------------
> /### [publisher connection] a message (ID:27) is published and cached
> successfully to be directly dispatched./
> 20:46:59.881 [ActiveMQ Transport: tcp:///10.73.199.7:49660@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=false,storeHasMessages=true,size=0,cacheEnabled=true,maxB
> atchSize:16 - addMessageLast: mid=ID:<producer_id>:27
> 20:46:59.881 [ActiveMQ Transport: tcp:///10.73.199.7:49660@61616] TRACE
> o.a.a.b.r.cursors.TopicStorePrefetch - recover: ID:<producer_id>:27,
> priority: 4
>
> /### [subscriber connection] subscriber stop (by
> javax.jms.MessageConsumer#close) kicked in./
> 20:46:59.881 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.DurableTopicSubscription   - Deactivating keepActive=true,
> DurableTopicSubscription-cli-0:sub-0, id=ID:<consumer_id>, active=true,
> destinations=1, total=27, pending=1, dispatched=53, inflight=15,
> prefetchExtension=0
>
> /### [publisher connection] the cached message (ID:27) was removed from
> pending list and dispatched./
> 20:46:59.881 [ActiveMQ Transport: tcp:///10.73.199.7:49660@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=false,storeHasMessages=true,size=1,cacheEnabled=true,maxB
> atchSize:16 - remove()
> 20:46:59.881 [ActiveMQ Transport: tcp:///10.73.199.7:49660@61616] TRACE
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=false,storeHasMessages=true,size=0,cacheEnabled=true,maxB
> atchSize:16 - fillBatch
>
> /### [subscriber connection] consumer deactivation only moved 15 inflight
> messages (ID:12 - ID:26) back to pending list but missed (ID:27) which is
> just dispatched./
> 20:46:59.881 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=0,cacheEnabled=false,maxB
> atchSize:16 - addMessageFirst(mid=ID:<producer_id>:26)
> 20:46:59.881 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=1,cacheEnabled=false,maxB
> atchSize:16 - addMessageFirst(mid=ID:<producer_id>:25)
> 20:46:59.881 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=2,cacheEnabled=false,maxB
> atchSize:16 - addMessageFirst(mid=ID:<producer_id>:24)
> 20:46:59.881 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=3,cacheEnabled=false,maxB
> atchSize:16 - addMessageFirst(mid=ID:<producer_id>:23)
> 20:46:59.881 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=4,cacheEnabled=false,maxB
> atchSize:16 - addMessageFirst(mid=ID:<producer_id>:22)
> 20:46:59.881 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=5,cacheEnabled=false,maxB
> atchSize:16 - addMessageFirst(mid=ID:<producer_id>:21)
> 20:46:59.881 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=6,cacheEnabled=false,maxB
> atchSize:16 - addMessageFirst(mid=ID:<producer_id>:20)
> 20:46:59.882 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=7,cacheEnabled=false,maxB
> atchSize:16 - addMessageFirst(mid=ID:<producer_id>:19)
> 20:46:59.882 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=8,cacheEnabled=false,maxB
> atchSize:16 - addMessageFirst(mid=ID:<producer_id>:18)
> 20:46:59.882 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=9,cacheEnabled=false,maxB
> atchSize:16 - addMessageFirst(mid=ID:<producer_id>:17)
> 20:46:59.882 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=10,cacheEnabled=false,max
> BatchSize:16 - addMessageFirst(mid=ID:<producer_id>:16)
> 20:46:59.882 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=11,cacheEnabled=false,max
> BatchSize:16 - addMessageFirst(mid=ID:<producer_id>:15)
> 20:46:59.882 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=12,cacheEnabled=false,max
> BatchSize:16 - addMessageFirst(mid=ID:<producer_id>:14)
> 20:46:59.882 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=13,cacheEnabled=false,max
> BatchSize:16 - addMessageFirst(mid=ID:<producer_id>:13)
> 20:46:59.882 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.b.r.c.AbstractStoreCursor      - TopicStorePrefetch(cli-0,sub-0)
> ID:<consumer_id> -
> org.apache.activemq.broker.region.cursors.TopicStorePrefetch@1d667df0
> :<topic_name>,batchResetNeeded=true,storeHasMessages=true,size=14,cacheEnabled=false,max
> BatchSize:16 - addMessageFirst(mid=ID:<producer_id>:12)
> 20:46:59.882 [ActiveMQ Transport: tcp:///10.73.199.7:49670@61616] DEBUG
> o.a.a.broker.region.AbstractRegion   - broker-tcp:__weehomespent-lm:61616
> removing consumer: ID:<consumer_id> for destination:
> ActiveMQ.Advisory.TempQueue,ActiveMQ.Advisory.TempTopic
> -----------------
>
>
> I'll be greatly appreciated if any user or activemq developer can help
> figure out if this is a bug, or let me know if I'm missing something.
>
> Regards,
> Jaewoong
>
>
>
> --
> View this message in context:
> http://activemq.2283324.n4.nabble.com/persistent-message-missing-to-a-durable-subscriber-when-it-reconnects-restarts-tp4665130.html
> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
>



-- 
*Christian Posta*
http://www.christianposta.com/blog
twitter: @christianposta

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message