Mailing-List: contact users-help@activemq.apache.org; run by ezmlm
Precedence: bulk
Reply-To: users@activemq.apache.org
Received-SPF: pass (athena.apache.org: domain of gary.tully@gmail.com
 designates 209.85.216.171 as permitted sender)
DomainKey-Signature: a=rsa-sha1; c=nofws;
        d=gmail.com; s=gamma;
        h=mime-version:in-reply-to:references:date:message-id:subject:from:to
         :content-type:content-transfer-encoding;
        b=N0ak+yF0TOgTc7kHZEbBgDQipyi+T6mgcBaT5zj7LxGgOY2H9uXSwxWDDZotXifb7v
         UWsMYx6kDohh5gzVKA16nFnTgcHQ3aXTl5MYqtPNVxSujA5ZjN3zP8yKWW8ocaOTdv5n
         C/75EZagg3l450FogdyhW1OJj7MsFLt1x1/pQ=
MIME-Version: 1.0
In-Reply-To: <C99BB801.2AB15%anirudha@yahoo-inc.com>
References: <AANLkTinOoZ19r8nO2_+-2LOFSX6zEzw_nuZKdo2ycfU4@mail.gmail.com>
	<C99BB801.2AB15%anirudha@yahoo-inc.com>
Date: Wed, 9 Mar 2011 10:58:15 +0000
Message-ID: <AANLkTikZkOrFT6vEEYF8936_cnB81LoNNV+ww6CoTNvo@mail.gmail.com>
Subject: Re: Message Cursors and Dispatch
From: Gary Tully <gary.tully@gmail.com>
To: users@activemq.apache.org
Content-Type: text/plain; charset=ISO-8859-1
Content-Transfer-Encoding: quoted-printable

you are using a durable subscriber but with non persistent messages.
Because a non persistent message is not in the brokers messages store,
it needs to be locally persisted by the durable subscription cursor,
so it uses the filebased message cursor and replays the messages to
that.
I think what you need is to have non persistent messages use a store
based cursor that is backed by your recovery kahadb instance.

You would need to extend the StoreDurableSubscriberCursor to implement
that (set nonPersistent) and provide a custom
PendingDurableSubscriberMessageStoragePolicy to have it used a
runtime.

Is there a good reason you don't use persistent messages such that a
durable topic subscriber can work with the brokers store.

On 8 March 2011 18:40, Anirudha Khanna <anirudha@yahoo-inc.com> wrote:
> I haven't explicitly configured any cursors on the Topic. From the docume=
ntation it seems like the default is the store based cursor, but when looki=
ng at the ActiveMQ Transport thread stack trace I noticed it was also using=
 the "FilePendingMessageCursor". Can I explicitly configure the broker to o=
