Wildfly 10.1.0.Final activemq.artemis load balancing (rebalancing)

[dkjeldgaard]

How can I configure activemq artemis to resolve the following problem.

I have a clustered Wildfly instance with 4 servers. Each server has a pool of MDBeans listening to a queue defined in the domain.xml messaging-activemq subsystem. Our application sends a large number of messages to the queue which then distributes them evenly across the 4 servers. The problem I'm having is that if one of the servers finishes consuming all of it's messages before another server, I'd like activemq.artemis the starving consumers on the server that finished to be able to pull some messages from the servers that are still processing. Likewise, if we start up another server we'd like activemq-artemis to start redistributing some messages to the newly available server. The nature of our application is such that different messages can take drastically different amounts of time to process, so some server queues can get backed up while others starve. We need a way to recognize starving consumers and redistribute messages to them. In the following configuration we are using the PASNettingProcessorQueue...

            <subsystem xmlns="urn:jboss:domain:messaging-activemq:1.0">

                <server name="default">

                    <security enabled="false"/>

                    <cluster password="${jboss.messaging.cluster.password:defaultPWD}"/>

                    <security-setting name="#">

                        <role name="guest" send="true" consume="true" create-non-durable-queue="true" delete-non-durable-queue="true"/>

                        <role name="bpc" send="true" consume="true" create-non-durable-queue="true" delete-non-durable-queue="true"/>

                    </security-setting>

                    <address-setting name="#" dead-letter-address="jms.queue.DLQ" expiry-address="jms.queue.ExpiryQueue" redelivery-delay="5000" max-delivery-attempts="5"/>

                    <address-setting name="jms.queue.BPCLoggingQueue" dead-letter-address="jms.queue.DLQ" expiry-address="jms.queue.ExpiryQueue" max-delivery-attempts="1"/>

                    <address-setting name="jms.queue.PASNettingProcessorQueue" max-delivery-attempts="1"/>

                    <http-connector name="http-connector" socket-binding="http" endpoint="http-acceptor"/>

                    <http-connector name="http-connector-throughput" socket-binding="http" endpoint="http-acceptor-throughput">

                        <param name="batch-delay" value="50"/>

                    </http-connector>

                    <in-vm-connector name="in-vm" server-id="0"/>

                    <http-acceptor name="http-acceptor" http-listener="default"/>

                    <http-acceptor name="http-acceptor-throughput" http-listener="default">

                        <param name="batch-delay" value="50"/>

                        <param name="direct-deliver" value="false"/>

                    </http-acceptor>

                    <in-vm-acceptor name="in-vm" server-id="0"/>

                    <broadcast-group name="bg-group1" jgroups-channel="activemq-cluster" connectors="http-connector"/>

                    <discovery-group name="dg-group1" jgroups-channel="activemq-cluster"/>

                    <cluster-connection name="my-cluster" address="jms" connector-name="http-connector" discovery-group="dg-group1"/>

                    <jms-queue name="ExpiryQueue" entries="java:/jms/queue/ExpiryQueue"/>

                    <jms-queue name="BPCLoggingQueue" entries="java:jboss/exported/jms/queue/BPCLoggingQueue" durable="false"/>

                    <jms-queue name="PASNettingProcessorQueue" entries="java:jboss/exported/jms/queue/PASNettingProcessorQueue" durable="false"/>

                    <connection-factory name="InVmConnectionFactory" entries="java:/ConnectionFactory" connectors="in-vm"/>

                    <connection-factory name="RemoteConnectionFactory" entries="java:jboss/exported/jms/RemoteConnectionFactory" connectors="http-connector" ha="true" block-on-acknowledge="true" reconnect-attempts="-1"/>

                    <pooled-connection-factory name="activemq-ra" entries="java:/JmsXA java:jboss/DefaultJMSConnectionFactory" connectors="in-vm" transaction="xa"/>

                </server>

            </subsystem>

[mnovak]

