activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Dejan Bosanac <de...@nighttale.net>
Subject Re: NetworkConnector and VirtualTopic delivery issues
Date Tue, 18 Dec 2012 11:32:47 GMT
Hi Matt,

I don't see any reason why virtual topics would not work when sending
messages using standard producer and work when sent though web
console. It's a complex environment you have, so I'd suggest you
trying to narrow it down if possible.

Try seeing what's happening to the destination in some of those cases,
by turning debug logging and watching JMX parameters on the
destinations and subscriptions.

Also, if possible try to isolate the broker.

I hope this helps a bit.


Regards
--
Dejan Bosanac
----------------------
Red Hat, Inc.
FuseSource is now part of Red Hat
dbosanac@redhat.com
Twitter: @dejanb
Blog: http://sensatic.net
ActiveMQ in Action: http://www.manning.com/snyder/


On Sat, Dec 15, 2012 at 1:32 AM, Matt Henkel <guildencrantz@gmail.com> wrote:
> 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
View raw message