Chapter 29. Clusters Overview

Server discovery is a mechanism by which servers can propagate their connection details to:

This information, or cluster topology, is sent around normal JBoss EAP messaging connections to clients and to other servers over cluster connections. However, you need a way to establish the initial first connection. This can be done using dynamic discovery techniques like UDP and JGroups, or by providing a static list of initial connectors.

<subsystem xmlns="urn:jboss:domain:messaging-activemq:1.0">
  <server name="default">
    ...
    <broadcast-group name="my-broadcast-group" connectors="http-connector" socket-binding="messaging-group"/>
    ...
  </server>
</subsystem>
...
<socket-binding-group name="standard-sockets" default-interface="public" port-offset="${jboss.socket.binding.port-offset:0}">
  ...
  <socket-binding name="messaging-group" interface="private" port="5432" multicast-address="231.7.7.7" multicast-port="9876"/>
  ...
</socket-binding-group>

<subsystem xmlns="urn:jboss:domain:messaging-activemq:1.0">
  <server name="default">
    ...
    <broadcast-group name="my-broadcast-group" connectors="http-connector" jgroups-channel="activemq-cluster"/>
    ...
  </server>
</subsystem>

/subsystem=messaging-activemq/server=default/broadcast-group=my-broadcast-group:add(connectors=[http-connector],jgroups-channel=activemq-cluster)

<subsystem xmlns="urn:jboss:domain:messaging-activemq:1.0">
  <server name="default">
    ...
    <discovery-group name="my-discovery-group" refresh-timeout="10000" socket-binding="messaging-group"/>
    ...
  </server>
</subsystem>
...
<socket-binding-group name="standard-sockets" default-interface="public" port-offset="${jboss.socket.binding.port-offset:0}">
  ...
  <socket-binding name="messaging-group" interface="private" port="5432" multicast-address="231.7.7.7" multicast-port="9876"/>
  ...
</socket-binding-group>

<subsystem xmlns="urn:jboss:domain:messaging-activemq:1.0">
  <server name="default">
    ...
    <discovery-group name="my-discovery-group" refresh-timeout="10000" jgroups-channel="activemq-cluster"/>
    ...
  </server>
</subsystem>

/subsystem=messaging-activemq/server=default/discovery-group=my-discovery-group:add(refresh-timeout=10000,jgroups-channel=activemq-cluster)

<subsystem xmlns="urn:jboss:domain:messaging-activemq:1.0">
  <server name="default">
    ...
    <connection-factory name="RemoteConnectionFactory" entries="java:jboss/exported/jms/RemoteConnectionFactory" connectors="http-connector"/>
    ...
  </server>
</subsystem>

<subsystem xmlns="urn:jboss:domain:messaging-activemq:1.0">
  <server name="default">
    ...
    <pooled-connection-factory name="activemq-ra" transaction="xa" entries="java:/JmsXA java:jboss/DefaultJMSConnectionFactory" connectors="in-vm"/>
    ...
  </server>
</subsystem>

final String groupAddress = "231.7.7.7";
final int groupPort = 9876;

ServerLocator factory =
  ActiveMQClient.createServerLocatorWithHA(new DiscoveryGroupConfiguration(
      groupAddress,
      groupPort,
      new UDPBroadcastGroupConfiguration(groupAddress, groupPort, null, -1)));
ClientSessionFactory factory = locator.createSessionFactory();
ClientSession session1 = factory.createSession();
ClientSession session2 = factory.createSession();

<subsystem xmlns="urn:jboss:domain:messaging-activemq:1.0">
  <server name="default">
    ...
    <connection-factory name="MyConnectionFactory" entries="..." connectors="http-connector http-node1 http-node2"/>
    ...
  </server>
</subsystem>

HashMap<String, Object> map = new HashMap<String, Object>();
map.put("host", "myhost");
map.put("port", "8080");

HashMap<String, Object> map2 = new HashMap<String, Object>();
map2.put("host", "myhost2");
map2.put("port", "8080");

TransportConfiguration server1 = new TransportConfiguration(NettyConnectorFactory.class.getName(), map);
TransportConfiguration server2 = new TransportConfiguration(NettyConnectorFactory.class.getName(), map2);

ServerLocator locator = ActiveMQClient.createServerLocatorWithHA(server1, server2);
ClientSessionFactory factory = locator.createSessionFactory();
ClientSession session = factory.createSession();

If a cluster connection is defined between nodes of a cluster, then JBoss EAP messaging will load balance messages arriving at a particular node from a client.

A messaging cluster connection can be configured to load balance messages in a round robin fashion, irrespective of whether there are any matching consumers on other nodes. It can also be configured to distribute to other nodes only if matching consumers exist. See the Message Redistribution section for more information.

Configuring the Cluster Connection

A cluster connection group servers into clusters so that messages can be load balanced between the nodes of the cluster. A cluster connection is defined in the JBoss EAP server configuration using the cluster-connection element.

Below is the default cluster-connection as defined in the *-full and *-full-ha profiles. See Cluster Connection Attributes for the complete list of attributes.

<subsystem xmlns="urn:jboss:domain:messaging-activemq:1.0">
  <server name="default">
    ...
    <cluster-connection name="my-cluster" discovery-group="dg-group1" connector-name="http-connector" address="jms"/>
    ...
  </server>
</subsystem>

In the case shown above the cluster connection will load balance messages sent to addresses that start with "jms". This cluster connection will, in effect, apply to all JMS queues and topics since they map to core queues that start with the substring "jms".

