Create local queue on a clustered server?

[yongz]

With HornetQ, is it possible to configure a local permanent queue on a clustered server? We do not want all queues to be clustered. Thanks in advance.

[timfox]

In HornetQ there is no such thing as a clustered queue.

A single queue lives on a single node.

[yongz]

Hi Tim,

Thanks for the response. I am sorry if my question confused you.

In JBoss messaging, we can specify a queue to be clustered by setting "Clustered" attribute to true. Otherwise the queue is local, i.e. messages sent to a local queue can only be consumed by a consumer on the same node.

This feature allows clustered and local queues to co-exist on the same server.

While I am looking at "clustered queue" example in HornetQ:

1. The server is configured as clustered in hornetq-configuration.xml
2. Messages to "exampleQueue" can be consumed by both nodes.

My question is, how can I create a local queue if the server is clustered in HornetQ? In above case, how can I create another queue that can only be consumed by a consumer on the same node?

This is very important for people who want to migrate to HornetQ from JBoss messaging.

Thanks a lot.

[timfox]

As I mentioned before all queues in HornetQ are only local, there is no clustered queue.

Clustering is done by configuring cluster connections, this is described in the user manual.

[yongz]

Thanks, I guess I am still in 'JBoss Messaging' world ..... :-)

Back to HornetQ, my question is, is it possible to create a non-cluster(local) connection if the server is clustered?

Basically, do we still have the option for a message to be consumed locally or by a cluster?

[timfox]

It would probably help you a lot to read the chapters on clustering and HA in the user manual.

[yongz]

Yes, I read the user manual, unfortunately it did not help much:

1. If no cluster connection was defined on node A, then as order messages arrive on node A they will all end up in the OrderQueue on node A, so will only get consumed by the order processor client attached to node A, Pa.
2. If we define a cluster connection on node A, then as ordered messages arrive on node A instead of all of them going into the local OrderQueue instance, they are distributed in a round-robin fashion between all the nodes of the cluster.

I have the impression that cluster connection and non cluster connection are mutual exclusive in HornetQ,

I am not sure whether it is still possible to get a workaround, but it will not be as easy as it used to be in JBoss messaging.

[jmesnil]

You need to make sure that the address of the local queue is not matched by the address configured for the cluster connection: http://hornetq.sourceforge.net/docs/hornetq-2.1.1.Final/user-manual/en/html/wildcard-syntax.html

When you configure a cluster-connection, you specify which addresses will be routed to the cluster. If the address of the local queue matches it, it will be routed to other nodes. If it does not match it, the message will be routed to the local queue only.

Doesn't that solve your issue?

[yongz]

Thanks and I just got some time to test this:

• Define address for cluster connection in hornetq-configuration.xml

   <cluster-connections>
      <cluster-connection name="proda1-cluster">
         <address>jms.queue.myClusterQueue</address>      
           <discovery-group-ref discovery-group-name="dg-group1"/>
      </cluster-connection>
   </cluster-connections>

• Define queues in hornetq-jms.xml

   <queue name="myLocalQueue">
      <entry name="/queue/myLocalQueue"/>
   </queue>

   <queue name="myClusterQueue">
      <entry name="/queue/myClusterQueue"/>
   </queue>

However, with above settings, the issue raised in HORNETQ-286 happened again: Temporary Queue not accessible on remote node

