From 240198bbbcab97b4d37a5a61bbd86c2f0b154682 Mon Sep 17 00:00:00 2001 From: Gary Tully Date: Thu, 23 Jul 2026 16:47:10 +0100 Subject: [PATCH] ARTEMIS-6164 MQTT retained message support for mirror and federation Centralise MQTT retain handling into MQTTRetainMessagePlugin, a broker message plugin registered by the MQTT protocol manager factory. The plugin fires via afterMessageRoute on every routing path: normal MQTT publish, mirror, federation, and AMQP bridge. MQTTRetainMessageManager is simplified to only the subscriber-side addRetainedMessagesToQueue(). Assisted-by: Claude --- .../mqtt/MQTTProtocolManagerFactory.java | 1 + .../protocol/mqtt/MQTTPublishManager.java | 6 - .../mqtt/MQTTRetainMessageManager.java | 37 - .../mqtt/MQTTRetainMessagePlugin.java | 81 ++ .../artemis/core/protocol/mqtt/MQTTUtil.java | 2 +- .../core/postoffice/impl/PostOfficeImpl.java | 4 + .../integration/mqtt5/MQTT5BridgeTest.java | 252 ++++ .../mqtt5/MQTT5FederationTest.java | 514 +++++++++ .../integration/mqtt5/MQTT5MirrorTest.java | 1012 +++++++++++++++++ 9 files changed, 1865 insertions(+), 44 deletions(-) create mode 100644 artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTRetainMessagePlugin.java create mode 100644 tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5BridgeTest.java create mode 100644 tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5FederationTest.java create mode 100644 tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5MirrorTest.java diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManagerFactory.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManagerFactory.java index 9d65b1de8d2..791d21fc9dc 100644 --- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManagerFactory.java +++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManagerFactory.java @@ -68,6 +68,7 @@ public String getModuleName() { @Override public void loadProtocolServices(ActiveMQServer server, List services) { services.add(new MQTTPeriodicTasks(server, server.getScheduledPool())); + server.registerBrokerPlugin(new MQTTRetainMessagePlugin()); } public class MQTTPeriodicTasks extends ActiveMQScheduledComponent { diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java index 5f8d8453f82..23e4ad3da9a 100644 --- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java +++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java @@ -24,7 +24,6 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBufAllocator; import io.netty.buffer.ByteBufUtil; -import io.netty.buffer.EmptyByteBuf; import io.netty.handler.codec.mqtt.MqttFixedHeader; import io.netty.handler.codec.mqtt.MqttMessageType; import io.netty.handler.codec.mqtt.MqttProperties; @@ -224,11 +223,6 @@ void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception state.getPubRec().add(packetId); } - if (message.fixedHeader().isRetain()) { - ByteBuf payload = message.payload(); - boolean reset = payload instanceof EmptyByteBuf || payload.capacity() == 0; - session.getRetainMessageManager().handleRetainedMessage(serverMessage, topic, reset, tx); - } tx.commit(); } catch (ActiveMQSecurityException e) { tx.rollback(); diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTRetainMessageManager.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTRetainMessageManager.java index 6dc6d9ac290..460bbcf9307 100644 --- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTRetainMessageManager.java +++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTRetainMessageManager.java @@ -17,15 +17,10 @@ package org.apache.activemq.artemis.core.protocol.mqtt; import org.apache.activemq.artemis.api.core.Message; -import org.apache.activemq.artemis.api.core.QueueConfiguration; import org.apache.activemq.artemis.api.core.SimpleString; -import org.apache.activemq.artemis.core.persistence.StorageManager; -import org.apache.activemq.artemis.core.persistence.impl.journal.LargeServerMessageImpl; import org.apache.activemq.artemis.core.server.BindingQueryResult; import org.apache.activemq.artemis.core.server.MessageReference; import org.apache.activemq.artemis.core.server.Queue; -import org.apache.activemq.artemis.core.server.RoutingContext; -import org.apache.activemq.artemis.core.server.impl.RoutingContextImpl; import org.apache.activemq.artemis.core.transaction.Transaction; import org.apache.activemq.artemis.utils.collections.LinkedListIterator; @@ -39,32 +34,6 @@ public MQTTRetainMessageManager(MQTTSession session) { this.session = session; } - /** - * FIXME - * Retained messages should be handled in the core API. There is currently no support for retained messages - * at the time of writing. Instead we handle retained messages here. This method will create a new queue for - * every address that is used to store retained messages. THere should only ever be one message in the retained - * message queue. When a new subscription is created the queue should be browsed and the message copied onto - * the subscription queue for the consumer. When a new retained message is received the message will be sent to - * the retained queue and the previous retain message consumed to remove it from the queue. - */ - void handleRetainedMessage(Message messageParameter, String address, boolean reset, Transaction tx) throws Exception { - String retainAddress = MQTTUtil.getCoreRetainAddressFromMqttTopic(address, session.getWildcardConfiguration()); - - Queue queue = session.getServer().locateQueue(retainAddress); - if (queue == null) { - queue = session.getServer().createQueue(QueueConfiguration.of(retainAddress).setAutoCreated(true)); - } - - queue.deleteAllReferences(); - - if (!reset) { - StorageManager storageManager = session.getServer().getStorageManager(); - Message message = LargeServerMessageImpl.checkLargeMessage(messageParameter, storageManager); - MQTTUtil.sendMessageDirectlyToQueue(storageManager, session.getServer().getPostOffice(), message.copy(storageManager.generateID()), queue, tx); - } - } - void addRetainedMessagesToQueue(Queue queue, String address) throws Exception { // The address filter that matches all retained message queues. String retainAddress = MQTTUtil.getCoreRetainAddressFromMqttTopic(address, session.getWildcardConfiguration()); @@ -96,10 +65,4 @@ void addRetainedMessagesToQueue(Queue queue, String address) throws Exception { } tx.commit(); } - - private void sendToQueue(Message message, Queue queue, Transaction tx) throws Exception { - RoutingContext context = new RoutingContextImpl(tx); - queue.route(message, context); - session.getServer().getPostOffice().processRoute(message, context, false); - } } diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTRetainMessagePlugin.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTRetainMessagePlugin.java new file mode 100644 index 00000000000..606f2f083ee --- /dev/null +++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTRetainMessagePlugin.java @@ -0,0 +1,81 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.core.protocol.mqtt; + +import org.apache.activemq.artemis.api.core.ActiveMQException; +import org.apache.activemq.artemis.api.core.Message; +import org.apache.activemq.artemis.api.core.QueueConfiguration; +import org.apache.activemq.artemis.core.postoffice.RoutingStatus; +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.server.Queue; +import org.apache.activemq.artemis.core.server.RoutingContext; +import org.apache.activemq.artemis.core.server.impl.RoutingContextImpl; +import org.apache.activemq.artemis.core.server.plugin.ActiveMQServerMessagePlugin; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.lang.invoke.MethodHandles; + +public class MQTTRetainMessagePlugin implements ActiveMQServerMessagePlugin { + + private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + private ActiveMQServer server; + + @Override + public void registered(ActiveMQServer server) { + this.server = server; + } + + @Override + public void afterMessageRoute(Message message, RoutingContext context, boolean direct, boolean rejectDuplicates, + RoutingStatus result) throws ActiveMQException { + try { + Boolean isRetain = message.getBooleanProperty(MQTTUtil.MQTT_MESSAGE_RETAIN_KEY); + if (isRetain == null || !isRetain) { + return; + } + + String address = message.getAddress(); + if (address == null) { + return; + } + + String retainQueueName = MQTTUtil.MQTT_RETAIN_ADDRESS_PREFIX + address; + Queue retainQueue = server.createQueue(QueueConfiguration.of(retainQueueName).setAutoCreated(true), true); + retainQueue.deleteAllReferences(); + + if (message.toCore().getBodyBufferSize() > 0) { + Message copy = message.copy(server.getStorageManager().generateID()); + RoutingContext retainContext = new RoutingContextImpl(context.getTransaction()).setMirrorOption(RoutingContext.MirrorOption.disabled); + retainQueue.route(copy, retainContext); + server.getPostOffice().processRoute(copy, retainContext, false); + } + } catch (Exception e) { + logger.warn("Failed to handle MQTT retained message for address {}: {}", message.getAddress(), e.getMessage(), e); + } + } + + @Override + public int hashCode() { + return MQTTRetainMessagePlugin.class.hashCode(); + } + + @Override + public boolean equals(Object obj) { + return obj instanceof MQTTRetainMessagePlugin; + } +} diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTUtil.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTUtil.java index 3316e55f827..4fa997d445d 100644 --- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTUtil.java +++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTUtil.java @@ -592,7 +592,7 @@ public static void sendMessageDirectlyToQueue(StorageManager storageManager, Pos tx = new TransactionImpl(storageManager); tx.setAsync(true); } - RoutingContext context = new RoutingContextImpl(tx); + RoutingContext context = new RoutingContextImpl(tx).setMirrorOption(RoutingContext.MirrorOption.disabled); queue.route(message, context); postOffice.processRoute(message, context, false); if (incomingTx == null) { diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java index 6ee9a03a1a9..c4d1682eacf 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java @@ -1251,6 +1251,10 @@ private RoutingStatus route(final Message message, final RoutingStatus finalStatus; try { if (status == RoutingStatus.NO_BINDINGS) { + if (mirrorControllerSource != null && !context.isMirrorDisabled() && + addressInfo != null && addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST)) { + mirrorControllerSource.sendMessage(context.getTransaction(), message, context); + } finalStatus = maybeSendToDLA(message, context, address, sendToDLA); } else { finalStatus = status; diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5BridgeTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5BridgeTest.java new file mode 100644 index 00000000000..acb89a151de --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5BridgeTest.java @@ -0,0 +1,252 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.tests.integration.mqtt5; + +import java.nio.charset.StandardCharsets; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; +import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBridgeAddressPolicyElement; +import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBridgeBrokerConnectionElement; +import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectConfiguration; +import org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil; +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.settings.impl.AddressSettings; +import org.apache.activemq.artemis.utils.Wait; +import org.eclipse.paho.mqttv5.client.IMqttToken; +import org.eclipse.paho.mqttv5.client.MqttCallback; +import org.eclipse.paho.mqttv5.client.MqttClient; +import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse; +import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence; +import org.eclipse.paho.mqttv5.common.MqttException; +import org.eclipse.paho.mqttv5.common.MqttMessage; +import org.eclipse.paho.mqttv5.common.packet.MqttProperties; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests MQTT retained message behaviour over a unidirectional AMQP bridge. + * Broker 2 bridges FROM broker 1 — when a subscriber on broker 2 creates + * demand, messages (including retained) are pulled from broker 1. + * The {@code MQTTRetainMessagePlugin} fires via {@code afterMessageRoute} + * on the bridge consumer's routing path, so retained messages should be + * populated in the local retain queue on broker 2 automatically. + */ +public class MQTT5BridgeTest extends ActiveMQTestBase { + + private static final int BROKER1_PORT = 1883; + private static final int BROKER2_PORT = 1884; + + @BeforeEach + @Override + public void setUp() throws Exception { + super.setUp(); + } + + private ActiveMQServer createServer(int serverID, int port) throws Exception { + ActiveMQServer server = createServer(true, createDefaultConfig(serverID, true)); + server.getConfiguration().getAcceptorConfigurations().clear(); + server.getConfiguration().addAcceptorConfiguration("server", "tcp://localhost:" + port); + server.getConfiguration().setSecurityEnabled(false); + server.getConfiguration().setMqttSessionScanInterval(200); + + AddressSettings addressSettings = new AddressSettings(); + addressSettings.setAutoCreateQueues(true); + addressSettings.setAutoCreateAddresses(true); + server.getConfiguration().getAddressSettings().put("#", addressSettings); + + return server; + } + + private ActiveMQServer createBridgedServer() throws Exception { + ActiveMQServer server = createServer(1, MQTT5BridgeTest.BROKER2_PORT); + + final AMQPBridgeAddressPolicyElement addressPolicy = new AMQPBridgeAddressPolicyElement(); + addressPolicy.setName("mqtt-bridge-policy"); + addressPolicy.addToIncludes("#"); + + final AMQPBridgeBrokerConnectionElement bridgeElement = new AMQPBridgeBrokerConnectionElement(); + bridgeElement.setName("bridgeFromServer1"); + bridgeElement.addBridgeFromAddressPolicy(addressPolicy); + + final AMQPBrokerConnectConfiguration amqpConnection = + new AMQPBrokerConnectConfiguration("bridgeFromServer1", "tcp://localhost:" + MQTT5BridgeTest.BROKER1_PORT) + .setReconnectAttempts(10) + .setRetryInterval(100); + amqpConnection.addElement(bridgeElement); + + server.getConfiguration().addAMQPConnection(amqpConnection); + + return server; + } + + private MqttClient createPahoClient(String clientId, int port) throws MqttException { + return new MqttClient("tcp://localhost:" + port, clientId, new MemoryPersistence()); + } + + @Test + @Timeout(60) + public void testRetainedMessageOverBridge() throws Exception { + // broker 1 is a plain server (source); broker 2 bridges FROM broker 1 + ActiveMQServer server1 = createServer(0, BROKER1_PORT); + ActiveMQServer server2 = createBridgedServer(); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + final String topic = "test/retain/bridge"; + final String payload = "retained-via-bridge"; + + // subscribe on broker 2 first — this creates demand so the bridge pulls from broker 1 + CountDownLatch latch = new CountDownLatch(1); + AtomicReference received = new AtomicReference<>(); + MqttClient sub = createPahoClient("subscriber", BROKER2_PORT); + sub.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch.countDown(); + } + }); + sub.connect(); + sub.subscribe(topic, 1); + + // publish retained message on broker 1 + MqttClient producer = createPahoClient("producer", BROKER1_PORT); + producer.connect(); + producer.publish(topic, payload.getBytes(StandardCharsets.UTF_8), 1, true); + producer.disconnect(); + producer.close(); + + // subscriber on broker 2 should receive the message via the bridge + assertTrue(latch.await(10, TimeUnit.SECONDS), "Subscriber on broker 2 should receive the bridged message"); + assertEquals(payload, received.get()); + + // verify retain queue on server1 (local MQTT retain handling) + final String retainQueueName = MQTTUtil.getCoreRetainAddressFromMqttTopic(topic, server1.getConfiguration().getWildcardConfiguration()); + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server1.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 5000, 100); + + // retain queue should also exist on server2 — populated by the afterMessageRoute plugin + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server2.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 5000, 100); + + sub.disconnect(); + sub.close(); + + // a new subscriber on broker 2 should get the retained message from the local retain queue + CountDownLatch retainLatch = new CountDownLatch(1); + AtomicReference retainReceived = new AtomicReference<>(); + MqttClient sub2 = createPahoClient("subscriber2", BROKER2_PORT); + sub2.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + retainReceived.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + retainLatch.countDown(); + } + }); + sub2.connect(); + sub2.subscribe(topic, 1); + assertTrue(retainLatch.await(5, TimeUnit.SECONDS), "New subscriber on broker 2 should receive the retained message"); + assertEquals(payload, retainReceived.get()); + + sub2.disconnect(); + sub2.close(); + } + + @Test + @Timeout(60) + public void testRegularMessageOverBridge() throws Exception { + ActiveMQServer server1 = createServer(0, BROKER1_PORT); + ActiveMQServer server2 = createBridgedServer(); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + final String topic = "test/bridge/regular"; + final String payload = "regular-via-bridge"; + + // subscribe on broker 2 to create demand + CountDownLatch latch = new CountDownLatch(1); + AtomicReference received = new AtomicReference<>(); + MqttClient sub = createPahoClient("subscriber", BROKER2_PORT); + sub.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch.countDown(); + } + }); + sub.connect(); + sub.subscribe(topic, 1); + + // publish a regular (non-retained) message on broker 1 + MqttClient producer = createPahoClient("producer", BROKER1_PORT); + producer.connect(); + producer.publish(topic, payload.getBytes(StandardCharsets.UTF_8), 1, false); + producer.disconnect(); + producer.close(); + + // subscriber on broker 2 should receive the message via the bridge + assertTrue(latch.await(10, TimeUnit.SECONDS), "Subscriber on broker 2 should receive the bridged message"); + assertEquals(payload, received.get()); + + sub.disconnect(); + sub.close(); + } + + private interface DefaultMqttCallback extends MqttCallback { + + @Override + default void disconnected(MqttDisconnectResponse disconnectResponse) { + } + + @Override + default void mqttErrorOccurred(MqttException exception) { + } + + @Override + default void messageArrived(String topic, MqttMessage message) { + } + + @Override + default void deliveryComplete(IMqttToken token) { + } + + @Override + default void connectComplete(boolean reconnect, String serverURI) { + } + + @Override + default void authPacketArrived(int reasonCode, MqttProperties properties) { + } + } +} diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5FederationTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5FederationTest.java new file mode 100644 index 00000000000..64ff6a24a77 --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5FederationTest.java @@ -0,0 +1,514 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.tests.integration.mqtt5; + +import java.nio.charset.StandardCharsets; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectConfiguration; +import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPFederatedBrokerConnectionElement; +import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPFederationAddressPolicyElement; +import org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil; +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.settings.impl.AddressSettings; +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; +import org.apache.activemq.artemis.utils.Wait; +import org.eclipse.paho.mqttv5.client.IMqttToken; +import org.eclipse.paho.mqttv5.client.MqttCallback; +import org.eclipse.paho.mqttv5.client.MqttClient; +import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse; +import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence; +import org.eclipse.paho.mqttv5.common.MqttException; +import org.eclipse.paho.mqttv5.common.MqttMessage; +import org.eclipse.paho.mqttv5.common.packet.MqttProperties; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class MQTT5FederationTest extends ActiveMQTestBase { + + private static final int BROKER1_PORT = 1883; + private static final int BROKER2_PORT = 1884; + + @BeforeEach + @Override + public void setUp() throws Exception { + super.setUp(); + } + + private ActiveMQServer createFederatedServer(int serverID, int port, int targetPort, String connectionName) throws Exception { + ActiveMQServer server = createServer(true, createDefaultConfig(serverID, true)); + server.getConfiguration().getAcceptorConfigurations().clear(); + server.getConfiguration().addAcceptorConfiguration("server", "tcp://localhost:" + port); + server.getConfiguration().setSecurityEnabled(false); + server.getConfiguration().setMqttSessionScanInterval(200); + + AddressSettings addressSettings = new AddressSettings(); + addressSettings.setAutoCreateQueues(true); + addressSettings.setAutoCreateAddresses(true); + server.getConfiguration().getAddressSettings().put("#", addressSettings); + + final AMQPFederationAddressPolicyElement localAddressPolicy = new AMQPFederationAddressPolicyElement(); + localAddressPolicy.setName("mqtt-address-policy"); + localAddressPolicy.addToIncludes("#"); + localAddressPolicy.setAutoDelete(false); + localAddressPolicy.setAutoDeleteDelay(-1L); + localAddressPolicy.setAutoDeleteMessageCount(-1L); + + final AMQPFederatedBrokerConnectionElement federationElement = new AMQPFederatedBrokerConnectionElement(); + federationElement.setName(connectionName); + federationElement.addLocalAddressPolicy(localAddressPolicy); + + final AMQPBrokerConnectConfiguration amqpConnection = + new AMQPBrokerConnectConfiguration(connectionName, "tcp://localhost:" + targetPort) + .setReconnectAttempts(10) + .setRetryInterval(100); + amqpConnection.addElement(federationElement); + + server.getConfiguration().addAMQPConnection(amqpConnection); + + return server; + } + + private ActiveMQServer createBindingsConsumerFederatedServer(int serverID, int port, int targetPort, String connectionName) throws Exception { + ActiveMQServer server = createServer(true, createDefaultConfig(serverID, true)); + server.getConfiguration().getAcceptorConfigurations().clear(); + server.getConfiguration().addAcceptorConfiguration("server", "tcp://localhost:" + port); + server.getConfiguration().setSecurityEnabled(false); + server.getConfiguration().setMqttSessionScanInterval(200); + + AddressSettings addressSettings = new AddressSettings(); + addressSettings.setAutoCreateQueues(true); + addressSettings.setAutoCreateAddresses(true); + server.getConfiguration().getAddressSettings().put("#", addressSettings); + + final AMQPFederationAddressPolicyElement localAddressPolicy = new AMQPFederationAddressPolicyElement(); + localAddressPolicy.setName("mqtt-address-policy"); + localAddressPolicy.addToIncludes("#"); + localAddressPolicy.setAutoDelete(false); + localAddressPolicy.setAutoDeleteDelay(-1L); + localAddressPolicy.setAutoDeleteMessageCount(-1L); + localAddressPolicy.addProperty("ignoreAddressBindingFilters", "false"); + + final AMQPFederatedBrokerConnectionElement federationElement = new AMQPFederatedBrokerConnectionElement(); + federationElement.setName(connectionName); + federationElement.addLocalAddressPolicy(localAddressPolicy); + + final AMQPBrokerConnectConfiguration amqpConnection = + new AMQPBrokerConnectConfiguration(connectionName, "tcp://localhost:" + targetPort) + .setReconnectAttempts(10) + .setRetryInterval(100); + amqpConnection.addElement(federationElement); + + server.getConfiguration().addAMQPConnection(amqpConnection); + + return server; + } + + private MqttClient createPahoClient(String clientId, int port) throws MqttException { + return new MqttClient("tcp://localhost:" + port, clientId, new MemoryPersistence()); + } + + @Test + @Timeout(60) + public void testPartitionedSubscriptionsWithFederation() throws Exception { + ActiveMQServer server1 = createFederatedServer(0, BROKER1_PORT, BROKER2_PORT, "federationToServer2"); + ActiveMQServer server2 = createFederatedServer(1, BROKER2_PORT, BROKER1_PORT, "federationToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + final String sportsTopic = "news/sports"; + final String techTopic = "news/tech"; + + // subscribers are partitioned: sub-a on broker 1 for sports, sub-b on broker 2 for tech + CountDownLatch sportsLatch = new CountDownLatch(2); + CopyOnWriteArrayList sportsReceived = new CopyOnWriteArrayList<>(); + MqttClient subA = createPahoClient("sub-a", BROKER1_PORT); + subA.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + sportsReceived.add(new String(m.getPayload(), StandardCharsets.UTF_8)); + sportsLatch.countDown(); + } + }); + subA.connect(); + subA.subscribe(sportsTopic, 1); + + CountDownLatch techLatch = new CountDownLatch(2); + CopyOnWriteArrayList techReceived = new CopyOnWriteArrayList<>(); + MqttClient subB = createPahoClient("sub-b", BROKER2_PORT); + subB.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + techReceived.add(new String(m.getPayload(), StandardCharsets.UTF_8)); + techLatch.countDown(); + } + }); + subB.connect(); + subB.subscribe(techTopic, 1); + + // publish on the broker that has NO local subscriber for that topic + MqttClient pub1 = createPahoClient("pub1", BROKER2_PORT); + pub1.connect(); + pub1.publish(sportsTopic, "goal".getBytes(StandardCharsets.UTF_8), 1, false); + pub1.disconnect(); + pub1.close(); + + MqttClient pub2 = createPahoClient("pub2", BROKER1_PORT); + pub2.connect(); + pub2.publish(techTopic, "update".getBytes(StandardCharsets.UTF_8), 1, false); + pub2.disconnect(); + pub2.close(); + + // also publish locally to verify local delivery still works + MqttClient pub3 = createPahoClient("pub3", BROKER1_PORT); + pub3.connect(); + pub3.publish(sportsTopic, "try".getBytes(StandardCharsets.UTF_8), 1, false); + pub3.disconnect(); + pub3.close(); + + MqttClient pub4 = createPahoClient("pub4", BROKER2_PORT); + pub4.connect(); + pub4.publish(techTopic, "release".getBytes(StandardCharsets.UTF_8), 1, false); + pub4.disconnect(); + pub4.close(); + + assertTrue(sportsLatch.await(10, TimeUnit.SECONDS), "sub-a should receive both sports messages"); + assertTrue(techLatch.await(10, TimeUnit.SECONDS), "sub-b should receive both tech messages"); + + assertEquals(2, sportsReceived.size()); + assertTrue(sportsReceived.contains("goal"), "sub-a should receive 'goal' published on broker 2"); + assertTrue(sportsReceived.contains("try"), "sub-a should receive 'try' published on broker 1"); + + assertEquals(2, techReceived.size()); + assertTrue(techReceived.contains("update"), "sub-b should receive 'update' published on broker 1"); + assertTrue(techReceived.contains("release"), "sub-b should receive 'release' published on broker 2"); + + subA.disconnect(); + subA.close(); + subB.disconnect(); + subB.close(); + } + + @Test + @Timeout(60) + public void testRetainedMessageWithFederation() throws Exception { + ActiveMQServer server1 = createFederatedServer(0, BROKER1_PORT, BROKER2_PORT, "federationToServer2"); + ActiveMQServer server2 = createFederatedServer(1, BROKER2_PORT, BROKER1_PORT, "federationToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + final String topic = "test/retain/federation"; + final String payload = "retained-message-payload"; + + // subscribe on broker 2 first — this creates federation demand + CountDownLatch latch = new CountDownLatch(1); + AtomicReference received = new AtomicReference<>(); + MqttClient sub = createPahoClient("subscriber", BROKER2_PORT); + sub.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch.countDown(); + } + }); + sub.connect(); + sub.subscribe(topic, 1); + + // publish retained message on broker 1 + MqttClient producer = createPahoClient("producer", BROKER1_PORT); + producer.connect(); + producer.publish(topic, payload.getBytes(StandardCharsets.UTF_8), 1, true); + producer.disconnect(); + producer.close(); + + // subscriber on broker 2 should receive the message via federation + assertTrue(latch.await(10, TimeUnit.SECONDS), "Subscriber on broker 2 should receive the federated message"); + assertEquals(payload, received.get()); + + // verify retain queue exists on server1 (local retain handling) + final String retainQueueName = MQTTUtil.getCoreRetainAddressFromMqttTopic(topic, server1.getConfiguration().getWildcardConfiguration()); + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server1.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 5000, 100); + + // retain queue should also exist on server2 — populated by federation consumer + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server2.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 5000, 100); + + // verify no duplicate retain processing: messagesAdded should be exactly 1 + assertEquals(1, server2.locateQueue(retainQueueName).getMessagesAdded(), + "Retain queue on server2 should have exactly 1 message added (no duplicate processing)"); + + sub.disconnect(); + sub.close(); + + // a new subscriber on broker 2 should get the retained message from the local retain queue + CountDownLatch retainLatch = new CountDownLatch(1); + AtomicReference retainReceived = new AtomicReference<>(); + MqttClient sub2 = createPahoClient("subscriber2", BROKER2_PORT); + sub2.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + retainReceived.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + retainLatch.countDown(); + } + }); + sub2.connect(); + sub2.subscribe(topic, 1); + assertTrue(retainLatch.await(5, TimeUnit.SECONDS), "New subscriber on broker 2 should receive the retained message"); + assertEquals(payload, retainReceived.get()); + + sub2.disconnect(); + sub2.close(); + } + + @Test + @Timeout(60) + public void testClearRetainedMessageWithFederation() throws Exception { + ActiveMQServer server1 = createFederatedServer(0, BROKER1_PORT, BROKER2_PORT, "federationToServer2"); + ActiveMQServer server2 = createFederatedServer(1, BROKER2_PORT, BROKER1_PORT, "federationToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + final String topic = "test/retain/federation/clear"; + final String payload = "retained-to-clear"; + final String retainQueueName = MQTTUtil.getCoreRetainAddressFromMqttTopic(topic, server1.getConfiguration().getWildcardConfiguration()); + + // subscribe on broker 2 to create federation demand + CountDownLatch latch = new CountDownLatch(1); + AtomicReference received = new AtomicReference<>(); + MqttClient sub = createPahoClient("subscriber", BROKER2_PORT); + sub.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch.countDown(); + } + }); + sub.connect(); + sub.subscribe(topic, 1); + + // publish a retained message on broker 1 + MqttClient producer = createPahoClient("producer", BROKER1_PORT); + producer.connect(); + producer.publish(topic, payload.getBytes(StandardCharsets.UTF_8), 1, true); + + assertTrue(latch.await(10, TimeUnit.SECONDS), "Subscriber on broker 2 should receive the federated message"); + assertEquals(payload, received.get()); + + // verify retain queue populated on both brokers + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server1.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 5000, 100); + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server2.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 5000, 100); + + // clear the retained message with a zero-length payload (MQTT spec §3.3.1.3) + producer.publish(topic, new byte[0], 1, true); + producer.disconnect(); + producer.close(); + + // retain queue on server1 should be empty + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server1.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 0; + }, 5000, 100); + + // retain queue on server2 should also be empty — cleared via federated zero-length message + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server2.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 0; + }, 10000, 100); + + sub.disconnect(); + sub.close(); + + // a new subscriber on broker 2 should NOT receive a retained message + CountDownLatch noRetainLatch = new CountDownLatch(1); + AtomicReference noRetainReceived = new AtomicReference<>(); + MqttClient sub2 = createPahoClient("subscriber2", BROKER2_PORT); + sub2.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + noRetainReceived.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + noRetainLatch.countDown(); + } + }); + sub2.connect(); + sub2.subscribe(topic, 1); + + // publish a regular message so we know the subscription is active + MqttClient probe = createPahoClient("probe", BROKER1_PORT); + probe.connect(); + probe.publish(topic, "probe".getBytes(StandardCharsets.UTF_8), 1, false); + probe.disconnect(); + probe.close(); + + assertTrue(noRetainLatch.await(10, TimeUnit.SECONDS), "Subscriber should receive the probe message"); + assertEquals("probe", noRetainReceived.get(), "Only the probe message should arrive — no retained message"); + + sub2.disconnect(); + sub2.close(); + } + + @Test + @Timeout(60) + public void testRetainedMessageNotHandledWithFederationBindingsConsumer() throws Exception { + ActiveMQServer server1 = createBindingsConsumerFederatedServer(0, BROKER1_PORT, BROKER2_PORT, "federationToServer2"); + ActiveMQServer server2 = createBindingsConsumerFederatedServer(1, BROKER2_PORT, BROKER1_PORT, "federationToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + final String topic = "test/retain/federation/bindings"; + final String payload = "retained-via-bindings-consumer"; + + // two subscribers on broker 2 — the bindings consumer routes directly to + // bindings via processRoute() which does not fire broker plugins, so the + // MQTTRetainMessagePlugin will not populate a retain queue on server2 + CountDownLatch latch = new CountDownLatch(2); + CopyOnWriteArrayList received = new CopyOnWriteArrayList<>(); + MqttClient subA = createPahoClient("subscriberA", BROKER2_PORT); + subA.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received.add(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch.countDown(); + } + }); + subA.connect(); + subA.subscribe(topic, 1); + + MqttClient subB = createPahoClient("subscriberB", BROKER2_PORT); + subB.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received.add(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch.countDown(); + } + }); + subB.connect(); + subB.subscribe(topic, 1); + + // publish retained on broker 1 + MqttClient producer = createPahoClient("producer", BROKER1_PORT); + producer.connect(); + producer.publish(topic, payload.getBytes(StandardCharsets.UTF_8), 1, true); + producer.disconnect(); + producer.close(); + + assertTrue(latch.await(10, TimeUnit.SECONDS), "Both subscribers on broker 2 should receive the federated message"); + assertEquals(2, received.size()); + + // retain queue on server1 is populated locally by the plugin on the publishing broker + final String retainQueueName = MQTTUtil.getCoreRetainAddressFromMqttTopic(topic, server1.getConfiguration().getWildcardConfiguration()); + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server1.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 5000, 100); + + subA.disconnect(); + subA.close(); + subB.disconnect(); + subB.close(); + + // retain queue on server2 should NOT exist — the bindings consumer uses + // processRoute() which bypasses broker plugins (unlike the conduit consumer + // which uses route() and fires the MQTTRetainMessagePlugin naturally) + assertNull(server2.locateQueue(retainQueueName), + "Retain queue should not be created on server2 via the bindings consumer path"); + + // a new subscriber on broker 2 should NOT receive a retained message + CountDownLatch probeLatch = new CountDownLatch(1); + AtomicReference probeReceived = new AtomicReference<>(); + MqttClient sub2 = createPahoClient("subscriber2", BROKER2_PORT); + sub2.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + probeReceived.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + probeLatch.countDown(); + } + }); + sub2.connect(); + sub2.subscribe(topic, 1); + + // send a non-retained probe to confirm the subscription is active + MqttClient probe = createPahoClient("probe", BROKER1_PORT); + probe.connect(); + probe.publish(topic, "probe".getBytes(StandardCharsets.UTF_8), 1, false); + probe.disconnect(); + probe.close(); + + assertTrue(probeLatch.await(10, TimeUnit.SECONDS), "Subscriber should receive the probe message"); + assertEquals("probe", probeReceived.get(), "Only the probe message should arrive — no retained message"); + + sub2.disconnect(); + sub2.close(); + } + + private interface DefaultMqttCallback extends MqttCallback { + + @Override + default void disconnected(MqttDisconnectResponse disconnectResponse) { + } + + @Override + default void mqttErrorOccurred(MqttException exception) { + } + + @Override + default void messageArrived(String topic, MqttMessage message) { + } + + @Override + default void deliveryComplete(IMqttToken token) { + } + + @Override + default void connectComplete(boolean reconnect, String serverURI) { + } + + @Override + default void authPacketArrived(int reasonCode, MqttProperties properties) { + } + } +} diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5MirrorTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5MirrorTest.java new file mode 100644 index 00000000000..70618fc7ab7 --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5MirrorTest.java @@ -0,0 +1,1012 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.tests.integration.mqtt5; + +import java.nio.charset.StandardCharsets; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.activemq.artemis.api.core.QueueConfiguration; +import org.apache.activemq.artemis.api.core.RoutingType; +import org.apache.activemq.artemis.api.core.SimpleString; +import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectConfiguration; +import org.apache.activemq.artemis.core.message.impl.CoreMessage; +import org.apache.activemq.artemis.core.server.impl.AddressInfo; +import org.apache.activemq.artemis.core.server.impl.RoutingContextImpl; +import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPMirrorBrokerConnectionElement; +import org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil; +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.settings.impl.AddressSettings; +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; +import org.apache.activemq.artemis.utils.Wait; +import org.eclipse.paho.mqttv5.client.IMqttToken; +import org.eclipse.paho.mqttv5.client.MqttCallback; +import org.eclipse.paho.mqttv5.client.MqttClient; +import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse; +import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence; +import org.eclipse.paho.mqttv5.common.MqttException; +import org.eclipse.paho.mqttv5.common.MqttMessage; +import org.eclipse.paho.mqttv5.common.packet.MqttProperties; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class MQTT5MirrorTest extends ActiveMQTestBase { + + private static final int BROKER1_PORT = 1883; + private static final int BROKER2_PORT = 1884; + + @BeforeEach + @Override + public void setUp() throws Exception { + super.setUp(); + } + + private ActiveMQServer createMirroredServer(int serverID, int port, int targetPort, String connectionName) throws Exception { + ActiveMQServer server = createServer(true, createDefaultConfig(serverID, true)); + server.getConfiguration().getAcceptorConfigurations().clear(); + server.getConfiguration().addAcceptorConfiguration("server", "tcp://localhost:" + port); + server.getConfiguration().setSecurityEnabled(false); + server.getConfiguration().setMqttSessionScanInterval(200); + + AddressSettings addressSettings = new AddressSettings(); + addressSettings.setAutoCreateQueues(true); + addressSettings.setAutoCreateAddresses(true); + server.getConfiguration().getAddressSettings().put("#", addressSettings); + + AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration(connectionName, "tcp://localhost:" + targetPort) + .setReconnectAttempts(-1) + .setRetryInterval(100); + amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true)); + server.getConfiguration().addAMQPConnection(amqpConnection); + + return server; + } + + private ActiveMQServer createMessageOnlyMirroredServer(int serverID, int port, int targetPort, String connectionName) throws Exception { + ActiveMQServer server = createServer(true, createDefaultConfig(serverID, true)); + server.getConfiguration().getAcceptorConfigurations().clear(); + server.getConfiguration().addAcceptorConfiguration("server", "tcp://localhost:" + port); + server.getConfiguration().setSecurityEnabled(false); + server.getConfiguration().setMqttSessionScanInterval(200); + + AddressSettings addressSettings = new AddressSettings(); + addressSettings.setAutoCreateQueues(true); + addressSettings.setAutoCreateAddresses(true); + server.getConfiguration().getAddressSettings().put("#", addressSettings); + + AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration(connectionName, "tcp://localhost:" + targetPort) + .setReconnectAttempts(-1) + .setRetryInterval(100); + amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement() + .setDurable(true) + .setQueueCreation(false) + .setQueueRemoval(false) + .setMessageAcknowledgements(false)); + server.getConfiguration().addAMQPConnection(amqpConnection); + + return server; + } + + private MqttClient createPahoClient(String clientId, int port) throws MqttException { + return new MqttClient("tcp://localhost:" + port, clientId, new MemoryPersistence()); + } + + @Test + @Timeout(60) + public void testRetainedMessageMirrored() throws Exception { + ActiveMQServer server1 = createMirroredServer(0, BROKER1_PORT, BROKER2_PORT, "mirrorToServer2"); + ActiveMQServer server2 = createMirroredServer(1, BROKER2_PORT, BROKER1_PORT, "mirrorToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + Wait.assertTrue(() -> server1.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer2") != null); + Wait.assertTrue(() -> server2.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer1") != null); + + final String topic = "test/retain/mirror"; + final String payload = "retained-message-payload"; + + // publish retained message to broker 1 + MqttClient producer = createPahoClient("producer", BROKER1_PORT); + producer.connect(); + producer.publish(topic, payload.getBytes(StandardCharsets.UTF_8), 1, true); + producer.disconnect(); + producer.close(); + + // verify the retain queue exists on server1 + final String retainQueueName = MQTTUtil.getCoreRetainAddressFromMqttTopic(topic, server1.getConfiguration().getWildcardConfiguration()); + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server1.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 5000, 100); + + // subscribe on broker 1 - should get the retained message + CountDownLatch latch1 = new CountDownLatch(1); + AtomicReference received1 = new AtomicReference<>(); + MqttClient sub1 = createPahoClient("subscriber1", BROKER1_PORT); + sub1.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received1.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch1.countDown(); + } + }); + sub1.connect(); + sub1.subscribe(topic, 1); + assertTrue(latch1.await(5, TimeUnit.SECONDS), "Subscriber on broker 1 should receive the retained message"); + assertEquals(payload, received1.get()); + sub1.disconnect(); + sub1.close(); + + // wait for the retained message to appear on broker 2 via mirroring + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server2.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 10000, 100); + + // subscribe on broker 2 - should also get the retained message + CountDownLatch latch2 = new CountDownLatch(1); + AtomicReference received2 = new AtomicReference<>(); + MqttClient sub2 = createPahoClient("subscriber2", BROKER2_PORT); + sub2.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received2.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch2.countDown(); + } + }); + sub2.connect(); + sub2.subscribe(topic, 1); + assertTrue(latch2.await(5, TimeUnit.SECONDS), "Subscriber on broker 2 should receive the mirrored retained message"); + assertEquals(payload, received2.get()); + sub2.disconnect(); + sub2.close(); + + // publish a second retained message - should replace the first on both brokers + final String payload2 = "retained-message-payload-2"; + MqttClient producer2 = createPahoClient("producer2", BROKER1_PORT); + producer2.connect(); + producer2.publish(topic, payload2.getBytes(StandardCharsets.UTF_8), 1, true); + producer2.disconnect(); + producer2.close(); + + // verify server1 retain queue replaced: still exactly 1 message with the new payload + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server1.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1 && queue.getMessagesAdded() == 2; + }, 5000, 100); + + // verify server2 retain queue replaced via mirroring: exactly 1 message + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server2.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 10000, 100); + + // subscribe on broker 1 - should get the second retained message + CountDownLatch latch3 = new CountDownLatch(1); + AtomicReference received3 = new AtomicReference<>(); + MqttClient sub3 = createPahoClient("subscriber3", BROKER1_PORT); + sub3.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received3.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch3.countDown(); + } + }); + sub3.connect(); + sub3.subscribe(topic, 1); + assertTrue(latch3.await(5, TimeUnit.SECONDS), "Subscriber on broker 1 should receive the second retained message"); + assertEquals(payload2, received3.get()); + sub3.disconnect(); + sub3.close(); + + // subscribe on broker 2 - should get the second retained message + CountDownLatch latch4 = new CountDownLatch(1); + AtomicReference received4 = new AtomicReference<>(); + MqttClient sub4 = createPahoClient("subscriber4", BROKER2_PORT); + sub4.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received4.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch4.countDown(); + } + }); + sub4.connect(); + sub4.subscribe(topic, 1); + assertTrue(latch4.await(5, TimeUnit.SECONDS), "Subscriber on broker 2 should receive the second retained message"); + assertEquals(payload2, received4.get()); + sub4.disconnect(); + sub4.close(); + } + + @Test + @Timeout(60) + public void testClearRetainedMessageMirrored() throws Exception { + ActiveMQServer server1 = createMirroredServer(0, BROKER1_PORT, BROKER2_PORT, "mirrorToServer2"); + ActiveMQServer server2 = createMirroredServer(1, BROKER2_PORT, BROKER1_PORT, "mirrorToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + Wait.assertTrue(() -> server1.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer2") != null); + Wait.assertTrue(() -> server2.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer1") != null); + + final String topic = "test/retain/clear"; + final String payload = "retained-to-clear"; + final String retainQueueName = MQTTUtil.getCoreRetainAddressFromMqttTopic(topic, server1.getConfiguration().getWildcardConfiguration()); + + // publish a retained message on broker 1 + MqttClient producer = createPahoClient("producer", BROKER1_PORT); + producer.connect(); + producer.publish(topic, payload.getBytes(StandardCharsets.UTF_8), 1, true); + + // verify retain queue populated on both brokers + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server1.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 5000, 100); + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server2.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 10000, 100); + + // publish a zero-length retained message to clear the retain (MQTT spec §3.3.1.3) + producer.publish(topic, new byte[0], 1, true); + producer.disconnect(); + producer.close(); + + // retain queue on server1 should be empty (cleared locally by MQTTRetainMessageManager) + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server1.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 0; + }, 5000, 100); + + // retain queue on server2 should also be empty (cleared via mirrored zero-length message) + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server2.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 0; + }, 10000, 100); + + // a new subscriber on broker 2 should NOT receive a retained message + CountDownLatch latch = new CountDownLatch(1); + AtomicReference received = new AtomicReference<>(); + MqttClient sub = createPahoClient("subscriber", BROKER2_PORT); + sub.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch.countDown(); + } + }); + sub.connect(); + sub.subscribe(topic, 1); + + // publish a regular (non-retained) message so we know the subscription is active + MqttClient probe = createPahoClient("probe", BROKER2_PORT); + probe.connect(); + probe.publish(topic, "probe".getBytes(StandardCharsets.UTF_8), 1, false); + probe.disconnect(); + probe.close(); + + assertTrue(latch.await(5, TimeUnit.SECONDS), "Subscriber should receive the probe message"); + assertEquals("probe", received.get(), "Only the probe message should arrive — no retained message"); + + sub.disconnect(); + sub.close(); + } + + @Test + @Timeout(60) + public void testRegularMessageMirrored() throws Exception { + ActiveMQServer server1 = createMirroredServer(0, BROKER1_PORT, BROKER2_PORT, "mirrorToServer2"); + ActiveMQServer server2 = createMirroredServer(1, BROKER2_PORT, BROKER1_PORT, "mirrorToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + Wait.assertTrue(() -> server1.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer2") != null); + Wait.assertTrue(() -> server2.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer1") != null); + + final String topic = "test/mirror/regular"; + final String payload = "regular-message-payload"; + + // subscribe on broker 2 first so the queue exists when the mirrored message arrives + CountDownLatch latch = new CountDownLatch(1); + AtomicReference received = new AtomicReference<>(); + MqttClient sub = createPahoClient("subscriber", BROKER2_PORT); + sub.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch.countDown(); + } + }); + sub.connect(); + sub.subscribe(topic, 1); + + // publish a regular (non-retained) message on broker 1 + MqttClient producer = createPahoClient("producer", BROKER1_PORT); + producer.connect(); + producer.publish(topic, payload.getBytes(StandardCharsets.UTF_8), 1, false); + producer.disconnect(); + producer.close(); + + // subscriber on broker 2 should receive the mirrored message + assertTrue(latch.await(5, TimeUnit.SECONDS), "Subscriber on broker 2 should receive the mirrored message"); + assertEquals(payload, received.get()); + sub.disconnect(); + sub.close(); + } + + @Test + @Timeout(60) + public void testRetainedAndRegularInterleavedMirrored() throws Exception { + ActiveMQServer server1 = createMirroredServer(0, BROKER1_PORT, BROKER2_PORT, "mirrorToServer2"); + ActiveMQServer server2 = createMirroredServer(1, BROKER2_PORT, BROKER1_PORT, "mirrorToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + Wait.assertTrue(() -> server1.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer2") != null); + Wait.assertTrue(() -> server2.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer1") != null); + + final String topic = "test/interleave/mirror"; + final String retainQueueName = MQTTUtil.getCoreRetainAddressFromMqttTopic(topic, server1.getConfiguration().getWildcardConfiguration()); + + // subscribe on broker 2 to receive live messages + CountDownLatch latch = new CountDownLatch(4); + CopyOnWriteArrayList receivedMessages = new CopyOnWriteArrayList<>(); + MqttClient sub = createPahoClient("subscriber", BROKER2_PORT); + sub.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + receivedMessages.add(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch.countDown(); + } + }); + sub.connect(); + sub.subscribe(topic, 1); + + // interleave retained and regular publishes on broker 1 + MqttClient producer = createPahoClient("producer", BROKER1_PORT); + producer.connect(); + producer.publish(topic, "retained-1".getBytes(StandardCharsets.UTF_8), 1, true); + producer.publish(topic, "regular-1".getBytes(StandardCharsets.UTF_8), 1, false); + producer.publish(topic, "retained-2".getBytes(StandardCharsets.UTF_8), 1, true); + producer.publish(topic, "regular-2".getBytes(StandardCharsets.UTF_8), 1, false); + producer.disconnect(); + producer.close(); + + // subscriber on broker 2 should receive all 4 messages + assertTrue(latch.await(10, TimeUnit.SECONDS), "Subscriber on broker 2 should receive all 4 mirrored messages"); + assertEquals(4, receivedMessages.size()); + assertTrue(receivedMessages.contains("retained-1"), "Should receive retained-1"); + assertTrue(receivedMessages.contains("regular-1"), "Should receive regular-1"); + assertTrue(receivedMessages.contains("retained-2"), "Should receive retained-2"); + assertTrue(receivedMessages.contains("regular-2"), "Should receive regular-2"); + sub.disconnect(); + sub.close(); + + // verify retain queue on server2 has exactly 1 message (retained-2 replaced retained-1) + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server2.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 10000, 100); + + // a new subscriber on broker 2 should get only the latest retained message + CountDownLatch retainLatch = new CountDownLatch(1); + AtomicReference retainedReceived = new AtomicReference<>(); + MqttClient sub2 = createPahoClient("subscriber2", BROKER2_PORT); + sub2.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + retainedReceived.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + retainLatch.countDown(); + } + }); + sub2.connect(); + sub2.subscribe(topic, 1); + assertTrue(retainLatch.await(5, TimeUnit.SECONDS), "New subscriber on broker 2 should receive the retained message"); + assertEquals("retained-2", retainedReceived.get()); + sub2.disconnect(); + sub2.close(); + } + + @Test + @Timeout(60) + public void testPartitionedSubscriptionsWithMirroredPublishes() throws Exception { + ActiveMQServer server1 = createMirroredServer(0, BROKER1_PORT, BROKER2_PORT, "mirrorToServer2"); + ActiveMQServer server2 = createMirroredServer(1, BROKER2_PORT, BROKER1_PORT, "mirrorToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + Wait.assertTrue(() -> server1.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer2") != null); + Wait.assertTrue(() -> server2.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer1") != null); + + final String sportsTopic = "news/sports"; + final String techTopic = "news/tech"; + + // subscribers are partitioned across brokers: sub-a on broker 1, sub-b on broker 2 + CountDownLatch sportsLatch = new CountDownLatch(2); + CopyOnWriteArrayList sportsReceived = new CopyOnWriteArrayList<>(); + MqttClient subA = createPahoClient("sub-a", BROKER1_PORT); + subA.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + sportsReceived.add(new String(m.getPayload(), StandardCharsets.UTF_8)); + sportsLatch.countDown(); + } + }); + subA.connect(); + subA.subscribe(sportsTopic, 1); + + CountDownLatch techLatch = new CountDownLatch(2); + CopyOnWriteArrayList techReceived = new CopyOnWriteArrayList<>(); + MqttClient subB = createPahoClient("sub-b", BROKER2_PORT); + subB.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + techReceived.add(new String(m.getPayload(), StandardCharsets.UTF_8)); + techLatch.countDown(); + } + }); + subB.connect(); + subB.subscribe(techTopic, 1); + + // wait for subscription queues to be mirrored to both brokers + String sportsQueueName = MQTTUtil.getCoreQueueFromMqttTopic(sportsTopic, "sub-a", server1.getConfiguration().getWildcardConfiguration()); + String techQueueName = MQTTUtil.getCoreQueueFromMqttTopic(techTopic, "sub-b", server2.getConfiguration().getWildcardConfiguration()); + Wait.assertTrue(() -> server2.locateQueue(sportsQueueName) != null, 5000, 100); + Wait.assertTrue(() -> server1.locateQueue(techQueueName) != null, 5000, 100); + + // publishers can connect to either broker — messages reach the right subscriber via mirroring + MqttClient pub1 = createPahoClient("pub1", BROKER2_PORT); + pub1.connect(); + pub1.publish(sportsTopic, "goal".getBytes(StandardCharsets.UTF_8), 1, false); + pub1.disconnect(); + pub1.close(); + + MqttClient pub2 = createPahoClient("pub2", BROKER1_PORT); + pub2.connect(); + pub2.publish(techTopic, "update".getBytes(StandardCharsets.UTF_8), 1, false); + pub2.disconnect(); + pub2.close(); + + // local publishes also work + MqttClient pub3 = createPahoClient("pub3", BROKER1_PORT); + pub3.connect(); + pub3.publish(sportsTopic, "try".getBytes(StandardCharsets.UTF_8), 1, false); + pub3.disconnect(); + pub3.close(); + + MqttClient pub4 = createPahoClient("pub4", BROKER2_PORT); + pub4.connect(); + pub4.publish(techTopic, "release".getBytes(StandardCharsets.UTF_8), 1, false); + pub4.disconnect(); + pub4.close(); + + assertTrue(sportsLatch.await(10, TimeUnit.SECONDS), "sub-a should receive both sports messages"); + assertTrue(techLatch.await(10, TimeUnit.SECONDS), "sub-b should receive both tech messages"); + + assertEquals(2, sportsReceived.size()); + assertTrue(sportsReceived.contains("goal"), "sub-a should receive 'goal' published on broker 2"); + assertTrue(sportsReceived.contains("try"), "sub-a should receive 'try' published on broker 1"); + + assertEquals(2, techReceived.size()); + assertTrue(techReceived.contains("update"), "sub-b should receive 'update' published on broker 1"); + assertTrue(techReceived.contains("release"), "sub-b should receive 'release' published on broker 2"); + + subA.disconnect(); + subA.close(); + subB.disconnect(); + subB.close(); + } + + @Test + @Timeout(60) + public void testPartitionedSubscriptionsMessageOnlyMirror() throws Exception { + ActiveMQServer server1 = createMessageOnlyMirroredServer(0, BROKER1_PORT, BROKER2_PORT, "mirrorToServer2"); + ActiveMQServer server2 = createMessageOnlyMirroredServer(1, BROKER2_PORT, BROKER1_PORT, "mirrorToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + Wait.assertTrue(() -> server1.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer2") != null); + Wait.assertTrue(() -> server2.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer1") != null); + + final String sportsTopic = "news/sports"; + final String techTopic = "news/tech"; + + // subscribers are partitioned: sub-a on broker 1 for sports, sub-b on broker 2 for tech + CountDownLatch sportsLatch = new CountDownLatch(2); + CopyOnWriteArrayList sportsReceived = new CopyOnWriteArrayList<>(); + MqttClient subA = createPahoClient("sub-a", BROKER1_PORT); + subA.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + sportsReceived.add(new String(m.getPayload(), StandardCharsets.UTF_8)); + sportsLatch.countDown(); + } + }); + subA.connect(); + subA.subscribe(sportsTopic, 1); + + CountDownLatch techLatch = new CountDownLatch(2); + CopyOnWriteArrayList techReceived = new CopyOnWriteArrayList<>(); + MqttClient subB = createPahoClient("sub-b", BROKER2_PORT); + subB.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + techReceived.add(new String(m.getPayload(), StandardCharsets.UTF_8)); + techLatch.countDown(); + } + }); + subB.connect(); + subB.subscribe(techTopic, 1); + + // no waiting for queue mirroring — subscription queues are NOT mirrored + // the mirror source forwards messages even with no local bindings + + // publish on the broker that has NO local subscriber for that topic + MqttClient pub1 = createPahoClient("pub1", BROKER2_PORT); + pub1.connect(); + pub1.publish(sportsTopic, "goal".getBytes(StandardCharsets.UTF_8), 1, false); + pub1.disconnect(); + pub1.close(); + + MqttClient pub2 = createPahoClient("pub2", BROKER1_PORT); + pub2.connect(); + pub2.publish(techTopic, "update".getBytes(StandardCharsets.UTF_8), 1, false); + pub2.disconnect(); + pub2.close(); + + // also publish locally to verify local delivery still works + MqttClient pub3 = createPahoClient("pub3", BROKER1_PORT); + pub3.connect(); + pub3.publish(sportsTopic, "try".getBytes(StandardCharsets.UTF_8), 1, false); + pub3.disconnect(); + pub3.close(); + + MqttClient pub4 = createPahoClient("pub4", BROKER2_PORT); + pub4.connect(); + pub4.publish(techTopic, "release".getBytes(StandardCharsets.UTF_8), 1, false); + pub4.disconnect(); + pub4.close(); + + assertTrue(sportsLatch.await(10, TimeUnit.SECONDS), "sub-a should receive both sports messages"); + assertTrue(techLatch.await(10, TimeUnit.SECONDS), "sub-b should receive both tech messages"); + + assertEquals(2, sportsReceived.size()); + assertTrue(sportsReceived.contains("goal"), "sub-a should receive 'goal' published on broker 2"); + assertTrue(sportsReceived.contains("try"), "sub-a should receive 'try' published on broker 1"); + + assertEquals(2, techReceived.size()); + assertTrue(techReceived.contains("update"), "sub-b should receive 'update' published on broker 1"); + assertTrue(techReceived.contains("release"), "sub-b should receive 'release' published on broker 2"); + + // verify subscription queues were NOT mirrored + String sportsQueueName = MQTTUtil.getCoreQueueFromMqttTopic(sportsTopic, "sub-a", server1.getConfiguration().getWildcardConfiguration()); + String techQueueName = MQTTUtil.getCoreQueueFromMqttTopic(techTopic, "sub-b", server2.getConfiguration().getWildcardConfiguration()); + assertNull(server2.locateQueue(sportsQueueName), "sports queue should NOT exist on broker 2"); + assertNull(server1.locateQueue(techQueueName), "tech queue should NOT exist on broker 1"); + + subA.disconnect(); + subA.close(); + subB.disconnect(); + subB.close(); + } + + @Test + @Timeout(60) + public void testOnlyMulticastMirroredWithNoBindings() throws Exception { + ActiveMQServer server1 = createMessageOnlyMirroredServer(0, BROKER1_PORT, BROKER2_PORT, "mirrorToServer2"); + ActiveMQServer server2 = createMessageOnlyMirroredServer(1, BROKER2_PORT, BROKER1_PORT, "mirrorToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + Wait.assertTrue(() -> server1.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer2") != null); + Wait.assertTrue(() -> server2.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer1") != null); + + // set up an anycast address on broker 1 (no queue) and a queue on broker 2 to receive if mirrored + server1.addAddressInfo(new AddressInfo(SimpleString.of("test.anycast"), RoutingType.ANYCAST)); + server2.createQueue(QueueConfiguration.of("test.anycast").setRoutingType(RoutingType.ANYCAST)); + + // set up a multicast subscriber on broker 2 — no local sub on broker 1 + final String topic = "test/multicast"; + CountDownLatch multicastLatch = new CountDownLatch(1); + AtomicReference multicastReceived = new AtomicReference<>(); + MqttClient sub = createPahoClient("subscriber", BROKER2_PORT); + sub.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + multicastReceived.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + multicastLatch.countDown(); + } + }); + sub.connect(); + sub.subscribe(topic, 1); + + // send an anycast message on broker 1 with no local queue — should NOT be mirrored + CoreMessage anycastMsg = new CoreMessage().initBuffer(1024); + anycastMsg.setAddress("test.anycast"); + anycastMsg.getBodyBuffer().writeString("anycast-payload"); + anycastMsg.setMessageID(server1.getStorageManager().generateID()); + server1.getPostOffice().route(anycastMsg, new RoutingContextImpl(null), true); + + // publish a multicast message on broker 1 with no local subscriber — should be mirrored + MqttClient pub = createPahoClient("producer", BROKER1_PORT); + pub.connect(); + pub.publish(topic, "multicast-payload".getBytes(StandardCharsets.UTF_8), 1, false); + pub.disconnect(); + pub.close(); + + // multicast message arrives on broker 2 — proves the mirror link works + assertTrue(multicastLatch.await(10, TimeUnit.SECONDS), "multicast message should be mirrored to broker 2"); + assertEquals("multicast-payload", multicastReceived.get()); + + // anycast queue on broker 2 should still be empty — anycast was NOT mirrored + assertEquals(0, server2.locateQueue("test.anycast").getMessageCount(), + "anycast message with no local bindings should NOT be mirrored"); + + sub.disconnect(); + sub.close(); + } + + @Test + @Timeout(60) + public void testConcurrentRetainedUpdatesOnMirrorPair() throws Exception { + ActiveMQServer server1 = createMirroredServer(0, BROKER1_PORT, BROKER2_PORT, "mirrorToServer2"); + ActiveMQServer server2 = createMirroredServer(1, BROKER2_PORT, BROKER1_PORT, "mirrorToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + Wait.assertTrue(() -> server1.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer2") != null); + Wait.assertTrue(() -> server2.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer1") != null); + + final String topic = "test/retain/concurrent"; + final String retainQueueName = MQTTUtil.getCoreRetainAddressFromMqttTopic(topic, server1.getConfiguration().getWildcardConfiguration()); + + // publish different retained messages from each broker concurrently + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(2); + + Thread t1 = new Thread(() -> { + try { + MqttClient pub = createPahoClient("pub1", BROKER1_PORT); + pub.connect(); + startLatch.await(5, TimeUnit.SECONDS); + pub.publish(topic, "from-broker-1".getBytes(StandardCharsets.UTF_8), 1, true); + pub.disconnect(); + pub.close(); + } catch (Exception e) { + throw new RuntimeException(e); + } finally { + doneLatch.countDown(); + } + }); + + Thread t2 = new Thread(() -> { + try { + MqttClient pub = createPahoClient("pub2", BROKER2_PORT); + pub.connect(); + startLatch.await(5, TimeUnit.SECONDS); + pub.publish(topic, "from-broker-2".getBytes(StandardCharsets.UTF_8), 1, true); + pub.disconnect(); + pub.close(); + } catch (Exception e) { + throw new RuntimeException(e); + } finally { + doneLatch.countDown(); + } + }); + + t1.start(); + t2.start(); + startLatch.countDown(); + assertTrue(doneLatch.await(10, TimeUnit.SECONDS), "Both publishers should complete"); + + // wait for retain queues to be populated on both brokers (may have >1 message + // due to concurrent mirroring — MQTTRetainMessageManager handles this by + // browsing to the last message and cleaning up extras on delivery) + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue q1 = server1.locateQueue(retainQueueName); + org.apache.activemq.artemis.core.server.Queue q2 = server2.locateQueue(retainQueueName); + return q1 != null && q1.getMessageCount() >= 1 && q2 != null && q2.getMessageCount() >= 1; + }, 10000, 100); + + // both brokers should serve a valid retained message to new subscribers + CountDownLatch latch1 = new CountDownLatch(1); + AtomicReference received1 = new AtomicReference<>(); + MqttClient sub1 = createPahoClient("sub1", BROKER1_PORT); + sub1.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received1.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch1.countDown(); + } + }); + sub1.connect(); + sub1.subscribe(topic, 1); + assertTrue(latch1.await(5, TimeUnit.SECONDS), "Subscriber on broker 1 should receive the retained message"); + sub1.disconnect(); + sub1.close(); + + CountDownLatch latch2 = new CountDownLatch(1); + AtomicReference received2 = new AtomicReference<>(); + MqttClient sub2 = createPahoClient("sub2", BROKER2_PORT); + sub2.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received2.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch2.countDown(); + } + }); + sub2.connect(); + sub2.subscribe(topic, 1); + assertTrue(latch2.await(5, TimeUnit.SECONDS), "Subscriber on broker 2 should receive the retained message"); + sub2.disconnect(); + sub2.close(); + + // both brokers should serve a valid retained message (either "from-broker-1" or "from-broker-2") + assertTrue("from-broker-1".equals(received1.get()) || "from-broker-2".equals(received1.get()), + "Broker 1 should serve a valid retained message, got: " + received1.get()); + assertTrue("from-broker-1".equals(received2.get()) || "from-broker-2".equals(received2.get()), + "Broker 2 should serve a valid retained message, got: " + received2.get()); + } + + @Test + @Timeout(60) + public void testRetainedMessageWithMessageOnlyMirror() throws Exception { + ActiveMQServer server1 = createMessageOnlyMirroredServer(0, BROKER1_PORT, BROKER2_PORT, "mirrorToServer2"); + ActiveMQServer server2 = createMessageOnlyMirroredServer(1, BROKER2_PORT, BROKER1_PORT, "mirrorToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + Wait.assertTrue(() -> server1.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer2") != null); + Wait.assertTrue(() -> server2.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer1") != null); + + final String topic = "test/retain/msgonly"; + final String payload = "retained-msg-only-mirror"; + final String retainQueueName = MQTTUtil.getCoreRetainAddressFromMqttTopic(topic, server1.getConfiguration().getWildcardConfiguration()); + + // subscribe on broker 2 first so the subscription queue exists locally + CountDownLatch latch = new CountDownLatch(1); + AtomicReference received = new AtomicReference<>(); + MqttClient sub = createPahoClient("subscriber", BROKER2_PORT); + sub.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch.countDown(); + } + }); + sub.connect(); + sub.subscribe(topic, 1); + + // publish retained on broker 1 + MqttClient producer = createPahoClient("producer", BROKER1_PORT); + producer.connect(); + producer.publish(topic, payload.getBytes(StandardCharsets.UTF_8), 1, true); + producer.disconnect(); + producer.close(); + + assertTrue(latch.await(10, TimeUnit.SECONDS), "Subscriber on broker 2 should receive the mirrored message"); + assertEquals(payload, received.get()); + + // retain queue on server1 should be populated + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server1.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 5000, 100); + + // retain queue on server2 should be created locally by the plugin (not via queue creation mirroring) + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server2.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 10000, 100); + + sub.disconnect(); + sub.close(); + + // a new subscriber on broker 2 should get the retained message from the local retain queue + CountDownLatch retainLatch = new CountDownLatch(1); + AtomicReference retainReceived = new AtomicReference<>(); + MqttClient sub2 = createPahoClient("subscriber2", BROKER2_PORT); + sub2.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + retainReceived.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + retainLatch.countDown(); + } + }); + sub2.connect(); + sub2.subscribe(topic, 1); + assertTrue(retainLatch.await(5, TimeUnit.SECONDS), "New subscriber on broker 2 should receive the retained message"); + assertEquals(payload, retainReceived.get()); + sub2.disconnect(); + sub2.close(); + } + + @Test + @Timeout(60) + public void testRetainedMessageSurvivesBrokerStop() throws Exception { + ActiveMQServer server1 = createMirroredServer(0, BROKER1_PORT, BROKER2_PORT, "mirrorToServer2"); + ActiveMQServer server2 = createMirroredServer(1, BROKER2_PORT, BROKER1_PORT, "mirrorToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + Wait.assertTrue(() -> server1.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer2") != null); + Wait.assertTrue(() -> server2.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer1") != null); + + final String topic = "test/retain/failover"; + final String payload = "retained-survives-stop"; + final String retainQueueName = MQTTUtil.getCoreRetainAddressFromMqttTopic(topic, server1.getConfiguration().getWildcardConfiguration()); + + // publish retained on broker 1 + MqttClient producer = createPahoClient("producer", BROKER1_PORT); + producer.connect(); + producer.publish(topic, payload.getBytes(StandardCharsets.UTF_8), 1, true); + producer.disconnect(); + producer.close(); + + // wait for retain queue on both brokers + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server1.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 5000, 100); + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server2.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 10000, 100); + + // stop broker 1 + server1.stop(); + + // a new subscriber on broker 2 should still get the retained message from the local retain queue + CountDownLatch latch = new CountDownLatch(1); + AtomicReference received = new AtomicReference<>(); + MqttClient sub = createPahoClient("subscriber", BROKER2_PORT); + sub.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch.countDown(); + } + }); + sub.connect(); + sub.subscribe(topic, 1); + assertTrue(latch.await(5, TimeUnit.SECONDS), "Subscriber on broker 2 should receive the retained message after broker 1 stopped"); + assertEquals(payload, received.get()); + sub.disconnect(); + sub.close(); + } + + @Test + @Timeout(60) + public void testRetainedMessageMirroredWithNoLocalSubscribers() throws Exception { + ActiveMQServer server1 = createMessageOnlyMirroredServer(0, BROKER1_PORT, BROKER2_PORT, "mirrorToServer2"); + ActiveMQServer server2 = createMessageOnlyMirroredServer(1, BROKER2_PORT, BROKER1_PORT, "mirrorToServer1"); + + server1.start(); + server1.waitForActivation(10, TimeUnit.SECONDS); + server2.start(); + server2.waitForActivation(10, TimeUnit.SECONDS); + + Wait.assertTrue(() -> server1.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer2") != null); + Wait.assertTrue(() -> server2.locateQueue("$ACTIVEMQ_ARTEMIS_MIRROR_mirrorToServer1") != null); + + final String topic = "test/retain/nosub"; + final String payload = "retained-no-subscribers"; + final String retainQueueName = MQTTUtil.getCoreRetainAddressFromMqttTopic(topic, server1.getConfiguration().getWildcardConfiguration()); + + // publish a retained message on broker 1 with NO subscribers anywhere + MqttClient producer = createPahoClient("producer", BROKER1_PORT); + producer.connect(); + producer.publish(topic, payload.getBytes(StandardCharsets.UTF_8), 1, true); + producer.disconnect(); + producer.close(); + + // retain queue on server1 should be populated locally by the plugin + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server1.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 5000, 100); + + // retain queue on server2 should also be populated — message mirrored via NO_BINDINGS path + Wait.assertTrue(() -> { + org.apache.activemq.artemis.core.server.Queue queue = server2.locateQueue(retainQueueName); + return queue != null && queue.getMessageCount() == 1; + }, 10000, 100); + + // a new subscriber on broker 2 should receive the retained message + CountDownLatch latch = new CountDownLatch(1); + AtomicReference received = new AtomicReference<>(); + MqttClient sub = createPahoClient("subscriber", BROKER2_PORT); + sub.setCallback(new DefaultMqttCallback() { + @Override + public void messageArrived(String t, MqttMessage m) { + received.set(new String(m.getPayload(), StandardCharsets.UTF_8)); + latch.countDown(); + } + }); + sub.connect(); + sub.subscribe(topic, 1); + assertTrue(latch.await(5, TimeUnit.SECONDS), "New subscriber on broker 2 should receive the retained message"); + assertEquals(payload, received.get()); + sub.disconnect(); + sub.close(); + } + + private interface DefaultMqttCallback extends MqttCallback { + + @Override + default void disconnected(MqttDisconnectResponse disconnectResponse) { + } + + @Override + default void mqttErrorOccurred(MqttException exception) { + } + + @Override + default void messageArrived(String topic, MqttMessage message) { + } + + @Override + default void deliveryComplete(IMqttToken token) { + } + + @Override + default void connectComplete(boolean reconnect, String serverURI) { + } + + @Override + default void authPacketArrived(int reasonCode, MqttProperties properties) { + } + } +}