activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Matt Henkel <guildencra...@gmail.com>
Subject NetworkConnector and VirtualTopic delivery issues
Date Sat, 15 Dec 2012 00:32:33 GMT
We've got a cluster of two brokers (5.7.0) running (same 
configuration--puppet deployed--on both boxes except for the brokerName) 
in our production environment and most of the VirtualTopics and queues 
appear to be fine, however one broker (amq00) is having problems.

We have the VirtualTopic.nonBlockingMessageFile which has a few 
Consumer.*.VirtualTopic.nonBlockingMessageFile queues consuming from it. 
If I use the web admin and "Send To" the VirtualTopic or a consuming 
queue that message is delivered to the queue(s) and processed as 
expected. However when our producers try to produce to the topic 
VirtualTopic.nonBlockingMessageFile the topic will be created, however 
nothing is ever enqueued. This is all on amq00. On amq01 the queue is 
created and messages are enqueued, then delivered to the consuming 
queues. Our producers are using the failover protocol and do switch 
between the brokers as we start and stop them.

The other issue is that if we pin some consumers to a 
Consumer.foo.VirtualTopic.nonBlockingMessageFile queue on amq00 and turn 
off any consumers on the same queue on amq01 I can see messages being 
delivered to the networkConnector, however I never see them enqueued or 
processed through amq00 (we have a number of networkConnectors 
configured based on the consumer process priority so we actually exclude 
the VirtualTopic and explicitly include the queues).

I've let all the queues drain, deleted all the queues (even trashed the 
whole kahadb directory), and bounced the broker only to have this 
problem persist.  Any ideas what's going on? As mentioned before amq01 
is working fine, and we can't reproduce this in our test environment.

