ActiveMQ : KahaDB and LevelDB slow disk space expansion

Question

Has anyone observed a creeping disk space growth of their ActiveMQ KahaDB or LevelDB stores ?

We are using the following :

• Windows Server 2008 SP1 (for deployment) and Windows 7 SP1 (for development)
• Java 1.7
• ActiveMQ 5.11.1 and 5.13.2 (tried both KahaDB and LevelDB)
• ServiceMix 5.4.0, with its Karaf server hosting Camel-based Java application beans and its ActiveMQ server unused
• Message rates = 200 XML messages per second coming in and about the same going out (translated into JSON by the Java application beans)
• Number of queues = about 50 queues across which are shared the above messages; 2 queues in particular are handling up to half the traffic
• All messages set to:
  • delivery persistence = false
  • time to live (TTL) = either 1 or 2 minutes (maximum)

KahaDB

When using ActiveMQ with a KahaDB, I have watched the db*.log files initially "leap-frog" from db-1.log to db-2.log then to db-3.log and so on, with the older db*.log files being cleaned up (deleted) as expected. Then at some point may be 10 hours later, the older db*.log files stop getting cleaned up; new db*.log files appear and the KahaDB begins to expand at about 4 GB per day. And when it eventually reaches our configured 50 GB cap, the server of course stops working.

It seems as if a small minority messages are failing to expire (in spite of being given a TTL) and thus block normal clean up. I've tried to manually delete the old db*.log files but I am blocked from doing so with a warning that those files are still in use.

The KahaDB db*.log files are "semi-readable" (!) in a text editor; i.e. I can recognise application messages but this does not tell me much about what has really happened.

LevelDB

I've recently tried using LevelDB with ActiveMQ 5.13.2, and the store still appears to expand albeit much more slowly (about 200 MB per day).

I believe the LevelDB data store uses compression; most files appear to be binary and thus quite opaque without a suitable reader/browser.

Has anyone found or implemented a tool to read/browse the non-persistent messages of a KahaDB or LevelDB message store ?

Our Configuration

Below is what our activemq.xml typically looks like :

<beans
    xmlns="http://www.springframework.org/schema/beans"
    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.xsd
    http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">

    <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="locations">
            <value>file:${activemq.conf}/credentials.properties</value>
        </property>
    </bean>
    <bean id="logQuery" class="io.fabric8.insight.log.log4j.Log4jLogQuery"
        lazy-init="false" scope="singleton"
        init-method="start" destroy-method="stop">
    </bean>

    <broker xmlns="http://activemq.apache.org/schema/core" brokerName="XXXX" dataDirectory="${activemq.data}">

        <destinationPolicy>
            <policyMap>
                <policyEntries>
                    <policyEntry topic=">" >
                        <pendingMessageLimitStrategy>
                            <constantPendingMessageLimitStrategy limit="1000"/>
                        </pendingMessageLimitStrategy>
                    </policyEntry>
                    <policyEntry queue=">" producerFlowControl="true" memoryLimit="1mb">
                        <deadLetterStrategy>
                            <discarding /> 
                        </deadLetterStrategy>
                    </policyEntry>
                </policyEntries>
            </policyMap>
        </destinationPolicy>

        <!--
        <managementContext>
            <managementContext createConnector="false"/>
        </managementContext>
        -->
        <managementContext>
            <managementContext createConnector="true" connectorPort="1099"/>
        </managementContext>

        <!--
        <persistenceAdapter>
            <kahaDB directory="${activemq.data}/kahadb"/>
        </persistenceAdapter>
        -->
        <persistenceAdapter>
            <levelDB directory="${activemq.data}/leveldb"/>
        </persistenceAdapter>

        <systemUsage>
            <systemUsage>
                <memoryUsage>
                    <memoryUsage percentOfJvmHeap="70" />
                </memoryUsage>
                <storeUsage>
                    <storeUsage limit="100 gb"/>
                </storeUsage>
                <tempUsage>
                    <tempUsage limit="50 gb"/>
                </tempUsage>
            </systemUsage>
        </systemUsage>

        <networkConnectors>
            <networkConnector
                    uri="XXX"
                    networkTTL="XXX"
                    duplex="XXX"
                    prefetchSize="XXX"
                    userName="${networkconnector.username}"
                    password="${networkconnector.password}"
                    name="XXX" >
                <excludedDestinations>
                </excludedDestinations>
                <staticallyIncludedDestinations>
                    <queue physicalName="XXX" /> 
                    <queue physicalName="XXX" />
                </staticallyIncludedDestinations>
            </networkConnector>
        </networkConnectors>

        <transportConnectors>
            <transportConnector name="amqp"              uri="amqp://0.0.0.0:XXXX?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="openwire"          uri="tcp://0.0.0.0:XXXX?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="openwire-outbound" uri="tcp://0.0.0.0:XXXX?maximumConnections=1000&amp;wireformat.maxFrameSize=104857600"/>
        </transportConnectors>
        <plugins>
            <simpleAuthenticationPlugin>
                <users>
                    <authenticationUser username="XXX"   password="XXX"   groups="XXX" />
                </users>
            </simpleAuthenticationPlugin>
            <authorizationPlugin>
                <map>
                    <authorizationMap>
                        <authorizationEntries>
                            <authorizationEntry topic=">"                    read="XXX"  write="XXX"  admin="XXX" />
                            <authorizationEntry queue=">"                    read="XXX"  write="XXX"  admin="XXX" />
                            <authorizationEntry queue="XXX"                  read="XXX"  write="XXX"  admin="XXX" />
                            <authorizationEntry queue="ActiveMQ.Advisory.>"  read="XXX"  write="XXX"  admin="XXX" />
                            <authorizationEntry topic="ActiveMQ.Advisory.>"  read="XXX"  write="XXX"  admin="XXX" />
                        </authorizationEntries>
                        <tempDestinationAuthorizationEntry>
                            <tempDestinationAuthorizationEntry read="tempDestinationAdmins" write="tempDestinationAdmins" admin="tempDestinationAdmins" />
                        </tempDestinationAuthorizationEntry>
                    </authorizationMap>
                </map>
            </authorizationPlugin>
            <runtimeConfigurationPlugin checkPeriod="1000" />
        </plugins>
        <shutdownHooks>
            <bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
        </shutdownHooks>
    </broker>
    <import resource="jetty.xml"/>
</beans>