content: /*
   * 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.soak.brokerConnection.mirror;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertNull;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  import static org.junit.jupiter.api.Assertions.fail;
  
  import javax.jms.Connection;
  import javax.jms.ConnectionFactory;
  import javax.jms.Message;
  import javax.jms.MessageConsumer;
  import javax.jms.MessageProducer;
  import javax.jms.Queue;
  import javax.jms.Session;
  import javax.jms.TextMessage;
  import javax.jms.Topic;
  import java.io.File;
  import java.io.StringWriter;
  import java.lang.invoke.MethodHandles;
  import java.util.HashSet;
  import java.util.Properties;
  import java.util.Set;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.Executor;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicInteger;
  
  import org.apache.activemq.artemis.api.core.QueueConfiguration;
  import org.apache.activemq.artemis.api.core.RoutingType;
  import org.apache.activemq.artemis.api.core.management.SimpleManagement;
  import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectionAddressType;
  import org.apache.activemq.artemis.core.server.ActiveMQServer;
  import org.apache.activemq.artemis.core.server.impl.AddressInfo;
  import org.apache.activemq.artemis.tests.soak.SoakTestBase;
  import org.apache.activemq.artemis.tests.util.CFUtil;
  import org.apache.activemq.artemis.utils.RandomUtil;
  import org.apache.activemq.artemis.util.ServerUtil;
  import org.apache.activemq.artemis.utils.FileUtil;
  import org.apache.activemq.artemis.utils.Wait;
  import org.apache.activemq.artemis.cli.commands.helper.HelperCreate;
  import org.junit.jupiter.api.Test;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  public class ClusteredMirrorSoakTest extends SoakTestBase {
  
     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  
     private static String largeBody;
     private static String smallBody = "This is a small body";
  
     static {
        StringWriter writer = new StringWriter();
        while (writer.getBuffer().length() < 1024 * 1024) {
           writer.append("This is a large string ..... ");
        }
        largeBody = writer.toString();
     }
  
     public static final String DC1_NODE_A = "ClusteredMirrorSoakTest/DC1/A";
     public static final String DC2_NODE_A = "ClusteredMirrorSoakTest/DC2/A";
     public static final String DC1_NODE_B = "ClusteredMirrorSoakTest/DC1/B";
     public static final String DC2_NODE_B = "ClusteredMirrorSoakTest/DC2/B";
  
     Process processDC1_node_A;
     Process processDC1_node_B;
     Process processDC2_node_A;
     Process processDC2_node_B;
  
     private static String DC1_NODEA_URI = "tcp://localhost:61616";
     private static String DC1_NODEB_URI = "tcp://localhost:61617";
     private static String DC2_NODEA_URI = "tcp://localhost:61618";
     private static String DC2_NODEB_URI = "tcp://localhost:61619";
  
     private static void createServer(String serverName, String connectionName, String clusterURI, String mirrorURI, int portOffset, boolean paging) throws Exception {
        File serverLocation = getFileServerLocation(serverName);
        deleteDirectory(serverLocation);
  
        HelperCreate cliCreateServer = helperCreate();
        cliCreateServer.setAllowAnonymous(true).setArtemisInstance(serverLocation);
        cliCreateServer.setMessageLoadBalancing("ON_DEMAND");
        cliCreateServer.setClustered(true);
        cliCreateServer.setNoWeb(true);
        cliCreateServer.setStaticCluster(clusterURI);
        cliCreateServer.setArgs("--no-stomp-acceptor", "--no-hornetq-acceptor", "--no-mqtt-acceptor", "--no-amqp-acceptor", "--max-hops", "1", "--name", DC1_NODE_A);
        cliCreateServer.addArgs("--addresses", "order");
        cliCreateServer.addArgs("--queues", "myQueue");
        cliCreateServer.setPortOffset(portOffset);
        cliCreateServer.createServer();
  
        Properties brokerProperties = new Properties();
        brokerProperties.put("AMQPConnections." + connectionName + ".uri", mirrorURI);
        brokerProperties.put("AMQPConnections." + connectionName + ".retryInterval", "1000");
        brokerProperties.put("AMQPConnections." + connectionName + ".type", AMQPBrokerConnectionAddressType.MIRROR.toString());
        brokerProperties.put("AMQPConnections." + connectionName + ".connectionElements.mirror.sync", "false");
        brokerProperties.put("largeMessageSync", "false");
        File brokerPropertiesFile = new File(serverLocation, "broker.properties");
        saveProperties(brokerProperties, brokerPropertiesFile);
  
        File brokerXml = new File(serverLocation, "/etc/broker.xml");
        assertTrue(brokerXml.exists());
        // Adding redistribution delay to broker configuration
        assertTrue(FileUtil.findReplace(brokerXml, "<address-setting match=\"#\">", "<address-setting match=\"#\">\n\n" + "            <redistribution-delay>0</redistribution-delay> <!-- added by ClusteredMirrorSoakTest.java --> \n"));
        if (paging) {
           assertTrue(FileUtil.findReplace(brokerXml, "<max-size-messages>-1</max-size-messages>", "<max-size-messages>1</max-size-messages>"));
        }
     }
  
     public static void createRealServers(boolean paging) throws Exception {
        createServer(DC1_NODE_A, "mirror", DC1_NODEB_URI, DC2_NODEA_URI, 0, paging);
        createServer(DC1_NODE_B, "mirror", DC1_NODEA_URI, DC2_NODEB_URI, 1, paging);
        createServer(DC2_NODE_A, "mirror", DC2_NODEB_URI, DC1_NODEA_URI, 2, paging);
        createServer(DC2_NODE_B, "mirror", DC2_NODEA_URI, DC1_NODEB_URI, 3, paging);
     }
  
     private void startServers() throws Exception {
        processDC1_node_A = startServer(DC1_NODE_A, -1, -1, new File(getServerLocation(DC1_NODE_A), "broker.properties"));
        processDC1_node_B = startServer(DC1_NODE_B, -1, -1, new File(getServerLocation(DC1_NODE_B), "broker.properties"));
        processDC2_node_A = startServer(DC2_NODE_A, -1, -1, new File(getServerLocation(DC2_NODE_A), "broker.properties"));
        processDC2_node_B = startServer(DC2_NODE_B, -1, -1, new File(getServerLocation(DC2_NODE_B), "broker.properties"));
  
        ServerUtil.waitForServerToStart(0, 10_000);
        ServerUtil.waitForServerToStart(1, 10_000);
        ServerUtil.waitForServerToStart(2, 10_000);
        ServerUtil.waitForServerToStart(3, 10_000);
     }
  
     @Test
     public void testAvoidReflections() throws Exception {
        createRealServers(true);
  
        String internalQueue = "INTERNAL_QUEUE";
  
        ActiveMQServer tempServer = createServer(true);
        tempServer.getConfiguration().setBindingsDirectory(getServerLocation(DC1_NODE_A) + "/data/bindings");
        tempServer.getConfiguration().setJournalDirectory(getServerLocation(DC1_NODE_A) + "/data/journal");
        tempServer.getConfiguration().setJournalFileSize(10 * 1024 * 1024);
        tempServer.start();
        tempServer.addAddressInfo(new AddressInfo(internalQueue).addRoutingType(RoutingType.ANYCAST).setInternal(true));
        tempServer.createQueue(QueueConfiguration.of(internalQueue).setDurable(true).setRoutingType(RoutingType.ANYCAST).setInternal(true).setAddress(internalQueue));
        tempServer.stop();
  
        startServers();
  
        SimpleManagement simpleManagementDC1A = new SimpleManagement(DC1_NODEA_URI, null, null);
        SimpleManagement simpleManagementDC2A = new SimpleManagement(DC2_NODEA_URI, null, null);
        SimpleManagement simpleManagementDC1B = new SimpleManagement(DC1_NODEA_URI, null, null);
        SimpleManagement simpleManagementDC2B = new SimpleManagement(DC2_NODEB_URI, null, null);
  
        String snfQueue = "$ACTIVEMQ_ARTEMIS_MIRROR_mirror";
  
        String queueName = "myQueue";
        String topicName = "order";
  
        for (int i = 0; i < 5; i++) {
           logger.info("DC1A={}", simpleManagementDC1A.getMessageAddedOnQueue(snfQueue));
           logger.info("DC1B={}", simpleManagementDC1B.getMessageAddedOnQueue(snfQueue));
           logger.info("DC2A={}", simpleManagementDC2A.getMessageAddedOnQueue(snfQueue));
           logger.info("DC2B={}", simpleManagementDC2B.getMessageAddedOnQueue(snfQueue));
  
           // no load generated.. just initial queues should have been sent
           assertTrue(simpleManagementDC1A.getMessageAddedOnQueue(snfQueue) < 20);
           assertTrue(simpleManagementDC2A.getMessageAddedOnQueue(snfQueue) < 20);
           assertTrue(simpleManagementDC1B.getMessageAddedOnQueue(snfQueue) < 20);
           assertTrue(simpleManagementDC2B.getMessageAddedOnQueue(snfQueue) < 20);
           Thread.sleep(100);
        }
  
        assertEquals(0, simpleManagementDC2A.getMessageCountOnQueue(queueName));
        assertEquals(0, simpleManagementDC1A.getMessageCountOnQueue(internalQueue));
        try {
           simpleManagementDC2A.getMessageCountOnQueue(internalQueue);
           fail("Exception expected");
        } catch (Exception expected) {
        }
  
        ConnectionFactory connectionFactoryDC1A = CFUtil.createConnectionFactory("amqp", DC1_NODEA_URI);
  
        int numberOfMessages = 1_000;
  
        try (Connection connection = connectionFactoryDC1A.createConnection()) {
           connection.setClientID("conn1");
           Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
           Topic topic = session.createTopic(topicName);
           MessageConsumer con = session.createDurableConsumer(topic, "hello1");
           MessageConsumer con2 = session.createDurableConsumer(topic, "hello2");
  
           MessageProducer producer = session.createProducer(topic);
           for (int i = 0; i < numberOfMessages; i++) {
              if (i % 100 == 0) {
                 logger.info("Sent topic {}", i);
              }
              producer.send(session.createTextMessage("hello " + i));
           }
           session.commit();
  
        }
  
        try (Connection connection = connectionFactoryDC1A.createConnection()) {
           Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
           Queue queue = session.createQueue(queueName);
  
           MessageProducer producer = session.createProducer(queue);
           for (int i = 0; i < numberOfMessages; i++) {
              if (i % 100 == 0) {
                 logger.info("Sent queue {}", i);
              }
              producer.send(session.createTextMessage("hello " + i));
           }
           session.commit();
        }
  
        Wait.assertEquals(numberOfMessages, () -> simpleManagementDC1A.getMessageCountOnQueue(queueName), 5000);
        Wait.assertEquals(numberOfMessages, () -> simpleManagementDC2A.getMessageCountOnQueue(queueName), 5000);
        Wait.assertEquals(numberOfMessages, () -> simpleManagementDC1A.getMessageCountOnQueue("conn1.hello2"), 5000);
        Wait.assertEquals(numberOfMessages, () -> simpleManagementDC1A.getMessageCountOnQueue("conn1.hello2"), 5000);
        Wait.assertEquals(numberOfMessages, () -> simpleManagementDC2A.getMessageCountOnQueue("conn1.hello2"), 5000);
        Wait.assertEquals(numberOfMessages, () -> simpleManagementDC2A.getMessageCountOnQueue("conn1.hello2"), 5000);
  
        try (Connection connection = connectionFactoryDC1A.createConnection()) {
           connection.setClientID("conn1");
           connection.start();
           Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
           Topic topic = session.createTopic(topicName);
           Queue queue = session.createQueue(queueName);
           MessageConsumer[] consumers = new MessageConsumer[] {session.createDurableSubscriber(topic, "hello1"), session.createDurableSubscriber(topic, "hello2"), session.createConsumer(queue)};
  
           for (MessageConsumer c : consumers) {
              for (int i = 0; i < numberOfMessages; i++) {
                 assertNotNull(c.receive(5000));
                 if (i % 100 == 0) {
                    session.commit();
                 }
              }
              session.commit();
           }
        }
  
        Wait.assertEquals(0, () -> simpleManagementDC1A.getMessageCountOnQueue(queueName), 5000);
        Wait.assertEquals(0, () -> simpleManagementDC2A.getMessageCountOnQueue(queueName), 5000);
        Wait.assertEquals(0, () -> simpleManagementDC1A.getMessageCountOnQueue("conn1.hello2"), 5000);
        Wait.assertEquals(0, () -> simpleManagementDC1A.getMessageCountOnQueue("conn1.hello2"), 5000);
        Wait.assertEquals(0, () -> simpleManagementDC2A.getMessageCountOnQueue("conn1.hello2"), 5000);
        Wait.assertEquals(0, () -> simpleManagementDC2A.getMessageCountOnQueue("conn1.hello2"), 5000);
  
        long countDC1A = simpleManagementDC1A.getMessageAddedOnQueue(snfQueue);
        long countDC1B = simpleManagementDC1B.getMessageAddedOnQueue(snfQueue);
  
        for (int i = 0; i < 10; i++) {
           // DC1 should be quiet and nothing moving out of it
           assertEquals(countDC1A, simpleManagementDC1A.getMessageAddedOnQueue(snfQueue));
           assertEquals(countDC1B, simpleManagementDC1B.getMessageAddedOnQueue(snfQueue));
  
           // DC2 is totally passive, nothing should have been generated
           assertTrue(simpleManagementDC2A.getMessageAddedOnQueue(snfQueue) < 20);
           assertTrue(simpleManagementDC2B.getMessageAddedOnQueue(snfQueue) < 20);
           // we take intervals, allowing to make sure it doesn't grow
           Thread.sleep(100);
           logger.info("DC1A={}", simpleManagementDC1A.getMessageAddedOnQueue(snfQueue));
           logger.info("DC1B={}", simpleManagementDC1B.getMessageAddedOnQueue(snfQueue));
           logger.info("DC2A={}", simpleManagementDC2A.getMessageAddedOnQueue(snfQueue));
           logger.info("DC2B={}", simpleManagementDC2B.getMessageAddedOnQueue(snfQueue));
        }
     }
  
     @Test
     public void testSimpleQueue() throws Exception {
        createRealServers(false);
        startServers();
  
        final int numberOfMessages = 200;
  
        assertEquals(0, numberOfMessages % 2, "numberOfMessages must be even");
  
        ConnectionFactory connectionFactoryDC1A = CFUtil.createConnectionFactory("amqp", DC1_NODEA_URI);
        ConnectionFactory connectionFactoryDC2A = CFUtil.createConnectionFactory("amqp", DC2_NODEA_URI);
        String snfQueue = "$ACTIVEMQ_ARTEMIS_MIRROR_mirror";
  
        SimpleManagement simpleManagementDC1A = new SimpleManagement(DC1_NODEA_URI, null, null);
  
        try (Connection connection = connectionFactoryDC1A.createConnection()) {
           Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
           Queue queue = session.createQueue("myQueue");
           MessageProducer producer = session.createProducer(queue);
  
           for (int i = 0; i < numberOfMessages; i++) {
              TextMessage message;
              boolean large;
              if (i % 1 == 2) {
                 message = session.createTextMessage(largeBody);
                 large = true;
              } else {
                 message = session.createTextMessage(smallBody);
                 large = false;
              }
              message.setIntProperty("i", i);
              message.setBooleanProperty("large", large);
              producer.send(message);
              if (i % 100 == 0) {
                 logger.debug("commit {}", i);
                 session.commit();
              }
           }
           session.commit();
        }
  
        logger.debug("All messages were sent");
  
        try (Connection connection = connectionFactoryDC1A.createConnection()) {
           connection.start();
           Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
           Queue queue = session.createQueue("myQueue");
           MessageConsumer consumer = session.createConsumer(queue);
  
           for (int i = 0; i < numberOfMessages / 2; i++) {
              TextMessage message = (TextMessage) consumer.receive(5000);
              assertNotNull(message);
              logger.debug("Received message {}, large={}", message.getIntProperty("i"), message.getBooleanProperty("large"));
           }
           session.commit();
        }
  
        processDC2_node_A.destroyForcibly();
        processDC2_node_A.waitFor();
        processDC2_node_A = startServer(DC2_NODE_A, 2, 5000, new File(getServerLocation(DC2_NODE_A), "broker.properties"));
  
        Wait.assertEquals(0L, () -> getMessageCount(simpleManagementDC1A, snfQueue), 250_000, 1000);
  
        try (Connection connection = connectionFactoryDC2A.createConnection()) {
           connection.start();
           Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
           Queue queue = session.createQueue("myQueue");
           MessageConsumer consumer = session.createConsumer(queue);
  
           for (int i = 0; i < numberOfMessages / 2; i++) {
              TextMessage message = (TextMessage) consumer.receive(5000);
              assertNotNull(message);
              logger.debug("Received message {}, large={}", message.getIntProperty("i"), message.getBooleanProperty("large"));
           }
           session.commit();
        }
     }
  
  
     private CountDownLatch startConsumer(Executor executor, ConnectionFactory factory, String queue, AtomicBoolean running, AtomicInteger errorCount, AtomicInteger receivedCount) {
        CountDownLatch done = new CountDownLatch(1);
  
        Set<Integer> receivedMessages = new HashSet<>();
  
        executor.execute(() -> {
           try {
              try (Connection connection = factory.createConnection()) {
                 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                 MessageConsumer consumer = session.createConsumer(session.createQueue(queue));
                 connection.start();
                 while (running.get()) {
                    Message message = consumer.receive(100);
                    if (message != null) {
                       receivedCount.incrementAndGet();
                       Integer receivedI = message.getIntProperty("i");
                       if (!receivedMessages.add(receivedI)) {
                          errorCount.incrementAndGet();
                          logger.warn("Message {}, isLarge={} received in duplicate", receivedI, message.getBooleanProperty("large"));
                       }
                    }
                 }
              }
           } catch (Exception e) {
              logger.warn(e.getMessage(), e);
              errorCount.incrementAndGet();
           } finally {
              done.countDown();
           }
  
        });
  
        return done;
     }
  
     private boolean findQueue(SimpleManagement simpleManagement, String queue) {
        try {
           simpleManagement.getMessageCountOnQueue(queue);
           return true;
        } catch (Exception e) {
           return false;
        }
     }
  
     private void sendMessages(ConnectionFactory factory, String queueName, int messages, int commitInterval) throws Exception {
  
        try (Connection connection = factory.createConnection()) {
           Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
           Queue queue = session.createQueue(queueName);
           MessageProducer producer = session.createProducer(queue);
  
           for (int i = 0; i < messages; i++) {
              TextMessage message;
              boolean large;
              if (i % 1 == 2) {
                 message = session.createTextMessage(largeBody);
                 large = true;
              } else {
                 message = session.createTextMessage(smallBody);
                 large = false;
              }
              message.setIntProperty("i", i);
              message.setBooleanProperty("large", large);
              producer.send(message);
              if (i > 0 && i % commitInterval == 0) {
                 logger.debug("commit {}", i);
                 session.commit();
              }
           }
           session.commit();
        }
     }
  
     @Test
     public void testAutoCreateQueue() throws Exception {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        runAfter(executorService::shutdownNow);
  
        createRealServers(false);
        startServers();
  
        String queueName = "testqueue" + RandomUtil.randomUUIDString();
  
        final int numberOfMessages = 50;
  
        ConnectionFactory connectionFactoryDC1A = CFUtil.createConnectionFactory("amqp", DC1_NODEA_URI);
        ConnectionFactory connectionFactoryDC2A = CFUtil.createConnectionFactory("amqp", DC2_NODEA_URI);
        ConnectionFactory connectionFactoryDC2B = CFUtil.createConnectionFactory("amqp", DC2_NODEB_URI);
  
        AtomicBoolean runningConsumers = new AtomicBoolean(true);
        runAfter(() -> runningConsumers.set(false));
        AtomicInteger errors = new AtomicInteger(0);
        AtomicInteger receiverCount = new AtomicInteger(0);
  
        try (SimpleManagement simpleManagementDC1A = new SimpleManagement(DC1_NODEA_URI, null, null);
             SimpleManagement simpleManagementDC1B = new SimpleManagement(DC1_NODEB_URI, null, null);
             SimpleManagement simpleManagementDC2A = new SimpleManagement(DC2_NODEA_URI, null, null);
             SimpleManagement simpleManagementDC2B = new SimpleManagement(DC2_NODEB_URI, null, null)) {
  
           assertFalse(findQueue(simpleManagementDC1A, queueName));
           assertFalse(findQueue(simpleManagementDC1B, queueName));
           assertFalse(findQueue(simpleManagementDC2A, queueName));
           assertFalse(findQueue(simpleManagementDC2B, queueName));
  
           // just to allow auto-creation to kick in....
           assertTrue(startConsumer(executorService, connectionFactoryDC2A, queueName, new AtomicBoolean(false), errors, receiverCount).await(1, TimeUnit.MINUTES));
           assertTrue(startConsumer(executorService, connectionFactoryDC2B, queueName, new AtomicBoolean(false), errors, receiverCount).await(1, TimeUnit.MINUTES));
  
           Wait.assertTrue(() -> findQueue(simpleManagementDC1A, queueName));
           Wait.assertTrue(() -> findQueue(simpleManagementDC1B, queueName));
           Wait.assertTrue(() -> findQueue(simpleManagementDC2A, queueName));
           Wait.assertTrue(() -> findQueue(simpleManagementDC2B, queueName));
  
           sendMessages(connectionFactoryDC1A, queueName, numberOfMessages, 10);
  
           Wait.assertEquals(numberOfMessages, () -> simpleManagementDC1A.getMessageCountOnQueue(queueName), 5000);
           Wait.assertEquals(0, () -> simpleManagementDC1B.getMessageCountOnQueue(queueName), 5000);
           Wait.assertEquals(numberOfMessages, () -> simpleManagementDC2A.getMessageCountOnQueue(queueName), 5000);
           Wait.assertEquals(0, () -> simpleManagementDC2B.getMessageCountOnQueue(queueName), 5000);
  
           CountDownLatch doneDC2B = startConsumer(executorService, connectionFactoryDC2B, queueName, runningConsumers, errors, receiverCount);
           Wait.assertEquals(numberOfMessages, receiverCount::get, 30_000);
  
           Wait.assertTrue(() -> findQueue(simpleManagementDC1A, queueName));
           Wait.assertTrue(() -> findQueue(simpleManagementDC1B, queueName));
           Wait.assertTrue(() -> findQueue(simpleManagementDC2A, queueName));
           Wait.assertTrue(() -> findQueue(simpleManagementDC2B, queueName));
  
           Wait.assertEquals(0, () -> simpleManagementDC1A.getDeliveringCountOnQueue(queueName), 5000);
           Wait.assertEquals(0, () -> simpleManagementDC1B.getDeliveringCountOnQueue(queueName), 5000);
           Wait.assertEquals(0, () -> simpleManagementDC2A.getDeliveringCountOnQueue(queueName), 5000);
           Wait.assertEquals(0, () -> simpleManagementDC2B.getDeliveringCountOnQueue(queueName), 5000);
           Wait.assertEquals(0, () -> simpleManagementDC1A.getMessageCountOnQueue(queueName), 5000);
           Wait.assertEquals(0, () -> simpleManagementDC1B.getMessageCountOnQueue(queueName), 5000);
           Wait.assertEquals(0, () -> simpleManagementDC2A.getMessageCountOnQueue(queueName), 5000);
           Wait.assertEquals(0, () -> simpleManagementDC2B.getMessageCountOnQueue(queueName), 5000);
  
           runningConsumers.set(false);
  
           assertTrue(doneDC2B.await(5, TimeUnit.SECONDS));
           assertEquals(0, errors.get());
        }
     }
  
     @Test
     public void testMirroredTopics() throws Exception {
        createRealServers(false);
        startServers();
  
        final int numberOfMessages = 200;
  
        assertEquals(0, numberOfMessages % 2, "numberOfMessages must be even");
  
        String clientIDA = "nodeA";
        String clientIDB = "nodeB";
        String subscriptionID = "my-order";
        String snfQueue = "$ACTIVEMQ_ARTEMIS_MIRROR_mirror";
  
        ConnectionFactory connectionFactoryDC1A = CFUtil.createConnectionFactory("amqp", "tcp://localhost:61616");
        ConnectionFactory connectionFactoryDC1B = CFUtil.createConnectionFactory("amqp", "tcp://localhost:61617");
        ConnectionFactory connectionFactoryDC2A = CFUtil.createConnectionFactory("amqp", "tcp://localhost:61618");
        ConnectionFactory connectionFactoryDC2B = CFUtil.createConnectionFactory("amqp", "tcp://localhost:61619");
  
        SimpleManagement simpleManagementDC1B = new SimpleManagement(DC1_NODEB_URI, null, null);
        SimpleManagement simpleManagementDC2B = new SimpleManagement(DC2_NODEB_URI, null, null);
  
        consume(connectionFactoryDC1A, clientIDA, subscriptionID, 0, 0, false);
        consume(connectionFactoryDC1B, clientIDB, subscriptionID, 0, 0, false);
  
        try (Connection connection = connectionFactoryDC1B.createConnection()) {
           Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
           Topic topic = session.createTopic("order");
           MessageProducer producer = session.createProducer(topic);
  
           for (int i = 0; i < numberOfMessages; i++) {
              TextMessage message;
              boolean large;
              message = session.createTextMessage(largeBody);
              large = true;
              message.setIntProperty("i", i);
              message.setBooleanProperty("large", large);
              producer.send(message);
              if (i % 100 == 0) {
                 logger.debug("commit {}", i);
                 session.commit();
              }
           }
           session.commit();
        }
  
        logger.debug("Consuming from DC1B");
        consume(connectionFactoryDC1B, clientIDB, subscriptionID, 0, numberOfMessages / 2, false);
  
        processDC2_node_B.destroyForcibly();
        processDC2_node_B.waitFor();
        processDC2_node_B = startServer(DC2_NODE_B, 3, 5000, new File(getServerLocation(DC2_NODE_B), "broker.properties"));
  
        Wait.assertEquals(0L, () -> getMessageCount(simpleManagementDC1B, snfQueue), 250_000, 1000);
        Wait.assertEquals(numberOfMessages / 2, () -> simpleManagementDC2B.getMessageCountOnQueue("nodeB.my-order"), 10000);
  
        logger.debug("Consuming from DC2B with {}", simpleManagementDC2B.getMessageCountOnQueue("nodeB.my-order"));
  
        consume(connectionFactoryDC2B, clientIDB, subscriptionID, numberOfMessages / 2, numberOfMessages / 2, true);
  
        Wait.assertEquals(0, () -> simpleManagementDC2B.getMessageCountOnQueue("nodeB.my-order"), 10000);
  
        Wait.assertEquals(0, () -> simpleManagementDC1B.getMessageCountOnQueue("nodeB.my-order"), 10000);
        consume(connectionFactoryDC1B, clientIDB, subscriptionID, numberOfMessages, 0, true);
        logger.debug("DC1B nodeB.my-order=0");
     }
  
     private static void consume(ConnectionFactory factory, String clientID, String subscriptionID, int start, int numberOfMessages, boolean expectEmpty) throws Exception {
        try (Connection connection = factory.createConnection()) {
           connection.setClientID(clientID);
           Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
           Topic topic = session.createTopic("order");
           connection.start();
           MessageConsumer consumer = session.createDurableConsumer(topic, subscriptionID);
           boolean failed = false;
  
           for (int i = start; i < start + numberOfMessages; i++) {
              TextMessage message = (TextMessage) consumer.receive(10_000);
              assertNotNull(message);
              logger.debug("Received message {}, large={}", message.getIntProperty("i"), message.getBooleanProperty("large"));
              if (message.getIntProperty("i") != i) {
                 failed = true;
                 logger.warn("Expected message {} but got {}", i, message.getIntProperty("i"));
              }
              if (message.getBooleanProperty("large")) {
                 assertEquals(largeBody, message.getText());
              } else {
                 assertEquals(smallBody, message.getText());
              }
              logger.debug("Consumed {}, large={}", i, message.getBooleanProperty("large"));
           }
           session.commit();
  
           assertFalse(failed);
  
           if (expectEmpty) {
              assertNull(consumer.receiveNoWait());
           }
        }
  
     }
  
  }
  