Here's our activemq.xml:

    <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.conf}/credentials.properties</value>
             </property>
         </bean>

         <!--
             The <broker> element is used to configure the ActiveMQ broker.
         -->
         <broker xmlns="http://activemq.apache.org/schema/core"
    brokerName="amq00.lan" dataDirectory="${activemq.data}">

             <!--
                 For better performances use VM cursor and small memory
    limit.
                 For more information, see:

                 http://activemq.apache.org/message-cursors.html

                 Also, if your producer is "hanging", it's probably due
    to producer flow control.
                 For more information, see:
                 http://activemq.apache.org/producer-flow-control.html
             -->

             <destinationPolicy>
                 <policyMap>
                   <policyEntries>
                     <policyEntry topic=">" producerFlowControl="false"
    memoryLimit="128mb">
                       <pendingSubscriberPolicy>
                         <vmCursor />
                       </pendingSubscriberPolicy>
                     </policyEntry>
                     <policyEntry queue=">" producerFlowControl="false"
    memoryLimit="128mb">
                       <!-- Use VM cursor for better latency
                            For more information, see:

    http://activemq.apache.org/message-cursors.html

                       <pendingQueuePolicy>
                         <vmQueueCursor />
                       </pendingQueuePolicy>
                       -->
                     </policyEntry>
                   </policyEntries>
                 </policyMap>
             </destinationPolicy>


             <!--
                 The managementContext is used to configure how ActiveMQ
    is exposed in
                 JMX. By default, ActiveMQ uses the MBean server that is
    started by
                 the JVM. For more information, see:

                 http://activemq.apache.org/jmx.html
             -->
             <managementContext>
                 <managementContext createConnector="true" />
             </managementContext>

             <!--
                 Configure message persistence for the broker. The
    default persistence
                 mechanism is the KahaDB store (identified by the kahaDB
    tag).
                 For more information, see:

                 http://activemq.apache.org/persistence.html
             -->
             <persistenceAdapter>
                            <!--
                 <kahaDB directory="${activemq.data}/kahadb"
                         ignoreMissingJournalfiles="true"
                         checkForCorruptJournalFiles="false"
                         checksumJournalFiles="false" />
                            -->
                            <mKahaDB
    directory="${activemq.base}/data/kahadb">
                                    <filteredPersistenceAdapters>
                                            <!-- kahaDB per destinations -->
                                            <filteredKahaDB
    perDestination="true" >
    <persistenceAdapter>
                                                            <kahaDB
    journalMaxFileLength="32mb" />
    </persistenceAdapter>
                                            </filteredKahaDB>
    </filteredPersistenceAdapters>
                            </mKahaDB>
             </persistenceAdapter>


               <systemUsage>
                 <systemUsage>
                     <memoryUsage>
                         <memoryUsage limit="2 gb" />
                     </memoryUsage>
                     <storeUsage>
                         <storeUsage limit="65 gb" />
                     </storeUsage>
                     <tempUsage>
                         <tempUsage limit="25 gb" />
                     </tempUsage>
                 </systemUsage>
             </systemUsage>

             <!--
                 networkConnectors added for clustering
             -->
             <networkConnectors>
                 <networkConnector
                     uri="multicast://default?group=nonblocking"
                     dynamicOnly="true"
                     networkTTL="3"
                     prefetchSize="1000"
                     decreaseNetworkConsumerPriority="true">

                     <excludedDestinations>
                         <!-- If you have the netcwork connector on both the
                              VirtualTopic and the underlying queues
    then your
                              consumers are likely to get duplicates. -->
                         <topic physicalName="VirtualTopic.>"/>
                         <!-- This queue has its own dedicated
    networkConnector -->
                         <queue
    physicalName="Consumer.Sorter.VirtualTopic.nonBlockingMessageFile" />
                         <queue
    physicalName="Consumer.Aggregator.VirtualTopic.BadForwarderBucket" />
                         <queue
    physicalName="Consumer.Aggregator.VirtualTopic.ConfProblemBucket" />
                         <queue
    physicalName="Consumer.Aggregator.VirtualTopic.GoodForwarderBucket" />
                         <queue
    physicalName="Consumer.Aggregator.VirtualTopic.NoProblemBucket" />
                         <queue
    physicalName="Consumer.Aggregator.VirtualTopic.SuspiciousBucket" />
                         <queue
    physicalName="Consumer.Engine.VirtualTopic.HbaseLoader" />
                     </excludedDestinations>
                 </networkConnector>
                 <networkConnector
                     name="NC_Sorter_nonBlockingMessageFile"
                     uri="multicast://default?group=nonblocking"
                     dynamicOnly="true"
                     networkTTL="3"
                     prefetchSize="10"
                     decreaseNetworkConsumerPriority="true">

                     <dynamicallyIncludedDestinations>
                         <queue
    physicalName="Consumer.Sorter.VirtualTopic.nonBlockingMessageFile" />
                     </dynamicallyIncludedDestinations>
                 </networkConnector>
                 <networkConnector
                     name="NC_Aggregator_Buckets"
                     uri="multicast://default?group=nonblocking"
                     dynamicOnly="true"
                     networkTTL="3"
                     prefetchSize="1000"
                     decreaseNetworkConsumerPriority="true">

                     <dynamicallyIncludedDestinations>
                         <queue
    physicalName="Consumer.Aggregator.VirtualTopic.BadForwarderBucket" />
                         <queue
    physicalName="Consumer.Aggregator.VirtualTopic.ConfProblemBucket" />
                         <queue
    physicalName="Consumer.Aggregator.VirtualTopic.GoodForwarderBucket" />
                         <queue
    physicalName="Consumer.Aggregator.VirtualTopic.NoProblemBucket" />
                         <queue
    physicalName="Consumer.Aggregator.VirtualTopic.SuspiciousBucket" />
                     </dynamicallyIncludedDestinations>
                 </networkConnector>
                 <networkConnector
                     name="NC_HbaseLoader"
                     uri="multicast://default?group=nonblocking"
                     dynamicOnly="true"
                     networkTTL="3"
                     prefetchSize="10000"
                     decreaseNetworkConsumerPriority="true">

                     <dynamicallyIncludedDestinations>
                         <queue
    physicalName="Consumer.Engine.VirtualTopic.HbaseLoader" />
                     </dynamicallyIncludedDestinations>
                 </networkConnector>
             </networkConnectors>


             <!--
                 The transport connectors expose ActiveMQ over a given
    protocol to
                 clients and other brokers. For more information, see:

                 http://activemq.apache.org/configuring-transports.html
             -->
             <transportConnectors>
                 <transportConnector
                     name="openwire"
                     uri="tcp://0.0.0.0:61616"
    discoveryUri="multicast://default?group=nonblocking"
                     updateClusterClients="true"
                     rebalanceClusterClients="true"
                     updateClusterClientsOnRemove="true" />
                 <transportConnector
                     name="stomp"
    uri="stomp://0.0.0.0:61613?transport.closeAsync=false"
    discoveryUri="multicast://default?group=nonblocking"
                     updateClusterClients="true"
                     rebalanceClusterClients="true"
                     updateClusterClientsOnRemove="true" />
             </transportConnectors>

             <plugins>
                 <discardingDLQBrokerPlugin dropAll="true"
    dropTemporaryTopics="true" dropTemporaryQueues="true" />
             </plugins>

         </broker>

         <!--
             Enable web consoles, REST and Ajax APIs and demos

             Take a look at ${ACTIVEMQ_HOME}/conf/jetty.xml for more details
         -->
         <import resource="jetty.xml" />

    </beans>


Any insight or suggestions as to what might be going on, or how to 
resolve it, would be greatly appreciated.

~/Matt

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