https://jira.jboss.org/browse/HORNETQ-286?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel

        at com.mycom.jms.RequestQueueMonitor.onMessage(RequestQueueMonitor.java:211)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
        at java.lang.reflect.Method.invoke(Method.java:597)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeTarget(MethodInvocation.java:122)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:111)
        at org.jboss.ejb3.EJBContainerInvocationWrapper.invokeNext(EJBContainerInvocationWrapper.java:69)
        at org.jboss.ejb3.interceptors.aop.InterceptorSequencer.invoke(InterceptorSequencer.java:73)
        at org.jboss.ejb3.interceptors.aop.InterceptorSequencer.aroundInvoke(InterceptorSequencer.java:59)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
        at java.lang.reflect.Method.invoke(Method.java:597)
        at org.jboss.aop.advice.PerJoinpointAdvice.invoke(PerJoinpointAdvice.java:174)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:102)
        at org.jboss.ejb3.interceptors.aop.InvocationContextInterceptor.fillMethod(InvocationContextInterceptor.java:72)
        at org.jboss.aop.advice.org.jboss.ejb3.interceptors.aop.InvocationContextInterceptor_z_fillMethod_5985294.invoke(InvocationContex
tInterceptor_z_fillMethod_5985294.java)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:102)
        at org.jboss.ejb3.interceptors.aop.InvocationContextInterceptor.setup(InvocationContextInterceptor.java:88)
        at org.jboss.aop.advice.org.jboss.ejb3.interceptors.aop.InvocationContextInterceptor_z_setup_5985294.invoke(InvocationContextInte
rceptor_z_setup_5985294.java)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:102)
        at org.jboss.ejb3.connectionmanager.CachedConnectionInterceptor.invoke(CachedConnectionInterceptor.java:62)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:102)
        at org.jboss.ejb3.entity.TransactionScopedEntityManagerInterceptor.invoke(TransactionScopedEntityManagerInterceptor.java:56)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:102)
        at org.jboss.ejb3.AllowedOperationsInterceptor.invoke(AllowedOperationsInterceptor.java:47)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:102)
        at org.jboss.ejb3.tx.StatelessBMTInterceptor.handleInvocation(StatelessBMTInterceptor.java:106)
        at org.jboss.ejb3.tx.BMTInterceptor.invoke(BMTInterceptor.java:55)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:102)
        at org.jboss.ejb3.stateless.StatelessInstanceInterceptor.invoke(StatelessInstanceInterceptor.java:68)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:102)
        at org.jboss.ejb3.tx.NullInterceptor.invoke(NullInterceptor.java:42)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:102)
        at org.jboss.ejb3.tx.NullInterceptor.invoke(NullInterceptor.java:42)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:102)
        at org.jboss.ejb3.security.Ejb3AuthenticationInterceptorv2.invoke(Ejb3AuthenticationInterceptorv2.java:80)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:102)
        at org.jboss.ejb3.BlockContainerShutdownInterceptor.invoke(BlockContainerShutdownInterceptor.java:67)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:102)
        at org.jboss.aspects.currentinvocation.CurrentInvocationInterceptor.invoke(CurrentInvocationInterceptor.java:67)
        at org.jboss.aop.joinpoint.MethodInvocation.invokeNext(MethodInvocation.java:102)
        at org.jboss.ejb3.mdb.MessagingContainer.localInvoke(MessagingContainer.java:282)
        at org.jboss.ejb3.mdb.inflow.MessageInflowLocalProxy.delivery(MessageInflowLocalProxy.java:270)
        at org.jboss.ejb3.mdb.inflow.MessageInflowLocalProxy.invoke(MessageInflowLocalProxy.java:140)
        at $Proxy339.onMessage(Unknown Source)
        at org.hornetq.ra.inflow.HornetQMessageHandler.onMessage(HornetQMessageHandler.java:256)
        at org.hornetq.core.client.impl.ClientConsumerImpl.callOnMessage(ClientConsumerImpl.java:822)
        at org.hornetq.core.client.impl.ClientConsumerImpl.access$100(ClientConsumerImpl.java:46)
        at org.hornetq.core.client.impl.ClientConsumerImpl$Runner.run(ClientConsumerImpl.java:940)
        at org.hornetq.utils.OrderedExecutorFactory$OrderedExecutor$1.run(OrderedExecutorFactory.java:100)
        at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
        at java.lang.Thread.run(Thread.java:619)
Caused by: javax.jms.InvalidDestinationException: Destination f6ec65c5-1255-41f8-b010-cd7843a0267d does not exist
        at org.hornetq.jms.client.HornetQSession.createProducer(HornetQSession.java:329)
        ... 54 more

It appears that the fixes to HORNETQ-286 only works when address is defined as "jms" which in effect apply to all JMS queues, and breaks when the address for the cluster connection is configured differently as you described above.

[timfox]

This is because (HornetQDestination.java):

private static final long serialVersionUID = 5027962425462382883L;
  
   public static final String JMS_QUEUE_ADDRESS_PREFIX = "jms.queue.";
  
   public static final String JMS_TEMP_QUEUE_ADDRESS_PREFIX = "jms.tempqueue.";
  
   public static final String JMS_TOPIC_ADDRESS_PREFIX = "jms.topic.";

   public static final String JMS_TEMP_TOPIC_ADDRESS_PREFIX = "jms.temptopic.";

Temp queue prefix is jms.tempqueue.

If you want to prevent any message routing or redistribution for certain queues, you could make sure each "local" queue has a different name on each node. In this case the system won't attempt to load balance messages for them across the nodes.

If you like you could add a feature request to "disable" load balancing for certain destinations, e.g. a flag on AddressSettings

[yongz]

If you want to prevent any message routing or redistribution for certain queues, you could make sure each "local" queue has a different name on each node. In this case the system won't attempt to load balance messages for them across the nodes.

Yes, this is an option, but it makes it harder for server configuration.

If you like you could add a feature request to "disable" load balancing for certain destinations, e.g. a flag on AddressSettings

Thanks and feature request has been added:

https://jira.jboss.org/browse/HORNETQ-494

[yongz]

This is because (HornetQDestination.java):

private static final long serialVersionUID = 5027962425462382883L;
   
   public static final String JMS_QUEUE_ADDRESS_PREFIX = "jms.queue.";
   
   public static final String JMS_TEMP_QUEUE_ADDRESS_PREFIX = "jms.tempqueue.";

Thanks for the information and I got a good news for you: I tried another workaround by including address "jms.tempqueue" into cluster connection, and it worked well.