ActiveMQ Artemis: messages not forwarded to another node with a consumer connected to queue

Question

Environmane: Ubuntu 20.04 LTS
ActiveMQ Artemis version: 2.24.0
JDK version: OpenJDK 11

I have a cluster composed of two nodes. The two broker.xml are as below

node1 (broker_1067):





   

      broker_1067

      true
      
      

      
         tcp://192.168.10.67:61616
      

      5000
      90
      true
      120000
      60000
      HALT
      1008000    
      -1

      
         
         tcp://192.168.10.67:61616?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;amqpMinLargeMessageSize=102400;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpDuplicateDetection=true;supportAdvisory=false;suppressInternalManagementObjects=false

         
         tcp://192.168.10.67:5672?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=AMQP;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpMinLargeMessageSize=102400;amqpDuplicateDetection=true
      

      alt_cluster
      alt_cluster

      
         
            231.7.7.6
            9876
            5000
            cluster-connect
         
      

      
         
            231.7.7.6
            9876
            10000
         
      

      
          
            test.TestQueue
            cluster-connect
            ON_DEMAND
            0
            
         
      

      
         
            
            
            
            
            
            
            
            
            
            
            
         
         
            
            
            
            
            
            
            
            
            
            
            
         
         
            
            
            
            
            
            
            
            
            
            
            
         
      

      
         
         
            DLQ
            ExpiryQueue
            0
            
            -1
            10
            PAGE
            true
            true
         
         
         
            DLQ
            ExpiryQueue
            0
            
            
            -1
            
            -1           
            10M            
            1000            
            1M
            10
            PAGE
            false
            false
            false
            false
            0
         
         
            0
            -1       
            -1
            10M
            1000
            1M
            10
            PAGE
            true
            false
            false
            false
            0

node2 (broker_1167):





   

      broker_1167

      true

      

      
         
         tcp://192.168.11.67:61616
      

      5000
      90
      true
      120000
      60000
      HALT
      600000
      -1

      
         tcp://192.168.11.67:61616?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;amqpMinLargeMessageSize=102400;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpDuplicateDetection=true;supportAdvisory=false;suppressInternalManagementObjects=false

         
         tcp://192.168.11.67:5672?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=AMQP;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpMinLargeMessageSize=102400;amqpDuplicateDetection=true
      

      alt_cluster
      alt_cluster

      
         
            231.7.7.6
            9876
            5000
            cluster-connect
         
      

      
         
            231.7.7.6
            9876
            10000
         
      

      
         
            test.TestQueue
            cluster-connect
            ON_DEMAND
            0
            
         
      


      
         
            
            
            
            
            
            
            
            
            
            
            
         

         
            
            
            
            
            
            
            
            
            
            
            
         


         
            
            
            
            
            
            
            
            
            
            
            
         
      

      
         
         
            DLQ
            ExpiryQueue
            0
            
            -1
            10
            PAGE
            true
            true
         
         
         
            DLQ
            ExpiryQueue
            0
            -1
            -1
            10M

            1000
            1M
            10
            PAGE
            false
            false
            false
            false
            0
         

         
            0
            -1

            -1
            10M
            1000
            1M
            10
            PAGE
            true
            false
            false
            false
            0

From each node's broker diagram I can see two brokers connecting.

Now I have a consumer subscribing test.TestQueue on broker_1067, and have another producer producing messages to test.TestQueue on broker_1167.

However, my consumer can not get anything. From the web console of broker_1067, the attributes "Message count" and "Message added" of queue test.TestQueue are all zero.

If I produce messages to broker_1067 then my consumer connecting to broker_1067 can get messages. Same situation to broker_1167.

Here's my consumer code and producer code as below.

Consumer:

public static final String amqUrl = "amqp://192.168.10.67:5672";
public static final String queueName = "test.TestQueue";
public static final String amqUser   = "FET";
public static final String amqPasswd = "1QAZ2WSX";
public static final String clientId  = "Tester-1";
public static final boolean useTransactionMode = false;
public static final int WAIT_SECONDS = 1200; // seconds
private ConnectionFactory connectionFactory;
private Connection amqConnection;
private Session amqSession;
private Topic subscribedTopic, producedTopic;
private MessageConsumer messageSubscriber;
private MessageProducer messageProducer;
private Queue queue;