Artemis is trying to avoid redistribution of messages in cluster. If there is consumer on local queue then it will not redistribute messages from this local queue to another node in cluster.  However you can trick it by stopping delivery of MDB on local queue by CLI command (it's command for standalone profile, in domain it should be called on the host):

/deployment=mdb1.jar/subsystem=ejb3/message-driven-bean=mdb1:stop-delivery()

It will stop your MDB and because other MDBs in cluster are active, Artemis will start to redistribute messages to those nodes. You can enable message delivery to local MDB by command again:

/deployment=mdb1.jar/subsystem=ejb3/message-driven-bean=mdb1:start-delivery()

Also you will need to adjust "redistribution-delay" for your queue in address-setting to lower value, for example 1000ms so redistribution starts quickly:

<address-setting name="..." ... redistribution-delay="1000"/>

Another and probably better  way might be to allow MDB to consume messages from all nodes in cluster. By default MDB has more than 10 sessions where each can consume from different node in cluster. You will need to change your configuration of pooled-connection-factory to use connector "http-connector" instead of "in-vm"

<pooled-connection-factory name="activemq-ra" entries="java:/JmsXA java:jboss/DefaultJMSConnectionFactory" connectors="http-connector" transaction="xa"/>

and also add activation property rebalanceConnections to your MDBs:

@MessageDriven(name = "mdb",
        activationConfig = {
...
                @ActivationConfigProperty(propertyName = "rebalanceConnections", propertyValue = "true"),
...
        })

rebalanceConnections will allow MDBs to rebalance connections every time cluster topology changes. So if any of the nodes is shutdown or started then MDBs will re-balance. Also it will better utilize CPU power of each node. It's still not ideal as messages will not be redistributed and just part of MDB sessions on other nodes will consume messages from queue which has some messages. This could be compensated by increasing number of MDB sessions but I would be careful with this.

I'm not sure if there is better way how to achieve it.

Thanks,

Mirek

[jbertram]

You might also check the "consumerWindowSize" activation configuration property.  The MDBs will buffer messages by default which may starve other consumers.  It sounds like you need to tune this to maximize throughput.

[jbertram]

Did you get this sorted?

[dkjeldgaard]

Thanks for the info. I finally had a chance to try this suggestion and it did did exactly as you said. I changed the connector to the http-connector and implemented the rebalanceConnections property. Now the consumers are randomly connecting to a queue on any node. However this has a negative side effect of some queues having many consumers and some connectors having very few. In some test cases some queues have 9 consumers while others only have 1 consumer. This causes an uneven workflow. Obviously the result is that the queue with 9 consumers finishes 9 times faster than the queue with 1 consumer. And the 9 consumers starve while the 1 consumer continues.

Do you have any other suggestions to distribute this work evenly between consumers and assure that none "starve" until all messages are processed on all nodes?

- Thanks.

[mnovak]

This should not happen as connections should be connected in round robin fashion (not random) to all nodes in cluster and if cluster topology changes then connections should get rebalanced. If I have some time, I'll try myself. However I can't promise it. Quite busy right now :-)

Could you check in CLI number of consumers (MDB sessions) on all nodes and check IP addresses from where they're coming and then restart this node. If there is 1 consumer (MDB session) on one node then there should be more than 10 on another (4 nodes x 10 MDB sessions = 40 consumers).

[dkjeldgaard]

Thanks for your help Miroslav.

I'd like to follow up on this issue. Even if we can get it to work as you state where MDB's are connected to queue's on nodes throughout the cluster evenly, we still have a starvation problem when one of the queue's runs out of messages before others do. In that scenario the MDB's that happened to be connected to that queue would starve even though the other queues may have many messages left. It is hard for me to believe that activemq artemis doesn't have the ability to keep all consumers in the entire cluster busy until all messages are consumed. We are upgrading from JBoss 5.1 with JBoss messaging, and it did this redistribution by default. I just think we are missing something basic here that would make the queues across the cluster to truely communicate with each other and truly be a clustered queue.

Our goal is to keep all the MDB's busy until all the messages are processed in the entire cluster. Hopefully I'm missing some basic concept here. Any help would be greatly appreciated.

[mnovak]

Artemis and JBoss Messaging are very different in this matter. All nodes in JBoss Messaging cluster are using one database (one schema) to store messages. This has advantage that moving messages between nodes in cluster means just simple SQL "update" command. Messages are not physically moved over network from one node to another. It's cheap operation.

In opposite each node in Artemis cluster is using its own journal on local file system to store messages. There is no shared storage with all messages. Thus moving messaging between nodes in cluster is expensive as they must be physically transferred over network to another node. Rather than to move messages across cluster, it's more effective to deliver them to local consumer on queue than to add additional overhead to the whole cluster.

[mnovak]

I tried to start 4 servers (EAP 7.1.0 Alpha/Artemis 1.5.3 [1]) with attached standalone-full-ha.xml and the same "rebalancing" MDB deployed to each server. I did a few restarts of all nodes in cluster and can see that each queue has roughly the same number of consumers(MDB sessions) on queue InQueue: 12,15,17,13

Could you compare it with your configuration, please?

Thanks,

Mirek

[1] Red Hat JBoss Enterprise Application Platform Download | Red Hat Developers

[mnovak]

I was thinking of this a little bit. It seems that the problem is more about that there are slow consumers. Artemis in WF10 is very fast and can handle 100s of clients. I think that you should have WF10/Artemis node (no cluster). All messages will be sent to this node. Then you can start other WF10 servers with your MDBs connected to this single "jms" node. I think this might be better suite your needs and eliminate problem with starving MDBs.

Also as suggested by Justin, set activation config property "consumerWindowSize" to 0 for all MDB to avoid buffering of messages on MDB's buffer. wdyt?

[tstiemerling]

I am basically trying the same thing using WildFly12. I have setup a cluster and enabled the http-connector, but I dont see any messages being handled by the remote server (there are two servers in the cluster). I did not add the "rebalanceConnections" property though (but I would not expect that to have an effect here). Any other suggestions on how to get this working?