Disk store usage with virtual topics

[Brian Tanner <bt...@edmunds.com>]

Hi, our development team is observing a lot of storage retention when using virtual topics under the following scenario (we are using the default store cursors with kahadb persistence, activemq.xml provided below):

We create one virtual topic with a single queue wired to it.  When we send a batch of messages, with no consumers on the queue and observe a certain amount of store usage, call it 100mb.  We turn on a consumer for the queue and drain the messages. Storage usage falls close to zero.  This makes sense to everybody.  Now we add two more queues to the same virtual topic, remove all consumers, then run another batch of messages.  Store usage spikes up to 300mb.  This suggests each queue gets its own complete copy of the message, which is fine, we can tolerate that.  We turn on consumers for all queues, drain the messages from the queues and our disk store usage returns close to zero.  Everything is as expected.  Now the problem scenario:  We  run a third batch of messages, again with no consumers on.  Disk store goes to 300mb.  Then we turn consumers on for TWO of the THREE queues on the virtual topic.  Two of the queues are drained, but the third queue is still full.  We would expect disk store usage to be 100mb reflecting the one full queue's worth of messages.  Instead disk store stays at 300mb.  

This suggests that when messages are multiplexed through a virtual topic, the disk store will not release space for a message until ALL queues on the virtual topic have consumed this message.  Is this expected behavior?  If so, why does disk store allocate 3x space for 3x queues if they seem to be tied to the same life cycle?  Can anyone help explain?

Our config running on activemq (fuse branch) 5.3.0.5:

<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">
   <!-- Allows us to use system properties as variables in this configuration file -->
   <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="${broker.name}" useJmx="true" dataDirectory="/content/activemq/amq-data/${broker.name}/persist-store-${broker.name}" persistenceAdapter="persist-store">
       <destinationPolicy>
           <policyMap>
               <policyEntries>
                   <policyEntry topic=">" producerFlowControl="false" memoryLimit="5mb" />
                   <policyEntry queue=">" producerFlowControl="false" memoryLimit="5mb" />
               </policyEntries>
           </policyMap>
       </destinationPolicy>
       <managementContext>
           <managementContext connectorPort="1101" rmiServerPort="1100" jmxDomainName="org.apache.activemq"/>
       </managementContext>
       <persistenceAdapter>
            <KahaDB directory="/content/activemq/amq-data/${broker.name}/kahadb-${broker.name}" enableIndexWriteAsync="true" enableJournalDiskSyncs="false" indexWriteBatchSize="1000" checkpointInterval="5000"  cleanupInterval="5000"/>
       </persistenceAdapter>
       <transportConnectors>
           <transportConnector name="nio" uri="nio://0.0.0.0:61616"/>
       </transportConnectors>
       <systemUsage>
         <systemUsage>
             <memoryUsage>
                 <memoryUsage limit="20 mb"/>
             </memoryUsage>
             <storeUsage>
                 <storeUsage limit="4 gb" name="persist-store"/>
             </storeUsage>
             <tempUsage>
                 <tempUsage limit="100 mb"/>
             </tempUsage>
         </systemUsage>
     </systemUsage>
   </broker>
   <import resource="jetty.xml"/>
</beans>