ActiveMQ Artemis: Muticast address deliver messages inconsistently

Question

We're using Artemis 2.19.0 and distributing XML messages.Recently, we found that some messages might get lost when sending them to a multicast address which has two durable multicast queues bound, those two durable queues both have a XPATH filter.

E.g.:

Create a multicast address named: IN.ADDRESS.FOO
Create two durable queues under it named: IN.QUEUE1.FOO and IN.QUEUE2.FOO
Set a XPATH filter to both queues: XPATH '/XML/DATA/Direction[text()="Right"]'
Send 1000 same matched XML messages to address IN.ADDRESS.FOO (With JMS and message size 20kb)

Somehow, IN.QUEUE1.FOO or IN.QUEUE2.FOO or both won't receive 1000 messages eventually.

We've tired to remove filter from one of them, then everything works fine, both queue will receive 1000 messages.

We're using a HA architecture with one alive node and one back-up node.
We've checked the DLQ and nothing there, so it should not be a max-redelivery-attempt exceeded.
This behaviour happens even with no consumer connect.

So, my question is:

Could it be the reason that a XML filter it may be significantly slower than normal filters that cause some messages get lost?
If not what could be the reason?

If you have anything unclear, please ask. Thanks

UPDATE 1:

Versions:

Java: 1.8
Spring-Integration: 5.5.11
Spring-jms: 5.3.19
Artemis: 2.19.0

XML File:

ConnectionFactory Bean:

@Bean(name = "ConnectionFactory")
    public SingleConnectionFactory ibConnectionFactory(
            @Value("${artemis.broker-url}") String brokerUrl,
            @Value("${artemis.user}") String username,
            @Value("${artemis.password}") String password) throws JMSException {

        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
        factory.setBrokerURL(brokerUrl);
        factory.setUser(username);
        factory.setPassword(password);
        return new SingleConnectionFactory(factory);
    }

Sender program:

try(ActiveMQConnectionFactory fac = new ActiveMQConnectionFactory(brokerUrl, user, password)) {
            Connection conn = fac.createConnection();
            Session session = conn.createSession();
            MessageProducer producer = session.createProducer(new ActiveMQQueue("AQ.QUEUE.FOO"));
            Message msg = session.createTextMessage("AK");

            int count = 0;
            while (count < 1000) {
                System.out.println(count);
                producer.send(msg);
                count ++;
            }
        }

XPATH Filter:

XPATH '/Root/Data/PrimaryKey/Key/DetailedIdentity/ATCode[text()="AK"]'

Message Sample:

AK

Either remove the task-executor attribute on publish-subscribe-channel or remove one of the queue's filter fix the problem.

UPDATE 2:

A minimal example with 10 concurrent tasks sending 1000 messages in total, if all the queues under the same address both have the XPATH filter then it would not receive 1000 messages, but with removing one of the filter it works.

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.jms.client.ActiveMQTopic;

import javax.jms.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class TestRunner {

    public static void main(String[] args) throws Exception {
        String brokerUrl = "(tcp://server1:61616,tcp://server2:61616)?ha=true&reconnectAttempts=-1&retryInterval=100&retryIntervalMultiplier=1.5&maxRetryInterval=6000";
        String user = "admin";
        String password = "admin";

        ActiveMQConnectionFactory fac = new ActiveMQConnectionFactory(brokerUrl, user, password);
        ExecutorService ser = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            ser.submit(() -> {
                Connection conn = null;
                try {
                    conn = fac.createConnection();
                    Session session = conn.createSession();
                    MessageProducer producer = session.createProducer(new ActiveMQTopic("IN.ADDRESS.FOO"));
                    Message msg = session.createTextMessage("AK");
                    int count = 0;
                    while (count < 100) {
                        System.out.println(count);
                        msg.setStringProperty("MessageId", String.valueOf(count));
                        producer.send(msg);
                        count++;
                    }
                    session.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            });
        }
    }
}

Justin Bertram · Accepted Answer

Thanks for the test-case. I was able to use it to reproduce the error you were seeing. I opened ARTEMIS-4687 and sent a PR to fix the issue. This will be fixed in 2.33.0 release which is due in the next few weeks.

ActiveMQ Artemis: Muticast address deliver messages inconsistently

Answers (1)

Related Questions