System.out.printf("Connect to %s
 ", amqUrl);
connectionFactory = new JmsConnectionFactory(amqUser, amqPasswd, amqUrl);
amqConnection     =  connectionFactory.createConnection();
amqConnection.setClientID(clientId);
amqConnection.start();

amqSession = amqConnection.createSession(useTransactionMode, Session.AUTO_ACKNOWLEDGE);
System.out.printf("Subscribe QUEUE: [%s]
", queueName);
queue      = amqSession.createQueue(queueName);
messageSubscriber = amqSession.createConsumer(queue);

System.out.printf("Wait message coming in from [%s] for %d seconds
", queueName, WAIT_SECONDS);
Message queueMessage = messageSubscriber.receive(WAIT_SECONDS*1000); // Wait some time
System.out.println("Timeout!");
if (queueMessage == null) {
    System.out.println("No message comes in!");
} else {
    System.out.printf(String.format("topic: [%s] Some message comes in....", queueName));
    TextMessage textMessage = (TextMessage) queueMessage;
    System.out.printf("Message content is: %s
", textMessage.getText());
}
messageSubscriber.close();
amqSession.close();
amqConnection.stop();
amqConnection.close();

Producer code:

public static final String amqUrl = "amqp://192.168.10.67:5672";
public static final String queueName = "test.TestQueue";
public static final String amqUser   = "FET";
public static final String amqPasswd = "1QAZ2WSX";
public static final String clientId  = "Tester-2";
public static final boolean useTransactionMode = false;
public static final int WAIT_SECONDS = 1200; // seconds
private ConnectionFactory connectionFactory;
private Connection amqConnection;
private Session amqSession;
private Topic subscribedTopic, producedTopic;
private MessageConsumer messageSubscriber;
private MessageProducer messageProducer;
private Queue queue;

System.out.printf("Connect to %s
 ", amqUrl);
connectionFactory = new JmsConnectionFactory(amqUser, amqPasswd, amqUrl);
amqConnection     =  connectionFactory.createConnection();
amqConnection.setClientID(clientId);
amqConnection.start();

amqSession = amqConnection.createSession(useTransactionMode, Session.AUTO_ACKNOWLEDGE);
System.out.println("Be a producer for putting message to queue:" + queueName);
queue      = amqSession.createQueue(queueName);
messageProducer = amqSession.createProducer(queue);

TextMessage textMessage = amqSession.createTextMessage();
String text = "AAAAAAAAAAAAAAAAAAAAAA";
textMessage.setText(text);
System.out.println("Before sending text message :["+ text+"]");
messageProducer.send(textMessage);
System.out.println("After sending text message");

messageProducer.close();
amqSession.close();
amqConnection.stop();
amqConnection.close();

Consumer runs logs:

Connect to amqp://192.168.10.67:5672
Subscribe QUEUE: [test.TestQueue]
Wait message coming in from [test.TestQueue] for 1200 seconds

Producer runs log:

Connect to amqp://192.168.11.67:5672
Be a producer for putting message to queue:test.TestQueue
Before sending text message :[AAAAAAAAAAAAAAAAAAAAAA]
After sending text message

When two MQ runs, their logs whow as below.

broler_1067 runs first:

2022-09-16 22:35:24,773 INFO  [org.apache.activemq.artemis] AMQ241001: Embedded web server started at http://0.0.0.0:8161
2022-09-16 22:35:24,775 INFO  [org.apache.activemq.artemis] AMQ241002: Artemis Jolokia REST API available at http://0.0.0.0:8161/console/jolokia
2022-09-16 22:35:24,776 INFO  [org.apache.activemq.artemis] AMQ241004: Artemis Console available at http://0.0.0.0:8161/console
2022-09-16 22:35:24,909 INFO  [io.hawt.web.auth.LoginServlet] Hawtio login is using 1800 sec. HttpSession timeout
2022-09-16 22:35:27,835 INFO  [io.hawt.web.auth.keycloak.KeycloakServlet] Keycloak integration is disabled
2022-09-16 22:35:37,613 WARN  [org.apache.activemq.artemis.core.server] AMQ224091: Bridge ClusterConnectionBridge@27f20c06 [name=$.artemis.internal.sf.my-cluster.feecdc06-35bd-11ed-a7c2-00505694586c, queue=QueueImpl[name=$.artemis.internal.sf.my-cluster.feecdc06-35bd-11ed-a7c2-00505694586c, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::name=broker_1067], temp=false]@55882f76 targetConnector=ServerLocatorImpl (identity=(Cluster-connection-bridge::ClusterConnectionBridge@27f20c06 [name=$.artemis.internal.sf.my-cluster.feecdc06-35bd-11ed-a7c2-00505694586c, queue=QueueImpl[name=$.artemis.internal.sf.my-cluster.feecdc06-35bd-11ed-a7c2-00505694586c, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::name=broker_1067], temp=false]@55882f76 targetConnector=ServerLocatorImpl [initialConnectors=[TransportConfiguration(name=artemis, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=192-168-11-67], discoveryGroupConfiguration=null]]::ClusterConnectionImpl@1418550995[nodeUUID=04384127-35bd-11ed-868f-00505694e058, connector=TransportConfiguration(name=artemis, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=192-168-10-67, address=, server=ActiveMQServerImpl::name=broker_1067])) [initialConnectors=[TransportConfiguration(name=artemis, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=192-168-11-67], discoveryGroupConfiguration=null]] is unable to connect to destination. Retrying
2022-09-16 22:35:38,390 INFO  [org.apache.activemq.artemis.core.server] AMQ221027: Bridge ClusterConnectionBridge@27f20c06 [name=$.artemis.internal.sf.my-cluster.feecdc06-35bd-11ed-a7c2-00505694586c, queue=QueueImpl[name=$.artemis.internal.sf.my-cluster.feecdc06-35bd-11ed-a7c2-00505694586c, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::name=broker_1067], temp=false]@55882f76 targetConnector=ServerLocatorImpl (identity=(Cluster-connection-bridge::ClusterConnectionBridge@27f20c06 [name=$.artemis.internal.sf.my-cluster.feecdc06-35bd-11ed-a7c2-00505694586c, queue=QueueImpl[name=$.artemis.internal.sf.my-cluster.feecdc06-35bd-11ed-a7c2-00505694586c, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::name=broker_1067], temp=false]@55882f76 targetConnector=ServerLocatorImpl [initialConnectors=[TransportConfiguration(name=artemis, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=192-168-11-67], discoveryGroupConfiguration=null]]::ClusterConnectionImpl@1418550995[nodeUUID=04384127-35bd-11ed-868f-00505694e058, connector=TransportConfiguration(name=artemis, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=192-168-10-67, address=, server=ActiveMQServerImpl::name=broker_1067])) [initialConnectors=[TransportConfiguration(name=artemis, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=192-168-11-67], discoveryGroupConfiguration=null]] is connected

broker_1167 (after broker_1167 runs successfully, broker_1167 runs):

2022-09-16 22:35:39,459 INFO  [io.hawt.web.servlets.JolokiaConfiguredAgentServlet] Jolokia overridden property: [key=policyLocation, value=file:/opt/250_broker_1167/etc/jolokia-access.xml]
2022-09-16 22:35:39,538 INFO  [org.apache.activemq.artemis] AMQ241001: Embedded web server started at http://0.0.0.0:8161
2022-09-16 22:35:39,540 INFO  [org.apache.activemq.artemis] AMQ241002: Artemis Jolokia REST API available at http://0.0.0.0:8161/console/jolokia
2022-09-16 22:35:39,540 INFO  [org.apache.activemq.artemis] AMQ241004: Artemis Console available at http://0.0.0.0:8161/console
2022-09-16 22:35:39,613 INFO  [org.apache.activemq.artemis.core.server] AMQ221027: Bridge ClusterConnectionBridge@5b2f8d02 [name=$.artemis.internal.sf.my-cluster.04384127-35bd-11ed-868f-00505694e058, queue=QueueImpl[name=$.artemis.internal.sf.my-cluster.04384127-35bd-11ed-868f-00505694e058, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::name=broker_1167], temp=false]@3218f4c5 targetConnector=ServerLocatorImpl (identity=(Cluster-connection-bridge::ClusterConnectionBridge@5b2f8d02 [name=$.artemis.internal.sf.my-cluster.04384127-35bd-11ed-868f-00505694e058, queue=QueueImpl[name=$.artemis.internal.sf.my-cluster.04384127-35bd-11ed-868f-00505694e058, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::name=broker_1167], temp=false]@3218f4c5 targetConnector=ServerLocatorImpl [initialConnectors=[TransportConfiguration(name=artemis, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=192-168-10-67], discoveryGroupConfiguration=null]]::ClusterConnectionImpl@1954355506[nodeUUID=feecdc06-35bd-11ed-a7c2-00505694586c, connector=TransportConfiguration(name=artemis, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=192-168-11-67, address=, server=ActiveMQServerImpl::name=broker_1167])) [initialConnectors=[TransportConfiguration(name=artemis, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=192-168-10-67], discoveryGroupConfiguration=null]] is connected
2022-09-16 22:35:43,153 INFO  [io.hawt.web.auth.LoginServlet] Hawtio login is using 1800 sec. HttpSession timeout
2022-09-16 22:35:45,823 INFO  [io.hawt.web.auth.keycloak.KeycloakServlet] Keycloak integration is disabled

What configuration did I miss?

Domenico Francesco Bruscino · Accepted Answer

This problem is caused by max-hops equals 0. The default value for this parameter is 1, which means messages are only load balanced to other Apache ActiveMQ Artemis serves which are directly connected to this server, see the documentation for further details.

ActiveMQ Artemis: messages not forwarded to another node with a consumer connected to queue

Answers (1)

Related Questions