<subsystem xmlns="urn:jboss:domain:messaging-activemq:1.0">
  <server name="default">
    ...
    <cluster-connection name="my-cluster" static-connectors="server0-connector server1-connector" .../>
    ...
  </server>
</subsystem>

Configuring a Cluster Connection for Duplicate Detection

The cluster connection internally uses a core bridge to move messages between nodes of the cluster. To configure a cluster connection for duplicate message detection, set the use-duplicate-detection attribute to true, which is the default value.

/subsystem=messaging-activemq/server=default/cluster-connection=my-cluster:write-attribute(name=use-duplicate-detection,value=true)

Cluster User Credentials

When creating connections between nodes of a cluster to form a cluster connection, JBoss EAP messaging uses a cluster user and password.

<subsystem xmlns="urn:jboss:domain:messaging-activemq:1.0">
  <server name="default">
    ...
    <cluster password="${jboss.messaging.cluster.password:CHANGE ME!!}"/>
    ...
  </server>
</subsystem>

With JBoss EAP messaging client-side load balancing, subsequent sessions created using a single session factory can be connected to different nodes of the cluster. This allows sessions to spread smoothly across the nodes of a cluster and not be clumped on any particular node.

The recommended way to declare a load balancing policy to be used by the client factory is to set the connection-load-balancing-policy-class-name attribute of the <connection-factory> resource. JBoss EAP messaging provides the following out-of-the-box load balancing policies, and you can also implement your own.

You can also implement your own policy by implementing the interface org.apache.activemq.artemis.api.core.client.loadbalance.ConnectionLoadBalancingPolicy

With message redistribution, JBoss EAP messaging can be configured to automatically redistribute messages from queues which have no consumers back to other nodes in the cluster which do have matching consumers. To enable this functionality, cluster connection’s message-load-balancing-type must be set to ON_DEMAND, which is the default value. You can set this using the following management CLI command.

/subsystem=messaging-activemq/server=default/cluster-connection=my-cluster:write-attribute(name=message-load-balancing-type,value=ON_DEMAND)

Message redistribution can be configured to kick in immediately after the last consumer on a queue is closed, or to wait a configurable delay after the last consumer on a queue is closed before redistributing. This is configured using the redistribution-delay attribute.

You use the redistribution-delay attribute to set how many milliseconds to wait after the last consumer is closed on a queue before redistributing messages from that queue to other nodes of the cluster that have matching consumers. A value of -1, which is the default value, means that messages will never be redistributed. A value of 0 means that messages will be immediately redistributed.

The address-setting in the default JBoss EAP configuration sets a redistribution-delay value of 1000, meaning that it will wait 1000 milliseconds before redistributing messages.

<subsystem xmlns="urn:jboss:domain:messaging-activemq:1.0">
  <server name="default">
    ...
    <address-setting name="#" redistribution-delay="1000" message-counter-history-day-limit="10" page-size-bytes="2097152" max-size-bytes="10485760" expiry-address="jms.queue.ExpiryQueue" dead-letter-address="jms.queue.DLQ"/>
    ...
  </server>
</subsystem>

It often makes sense to introduce a delay before redistributing as it is a common case that a consumer closes but another one quickly is created on the same queue. In this case, you may not want to redistribute immediately since the new consumer will arrive shortly.

Below is an example of an address-setting that sets a redistribution-delay of 0 for any queue or topic that is bound to an address that starts with "jms.". In this case, messages will be redistributed immediately.

<subsystem xmlns="urn:jboss:domain:messaging-activemq:1.0">
  <server name="default">
    ...
    <address-setting name="jms.#" redistribution-delay="0"/>
    ...
  </server>
</subsystem>

This address setting can be added using the following management CLI command.

/subsystem=messaging-activemq/server=default/address-setting=jms.#:add(redistribution-delay=1000)

Clustered grouping follows a different approach relative to normal message grouping. In a cluster, message groups with specific group ids can arrive on any of the nodes. It is important for a node to determine which group ids are bound to which consumer on which node. Each node is responsible for routing message groups correctly to the node which has the consumer processing those group ids, irrespective of where the message groups arrive by default. Once messages with a given group id are sent to a specific consumer connected to the given node in the cluster, then those messages are never sent to another node even if the consumer is disconnected.

This situation is addressed by a grouping handler. Each node has a grouping handler and this grouping handler (along with other handlers) is responsible for routing the message groups to the correct node. There are two types of grouping handlers: LOCAL and REMOTE.

The local handler is responsible for deciding the route that a message group should take. The remote handlers communicate with the local handler and work accordingly. Each cluster should choose a specific node to have a local grouping handler and all the other nodes should have remote handlers.

The node that initially receives a message group takes the routing decision based on regular cluster routing conditions (round-robin queue availability). The node proposes this decision to the respective grouping handler which then routes the messages to the proposed queue if it accepts the proposal.

If the grouping handler rejects the proposal, it proposes some other route and the routing takes place accordingly. The other nodes follow suite and forward the message groups to the chosen queue. After a message arrives on a queue, it is pinned to a customer on that queue.

You can configure grouping handlers using the management CLI. The following command adds a LOCAL grouping handler with the address news.europe.#.

/subsystem=messaging-activemq/server=default/grouping-handler=my-group-handler:add(grouping-handler-address="news.europe.#",type=LOCAL)

This will require a server reload.

reload

The below table lists the configurable attributes for a grouping-handler.