nly use the Store based cursor?
>
> attaching a part of the stack trace,
> org.apache.activemq.broker.region.cursors.FilePendingMessageCursor.addMes=
sageLast(FilePendingMessageCursor.java:205)
> =A0 - locked org.apache.activemq.broker.region.cursors.FilePendingMessage=
Cursor@56e48687
> org.apache.activemq.broker.region.cursors.StoreDurableSubscriberCursor.ad=
dRecoveredMessage(StoreDurableSubscriberCursor.java:183)
> =A0 - locked org.apache.activemq.broker.region.cursors.StoreDurableSubscr=
iberCursor@3e539fc
> org.apache.activemq.broker.region.DurableTopicSubscription.doAddRecovered=
Message(DurableTopicSubscription.java:204)
> =A0 - locked org.apache.activemq.broker.region.cursors.StoreDurableSubscr=
iberCursor@3e539fc
> org.apache.activemq.broker.region.AbstractSubscription.addRecoveredMessag=
e(AbstractSubscription.java:174)
> com.yahoo.servinggrid.streaming.replay.ReplaySubscriptionRecoveryPolicy$1=
.recoverMessage(ReplaySubscriptionRecoveryPolicy.java:224)
> org.apache.activemq.store.kahadb.KahaDBStore$KahaDBTopicMessageStore$4.ex=
ecute(KahaDBStore.java:402)
> org.apache.kahadb.page.Transaction.execute(Transaction.java:728)
> org.apache.activemq.store.kahadb.KahaDBStore$KahaDBTopicMessageStore.reco=
verSubscription(KahaDBStore.java:394)
> =A0 - locked java.lang.Object@4edc024a
> com.yahoo.servinggrid.streaming.replay.ReplaySubscriptionRecoveryPolicy.r=
ecover(ReplaySubscriptionRecoveryPolicy.java:214)
>
> My broker.xml
>
> <beans>
> =A0 =A0<broker brokerName=3D"replay-broker" dataDirectory=3D"${activemq.b=
ase}/data"
> =A0 =A0 =A0 =A0useJmx=3D"true" persistent=3D"true" useShutdownHook=3D"fal=
se">
> =A0 =A0 =A0 =A0 =A0 =A0<!-- =A0 =A0 =A0 =A0deleteAllMessagesOnStartup=3D"=
true"> -->
> <destinations>
> =A0 =A0 =A0 =A0<topic physicalName=3D"streaming.replay.topic1" />
> </destinations>
> =A0 <destinationPolicy>
> =A0 =A0 =A0<policyMap>
> =A0 =A0 =A0 =A0<policyEntries>
> =A0 =A0 =A0 =A0 =A0<policyEntry topic=3D"streaming.replay.>">
> =A0 =A0 =A0 =A0 =A0 =A0<subscriptionRecoveryPolicy xmlns:spring=3D"http:/=
/www.springframework.org/schema/beans">
> =A0 =A0 =A0 =A0 =A0 =A0 =A0<spring:bean id=3D"replayPolicy" class=3D"com.=
yahoo.servinggrid.streaming.replay.ReplaySubscriptionRecoveryPolicy">
> =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0<spring:property name=3D"kahadbDir">
> =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0<spring:value>${activemq.base}/data/re=
play-kahadb</spring:value>
> =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0</spring:property>
> =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0<spring:property name=3D"indexCacheSize">
> =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0<spring:value>5000</spring:value>
> =A0 =A0 =A0 =A0 =A0 =A0 =A0</spring:bean>
> =A0 =A0 =A0 =A0 =A0 =A0</subscriptionRecoveryPolicy>
> =A0 =A0 =A0 =A0 =A0</policyEntry>
> =A0 =A0 =A0 =A0</policyEntries>
> =A0 =A0 =A0</policyMap>
> =A0</destinationPolicy>
> =A0 =A0<persistenceAdapter>
> =A0 =A0 =A0 =A0<kahaDB directory=3D"${activemq.base}/data/kahadb"
> =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0enableIndexWriteAsync=3D"true"
> =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0indexCacheSize=3D"5000"
> =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0cleanupInterval=3D"900000"
> =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0checkpointInterval=3D"300000"/>
> =A0 =A0</persistenceAdapter>
> =A0 =A0<transportConnectors>
> =A0 =A0 =A0 =A0<transportConnector name=3D"openwire" uri=3D"tcp://0.0.0.0=
:61616"/>
> =A0 =A0</transportConnectors>
> =A0 </broker>
> </beans>
>
>
> Thanks,
> Anirudha
>
> On 3/8/11 4:34 AM, "Gary Tully" <gary.tully@gmail.com> wrote:
>
> That is expected for the file based cursor. It tries to proxy the
> store, so the entire store is replayed through the cursor to ensure
> correct message order. This behavoour only makes sense if the
> underlying store is very slow, which in your case, kahaDB, is not.
>
> The store based cursor will be a better match a it does not attempt to
> proxy the whole store, rather it reads it in batches and only starts
> to cursor new messages in memory when the store is exhausted.
>
> On 8 March 2011 02:06, Anirudha Khanna <anirudha@yahoo-inc.com> wrote:
>> Hi,
>>
>> So I added a custom subscription recovery policy to ActiveMQ-5.3.2 that =
stores messages in a different instance of Kaha Db. To load test this recov=
ery policy we sent close to ~ 82 million messages(24 hours worth of message=
s) to the topic that has the recovery policy enabled on it.
>> When connecting a single consumer after broker restart, I noticed that t=
he consumer takes a long time to even start getting the messages. Looking a=
t the stack trace it seems that ActiveMQ first attempts to add all the reco=
vered messages to a Message Cursor(in this case a File Message cursor) and =
only then start the dispatch(the dispatch thread is WAITING). Before restar=
ting the broker a single consumer was able to receive messages fairly quick=
ly.
>> Is this the expected behaviour for ActiveMQ, that ALL messages will firs=
t be added to the Message cursor before they are dispatched? Is there a way=
 to get around this?
>>
>> Thanks,
>> Anirudha
>>
>>
>
>
>
> --
> http://blog.garytully.com
> http://fusesource.com
>
>


--=20
http://blog.garytully.com
http://fusesource.com