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.integration.amqp.connect;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertInstanceOf;
  import static org.junit.jupiter.api.Assertions.assertNull;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  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.MessageConsumer;
  import javax.jms.MessageProducer;
  import javax.jms.Queue;
  import javax.jms.Session;
  import javax.jms.TemporaryQueue;
  import javax.jms.TextMessage;
  import java.io.PrintStream;
  import java.net.URI;
  import java.util.Enumeration;
  import java.util.Map;
  import java.util.concurrent.TimeUnit;
  
  import org.apache.activemq.artemis.api.core.Message;
  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.cli.commands.tools.PrintData;
  import org.apache.activemq.artemis.core.config.CoreAddressConfiguration;
  import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectConfiguration;
  import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPMirrorBrokerConnectionElement;
  import org.apache.activemq.artemis.core.filter.Filter;
  import org.apache.activemq.artemis.core.server.ActiveMQServer;
  import org.apache.activemq.artemis.core.server.impl.AddressInfo;
  import org.apache.activemq.artemis.core.server.impl.LastValueQueue;
  import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
  import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
  import org.apache.activemq.artemis.selector.filter.Filterable;
  import org.apache.activemq.artemis.tests.integration.amqp.AmqpClientTestSupport;
  import org.apache.activemq.artemis.tests.util.CFUtil;
  import org.apache.activemq.artemis.utils.RandomUtil;
  import org.apache.activemq.artemis.utils.StringPrintStream;
  import org.apache.activemq.artemis.utils.Wait;
  import org.apache.activemq.transport.amqp.client.AmqpClient;
  import org.apache.activemq.transport.amqp.client.AmqpConnection;
  import org.apache.activemq.transport.amqp.client.AmqpMessage;
  import org.apache.activemq.transport.amqp.client.AmqpReceiver;
  import org.apache.activemq.transport.amqp.client.AmqpSender;
  import org.apache.activemq.transport.amqp.client.AmqpSession;
  import org.junit.jupiter.api.AfterEach;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import java.lang.invoke.MethodHandles;
  
  public class BrokerInSyncTest extends AmqpClientTestSupport {
  
     protected static final int AMQP_PORT_2 = 5673;
     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
     ActiveMQServer server_2;
     private AssertionLoggerHandler loggerHandler;
  
     @AfterEach
     public void stopServer1() throws Exception {
        if (server != null) {
           server.stop();
        }
     }
  
     @AfterEach
     public void stopServer2() throws Exception {
        if (server_2 != null) {
           server_2.stop();
        }
     }
  
     @BeforeEach
     public void startLogging() {
        loggerHandler = new AssertionLoggerHandler();
  
     }
  
     @AfterEach
     public void stopLogging() throws Exception {
        try {
           assertFalse(loggerHandler.findText("AMQ222214"));
        } finally {
           loggerHandler.close();
        }
     }
  
     @Override
     protected ActiveMQServer createServer() throws Exception {
        return createServer(AMQP_PORT, false);
     }
  
     @Test
     public void testSyncOnCreateQueues() throws Exception {
        server.setIdentity("Server1");
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server.getConfiguration().addAMQPConnection(amqpConnection);
        }
        server.start();
  
        server_2 = createServer(AMQP_PORT_2, false);
        server_2.setIdentity("Server2");
  
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server_2.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server_2.start();
  
        server_2.addAddressInfo(new AddressInfo("sometest").setAutoCreated(false));
        server_2.createQueue(QueueConfiguration.of("sometest").setDurable(true));
  
        Wait.assertTrue(() -> server_2.locateQueue("sometest") != null);
        Wait.assertTrue(() -> server.locateQueue("sometest") != null);
  
        server.addAddressInfo(new AddressInfo("OnServer1").setAutoCreated(false));
        server.createQueue(QueueConfiguration.of("OnServer1").setDurable(true));
  
        Wait.assertTrue(() -> server.locateQueue("OnServer1") != null);
        Wait.assertTrue("Sync is not working on the way back", () -> server_2.locateQueue("OnServer1") != null, 2000);
  
        Wait.assertTrue(() -> server_2.locateQueue("sometest") != null);
        Wait.assertTrue(() -> server.locateQueue("sometest") != null);
  
        for (int i = 0; i < 10; i++) {
           final int queueID = i;
           server_2.createQueue(QueueConfiguration.of("test2_" + i).setDurable(true));
           server.createQueue(QueueConfiguration.of("test1_" + i).setDurable(true));
           Wait.assertTrue(() -> server.locateQueue("test2_" + queueID) != null);
           Wait.assertTrue(() -> server.locateQueue("test1_" + queueID) != null);
           Wait.assertTrue(() -> server_2.locateQueue("test2_" + queueID) != null);
           Wait.assertTrue(() -> server_2.locateQueue("test1_" + queueID) != null);
        }
  
        server_2.stop();
        server.stop();
     }
  
     @Test
     public void testSingleMessage() throws Exception {
        testSingleMessage("AMQP");
     }
  
     @Test
     public void testSingleMessageCore() throws Exception {
        testSingleMessage("CORE");
     }
  
     @Test
     public void testSingleMessageOpenWire() throws Exception {
        testSingleMessage("OPENWIRE");
     }
  
     public void testSingleMessage(String protocol) throws Exception {
        server.setIdentity("Server1");
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server.getConfiguration().addAMQPConnection(amqpConnection);
        }
        server.start();
  
        server_2 = createServer(AMQP_PORT_2, false);
        server_2.setIdentity("Server2");
  
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server_2.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server_2.start();
  
        server_2.addAddressInfo(new AddressInfo(getQueueName()).setAutoCreated(false).addRoutingType(RoutingType.ANYCAST));
        server_2.createQueue(QueueConfiguration.of(getQueueName()).setDurable(true).setRoutingType(RoutingType.ANYCAST));
  
        Wait.assertTrue(() -> server_2.locateQueue(getQueueName()) != null);
        Wait.assertTrue(() -> server.locateQueue(getQueueName()) != null);
  
        ConnectionFactory cf1 = CFUtil.createConnectionFactory(protocol, "tcp://localhost:" + AMQP_PORT);
        Connection connection1 = cf1.createConnection();
        Session session1 = connection1.createSession(true, Session.SESSION_TRANSACTED);
        connection1.start();
  
        ConnectionFactory cf2 = CFUtil.createConnectionFactory(protocol, "tcp://localhost:" + AMQP_PORT_2);
        Connection connection2 = cf2.createConnection();
        Session session2 = connection2.createSession(true, Session.SESSION_TRANSACTED);
        connection2.start();
  
        Queue queue = session1.createQueue(getQueueName());
  
        MessageProducer producerServer1 = session1.createProducer(queue);
        MessageProducer producerServer2 = session2.createProducer(queue);
  
        TextMessage message = session1.createTextMessage("test");
        message.setIntProperty("i", 0);
        message.setStringProperty("server", server.getIdentity());
        producerServer1.send(message);
        session1.commit();
  
        org.apache.activemq.artemis.core.server.Queue queueOnServer1 = server.locateQueue(getQueueName());
        org.apache.activemq.artemis.core.server.Queue queueOnServer2 = server_2.locateQueue(getQueueName());
        assertNotNull(queueOnServer1);
        assertNotNull(queueOnServer2);
  
        Wait.assertEquals(1, queueOnServer1::getMessageCount);
        Wait.assertEquals(1, queueOnServer2::getMessageCount);
  
        message = session1.createTextMessage("test");
        message.setIntProperty("i", 1);
        message.setStringProperty("server", server_2.getIdentity());
        producerServer2.send(message);
        session2.commit();
  
        if (logger.isDebugEnabled() && !Wait.waitFor(() -> queueOnServer1.getMessageCount() == 2)) {
           debugData();
        }
  
        Wait.assertEquals(2, queueOnServer1::getMessageCount);
        Wait.assertEquals(2, queueOnServer2::getMessageCount);
  
  
        connection2.start();
        try (MessageConsumer consumer = session2.createConsumer(queue)) {
           javax.jms.Message receivedMessage = consumer.receive(5000);
           assertNotNull(message);
           checkProperties(connection2, receivedMessage);
           session2.commit();
        }
  
        Wait.assertEquals(1L, queueOnServer1::getMessageCount, 5000, 100);
        Wait.assertEquals(1L, queueOnServer2::getMessageCount, 5000, 100);
  
        connection1.start();
        try (MessageConsumer consumer = session1.createConsumer(queue)) {
           javax.jms.Message receivedMessage = consumer.receive(5000);
           assertNotNull(message);
           checkProperties(connection1, receivedMessage);
           session1.commit();
        }
  
        connection1.close();
        connection2.close();
  
        Wait.assertEquals(0L, queueOnServer1::getMessageCount, 5000, 100);
        Wait.assertEquals(0L, queueOnServer2::getMessageCount, 5000, 100);
  
        server_2.stop();
        server.stop();
     }
  
  
     private void checkProperties(Connection connection, javax.jms.Message message) throws Exception {
        try (Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
           TemporaryQueue temporaryQueue = session.createTemporaryQueue();
           MessageProducer producer = session.createProducer(temporaryQueue);
           producer.send(message);
           connection.start();
           MessageConsumer consumer = session.createConsumer(temporaryQueue);
           javax.jms.Message receivedMessage = consumer.receive(5000);
           assertNotNull(receivedMessage);
  
           // The cleanup for x-opt happens on server's side.
           // we may receive if coming directly from a mirrored queue,
           // however we should cleanup on the next send to avoid invalid IDs on the server
           Enumeration propertyNames = receivedMessage.getPropertyNames();
           while (propertyNames.hasMoreElements()) {
              String property = String.valueOf(propertyNames.nextElement());
              assertFalse(property.startsWith("x-opt"));
           }
        }
     }
  
  
     private org.apache.activemq.artemis.core.server.Queue locateQueueWithWait(ActiveMQServer server, String queueName) throws Exception {
        Wait.assertTrue(() -> server.locateQueue(queueName) != null);
        org.apache.activemq.artemis.core.server.Queue queue = server.locateQueue(queueName);
        assertNotNull(queue);
        return queue;
     }
  
     @Test
     public void testExpiryNoReaper() throws Exception {
        internalExpiry(false);
     }
  
     @Test
     public void testExpiry() throws Exception {
        internalExpiry(true);
     }
  
     private void internalExpiry(boolean useReaper) throws Exception {
        server.setIdentity("Server1");
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("to_2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server.getConfiguration().addAddressSetting("#", new AddressSettings().setExpiryAddress(SimpleString.of("expiryQueue")));
  
        server.getConfiguration().addAddressConfiguration(new CoreAddressConfiguration().setName("expiryQueue"));
        server.getConfiguration().addQueueConfiguration(QueueConfiguration.of("expiryQueue").setRoutingType(RoutingType.ANYCAST));
        if (!useReaper) {
           server.getConfiguration().setMessageExpiryScanPeriod(-1);
        }
  
        server.start();
  
        server_2 = createServer(AMQP_PORT_2, false);
  
        server.getConfiguration().addAddressConfiguration(new CoreAddressConfiguration().setName("expiryQueue"));
        server.getConfiguration().addQueueConfiguration(QueueConfiguration.of("expiryQueue").setRoutingType(RoutingType.ANYCAST));
        if (!useReaper) {
           server_2.getConfiguration().setMessageExpiryScanPeriod(-1);
        }
  
        server_2.setIdentity("Server2");
  
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("to_1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server_2.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server_2.getConfiguration().addAddressSetting("#", new AddressSettings().setExpiryAddress(SimpleString.of("expiryQueue")));
  
        server_2.start();
  
        org.apache.activemq.artemis.core.server.Queue to1 = locateQueueWithWait(server_2, "$ACTIVEMQ_ARTEMIS_MIRROR_to_1");
        org.apache.activemq.artemis.core.server.Queue to2 = locateQueueWithWait(server, "$ACTIVEMQ_ARTEMIS_MIRROR_to_2");
        org.apache.activemq.artemis.core.server.Queue expiry1 = locateQueueWithWait(server, "expiryQueue");
        org.apache.activemq.artemis.core.server.Queue expiry2 = locateQueueWithWait(server_2, "expiryQueue");
  
        server_2.addAddressInfo(new AddressInfo(getQueueName()).setAutoCreated(false).addRoutingType(RoutingType.ANYCAST));
        server_2.createQueue(QueueConfiguration.of(getQueueName()).setDurable(true).setRoutingType(RoutingType.ANYCAST));
  
        org.apache.activemq.artemis.core.server.Queue queueOnServer1 = locateQueueWithWait(server, getQueueName());
        org.apache.activemq.artemis.core.server.Queue queueOnServer2 = locateQueueWithWait(server_2, getQueueName());
  
        ConnectionFactory cf1 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
        Connection connection1 = cf1.createConnection();
        Session session1 = connection1.createSession(true, Session.SESSION_TRANSACTED);
        connection1.start();
  
        Queue queue = session1.createQueue(getQueueName());
  
        MessageProducer producerServer1 = session1.createProducer(queue);
  
        producerServer1.setTimeToLive(1000);
  
        TextMessage message = session1.createTextMessage("test");
        message.setIntProperty("i", 0);
        message.setStringProperty("server", server.getIdentity());
        producerServer1.send(message);
        session1.commit();
  
        Wait.assertEquals(1L, queueOnServer1::getMessageCount, 1000, 100);
        Wait.assertEquals(1L, queueOnServer2::getMessageCount, 1000, 100);
  
        if (useReaper) {
           // if using the reaper, I want to test what would happen with races between the reaper and the Mirror target
           // for that reason we only pause the SNFs if we are using the reaper
           to1.pause();
           to2.pause();
        }
  
        Thread.sleep(1500);
        if (!useReaper) {
           queueOnServer1.expireReferences(); // we will expire in just on queue hoping the other gets through mirror
        }
  
        Wait.assertEquals(0L, queueOnServer1::getMessageCount, 2000, 100);
        Wait.assertEquals(0L, queueOnServer2::getMessageCount, 2000, 100);
  
        Wait.assertEquals(1L, expiry1::getMessageCount, 1000, 100);
        Wait.assertEquals(1L, expiry2::getMessageCount, 1000, 100);
  
        to1.resume();
        to2.resume();
  
        if (!useReaper) {
           queueOnServer1.expireReferences(); // in just one queue
        }
  
        Wait.assertEquals(1L, expiry1::getMessageCount, 1000, 100);
        Wait.assertEquals(1L, expiry2::getMessageCount, 1000, 100);
  
        connection1.close();
  
        server_2.stop();
        server.stop();
     }
  
     @Test
     public void testDLA() throws Exception {
        server.setIdentity("Server1");
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("to_2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server.getConfiguration().addAddressSetting("#", new AddressSettings().setDeadLetterAddress(SimpleString.of("deadLetterQueue")).setMaxDeliveryAttempts(2));
  
        server.getConfiguration().addAddressConfiguration(new CoreAddressConfiguration().setName("deadLetterQueue"));
        server.getConfiguration().addQueueConfiguration(QueueConfiguration.of("deadLetterQueue").setRoutingType(RoutingType.ANYCAST));
        server.getConfiguration().setMessageExpiryScanPeriod(-1);
  
        server.start();
  
        server_2 = createServer(AMQP_PORT_2, false);
  
        server.getConfiguration().addAddressConfiguration(new CoreAddressConfiguration().setName("deadLetterQueue"));
        server.getConfiguration().addQueueConfiguration(QueueConfiguration.of("deadLetterQueue").setRoutingType(RoutingType.ANYCAST));
        server_2.getConfiguration().setMessageExpiryScanPeriod(-1);
  
        server_2.setIdentity("Server2");
  
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("to_1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server_2.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server_2.getConfiguration().addAddressSetting("#", new AddressSettings().setDeadLetterAddress(SimpleString.of("deadLetterQueue")).setMaxDeliveryAttempts(2));
  
        server_2.start();
  
        org.apache.activemq.artemis.core.server.Queue to1 = locateQueueWithWait(server_2, "$ACTIVEMQ_ARTEMIS_MIRROR_to_1");
        org.apache.activemq.artemis.core.server.Queue to2 = locateQueueWithWait(server, "$ACTIVEMQ_ARTEMIS_MIRROR_to_2");
        org.apache.activemq.artemis.core.server.Queue dlq1 = locateQueueWithWait(server, "deadLetterQueue");
        org.apache.activemq.artemis.core.server.Queue dlq2 = locateQueueWithWait(server_2, "deadLetterQueue");
  
        server_2.addAddressInfo(new AddressInfo(getQueueName()).setAutoCreated(false).addRoutingType(RoutingType.ANYCAST));
        server_2.createQueue(QueueConfiguration.of(getQueueName()).setDurable(true).setRoutingType(RoutingType.ANYCAST));
  
        org.apache.activemq.artemis.core.server.Queue queueOnServer1 = locateQueueWithWait(server, getQueueName());
        org.apache.activemq.artemis.core.server.Queue queueOnServer2 = locateQueueWithWait(server_2, getQueueName());
  
        ConnectionFactory cf1 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
        Connection connection1 = cf1.createConnection();
        Session session1 = connection1.createSession(true, Session.SESSION_TRANSACTED);
        connection1.start();
  
        Queue queue = session1.createQueue(getQueueName());
  
        MessageProducer producerServer1 = session1.createProducer(queue);
  
        TextMessage message = session1.createTextMessage("test");
        message.setIntProperty("i", 0);
        message.setStringProperty("server", server.getIdentity());
        producerServer1.send(message);
        session1.commit();
  
  
        MessageConsumer consumer1 = session1.createConsumer(queue);
        connection1.start();
  
        for (int i = 0; i < 2; i++) {
           TextMessage messageToCancel = (TextMessage) consumer1.receive(1000);
           assertNotNull(messageToCancel);
           session1.rollback();
        }
  
        assertNull(consumer1.receiveNoWait());
  
        Wait.assertEquals(0L, queueOnServer1::getMessageCount, 1000, 100);
        Wait.assertEquals(0L, queueOnServer2::getMessageCount, 1000, 100);
  
        Wait.assertEquals(1L, dlq1::getMessageCount, 1000, 100);
        Wait.assertEquals(1L, dlq2::getMessageCount, 1000, 100);
  
  
        dlq1.retryMessages(new Filter() {
           @Override
           public boolean match(Message message) {
              return true;
           }
  
           @Override
           public boolean match(Map<String, String> map) {
              return true;
           }
  
           @Override
           public boolean match(Filterable filterable) {
              return true;
           }
  
           @Override
           public SimpleString getFilterString() {
              return SimpleString.of("Test");
           }
        });
  
        Wait.assertEquals(1L, queueOnServer1::getMessageCount, 1000, 100);
        Wait.assertEquals(1L, queueOnServer2::getMessageCount, 1000, 100);
  
        Wait.assertEquals(0L, dlq1::getMessageCount, 1000, 100);
        Wait.assertEquals(0L, dlq2::getMessageCount, 1000, 100);
  
  
        connection1.close();
  
        server_2.stop();
        server.stop();
     }
  
  
     @Test
     public void testCreateInternalQueue() throws Exception {
        server.setIdentity("Server1");
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("to_2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server.getConfiguration().addAddressSetting("#", new AddressSettings().setDeadLetterAddress(SimpleString.of("deadLetterQueue")).setMaxDeliveryAttempts(2));
  
        server.getConfiguration().addAddressConfiguration(new CoreAddressConfiguration().setName("deadLetterQueue"));
        server.getConfiguration().addQueueConfiguration(QueueConfiguration.of("deadLetterQueue").setRoutingType(RoutingType.ANYCAST));
        server.getConfiguration().setMessageExpiryScanPeriod(-1);
  
        server.start();
  
        server_2 = createServer(AMQP_PORT_2, false);
  
        server.getConfiguration().addAddressConfiguration(new CoreAddressConfiguration().setName("deadLetterQueue"));
        server.getConfiguration().addQueueConfiguration(QueueConfiguration.of("deadLetterQueue").setRoutingType(RoutingType.ANYCAST));
        server_2.getConfiguration().setMessageExpiryScanPeriod(-1);
  
        server_2.setIdentity("Server2");
  
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("to_1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server_2.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server_2.getConfiguration().addAddressSetting("#", new AddressSettings().setDeadLetterAddress(SimpleString.of("deadLetterQueue")).setMaxDeliveryAttempts(2));
  
        server_2.start();
  
        org.apache.activemq.artemis.core.server.Queue to1 = locateQueueWithWait(server_2, "$ACTIVEMQ_ARTEMIS_MIRROR_to_1");
        assertNotNull(to1);
  
        long messagesAddedOnS2 = to1.getMessagesAdded();
  
        String internalQueueName = getQueueName() + "Internal";
  
        server.addAddressInfo(new AddressInfo(internalQueueName).setAutoCreated(false).addRoutingType(RoutingType.ANYCAST).setInternal(true));
        server.createQueue(QueueConfiguration.of(internalQueueName).setDurable(true).setRoutingType(RoutingType.ANYCAST).setInternal(true));
  
        server.addAddressInfo(new AddressInfo(getQueueName()).setAutoCreated(false).addRoutingType(RoutingType.ANYCAST));
        server.createQueue(QueueConfiguration.of(getQueueName()).setDurable(true).setRoutingType(RoutingType.ANYCAST));
  
        Wait.assertTrue(() -> server_2.locateQueue(getQueueName()) != null, 5000);
        assertNull(server_2.getAddressInfo(SimpleString.of(internalQueueName)));
        assertNull(server_2.locateQueue(internalQueueName));
  
        assertEquals(messagesAddedOnS2, to1.getMessagesAdded());
  
        server_2.stop();
        server.stop();
     }
  
  
  
  
     @Test
     public void testLVQ() throws Exception {
        AssertionLoggerHandler loggerHandler = new AssertionLoggerHandler();
        runAfter(() -> loggerHandler.close());
  
        server.setIdentity("Server1");
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("to_2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server.getConfiguration().addAddressSetting("#", new AddressSettings().setDeadLetterAddress(SimpleString.of("deadLetterQueue")).setMaxDeliveryAttempts(2));
  
        server.getConfiguration().addAddressConfiguration(new CoreAddressConfiguration().setName("deadLetterQueue"));
        server.getConfiguration().addQueueConfiguration(QueueConfiguration.of("deadLetterQueue").setRoutingType(RoutingType.ANYCAST));
  
        String lvqName = "testLVQ_" + RandomUtil.randomUUIDString();
  
        server.getConfiguration().addAddressConfiguration(new CoreAddressConfiguration().setName(lvqName));
        server.getConfiguration().addQueueConfiguration(QueueConfiguration.of(lvqName).setRoutingType(RoutingType.ANYCAST).setLastValue(true).setLastValueKey("KEY"));
        server.getConfiguration().setMessageExpiryScanPeriod(-1);
  
        server.start();
  
        server_2 = createServer(AMQP_PORT_2, false);
  
        server.getConfiguration().addAddressConfiguration(new CoreAddressConfiguration().setName("deadLetterQueue"));
        server.getConfiguration().addQueueConfiguration(QueueConfiguration.of("deadLetterQueue").setRoutingType(RoutingType.ANYCAST));
        server_2.getConfiguration().setMessageExpiryScanPeriod(-1);
  
        server_2.setIdentity("Server2");
  
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("to_1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server_2.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server_2.getConfiguration().addAddressSetting("#", new AddressSettings().setDeadLetterAddress(SimpleString.of("deadLetterQueue")).setMaxDeliveryAttempts(2));
        server_2.start();
  
        org.apache.activemq.artemis.core.server.Queue lvqQueue1 = locateQueueWithWait(server, lvqName);
        org.apache.activemq.artemis.core.server.Queue lvqQueue2 = locateQueueWithWait(server, lvqName);
  
        assertTrue(lvqQueue1.isLastValue());
        assertTrue(lvqQueue2.isLastValue());
        assertInstanceOf(LastValueQueue.class, lvqQueue1);
        assertInstanceOf(LastValueQueue.class, lvqQueue2);
        assertEquals("KEY", lvqQueue1.getQueueConfiguration().getLastValueKey().toString());
        assertEquals("KEY", lvqQueue2.getQueueConfiguration().getLastValueKey().toString());
  
        ConnectionFactory cf1 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
        try (Connection connection1 = cf1.createConnection()) {
           Session session1 = connection1.createSession(false, Session.AUTO_ACKNOWLEDGE);
           Queue queue = session1.createQueue(lvqName);
           MessageProducer producerServer1 = session1.createProducer(queue);
           connection1.start();
           for (int i = 0; i < 1000; i++) {
              TextMessage message = session1.createTextMessage("test");
              message.setIntProperty("i", 0);
              message.setStringProperty("KEY", "" + (i % 10));
              producerServer1.send(message);
           }
        }
        assertFalse(loggerHandler.findText("AMQ222214"));
  
        Wait.assertEquals(10L, lvqQueue1::getMessageCount, 2000, 100);
        Wait.assertEquals(10L, lvqQueue2::getMessageCount, 2000, 100);
  
  
        server_2.stop();
        server.stop();
  
        assertFalse(loggerHandler.findText("AMQ222153"));
     }
  
  
     @Test
     public void testSyncData() throws Exception {
        int NUMBER_OF_MESSAGES = 100;
        server.setIdentity("Server1");
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server.getConfiguration().addAMQPConnection(amqpConnection);
        }
        server.start();
  
        server_2 = createServer(AMQP_PORT_2, false);
        server_2.setIdentity("Server2");
  
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server_2.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server_2.start();
  
        server_2.addAddressInfo(new AddressInfo(getQueueName()).setAutoCreated(false).addRoutingType(RoutingType.ANYCAST));
        server_2.createQueue(QueueConfiguration.of(getQueueName()).setDurable(true).setRoutingType(RoutingType.ANYCAST));
  
        Wait.assertTrue(() -> server_2.locateQueue(getQueueName()) != null);
        Wait.assertTrue(() -> server.locateQueue(getQueueName()) != null);
  
        ConnectionFactory cf1 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
        Connection connection1 = cf1.createConnection();
        Session session1 = connection1.createSession(true, Session.SESSION_TRANSACTED);
        connection1.start();
  
        ConnectionFactory cf2 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);
        Connection connection2 = cf2.createConnection();
        Session session2 = connection2.createSession(true, Session.SESSION_TRANSACTED);
        connection2.start();
  
        Queue queue = session1.createQueue(getQueueName());
  
        MessageProducer producerServer1 = session1.createProducer(queue);
        MessageProducer producerServer2 = session2.createProducer(queue);
  
        for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
           TextMessage message = session1.createTextMessage("test " + i);
           message.setIntProperty("i", i);
           message.setStringProperty("server", server.getIdentity());
           producerServer1.send(message);
        }
        session1.commit();
  
        org.apache.activemq.artemis.core.server.Queue queueOnServer1 = server.locateQueue(getQueueName());
        org.apache.activemq.artemis.core.server.Queue queueOnServer2 = server_2.locateQueue(getQueueName());
        assertNotNull(queueOnServer1);
        assertNotNull(queueOnServer2);
  
        Wait.assertEquals(NUMBER_OF_MESSAGES, queueOnServer1::getMessageCount);
        Wait.assertEquals(NUMBER_OF_MESSAGES, queueOnServer2::getMessageCount);
  
        for (int i = NUMBER_OF_MESSAGES; i < NUMBER_OF_MESSAGES * 2; i++) {
           TextMessage message = session1.createTextMessage("test " + i);
           message.setIntProperty("i", i);
           message.setStringProperty("server", server_2.getIdentity());
           producerServer2.send(message);
        }
        session2.commit();
  
        if (logger.isDebugEnabled() && !Wait.waitFor(() -> queueOnServer1.getMessageCount() == NUMBER_OF_MESSAGES * 2)) {
           debugData();
        }
  
        Wait.assertEquals(NUMBER_OF_MESSAGES * 2, queueOnServer1::getMessageCount);
        Wait.assertEquals(NUMBER_OF_MESSAGES * 2, queueOnServer2::getMessageCount);
  
        MessageConsumer consumerOn1 = session1.createConsumer(queue);
        for (int i = 0; i < NUMBER_OF_MESSAGES * 2; i++) {
           TextMessage message = (TextMessage) consumerOn1.receive(5000);
           logger.debug("### Client acking message({}) on server 1, a message that was original sent on {} text = {}", i, message.getStringProperty("server"), message.getText());
           assertNotNull(message);
           assertEquals(i, message.getIntProperty("i"));
           assertEquals("test " + i, message.getText());
           session1.commit();
        }
  
        boolean bothConsumed = Wait.waitFor(() -> {
           long q1 = queueOnServer1.getMessageCount();
           long q2 = queueOnServer2.getMessageCount();
           logger.debug("Queue on Server 1 = {}", q1);
           logger.debug("Queue on Server 2 = {}", q2);
           return q1 == 0 && q2 == 0;
        }, 5_000, 1000);
  
        if (logger.isDebugEnabled() && !bothConsumed) {
           debugData();
           fail("q1 = " + queueOnServer1.getMessageCount() + ", q2 = " + queueOnServer2.getMessageCount());
        }
  
        assertEquals(0, queueOnServer1.getMessageCount());
        assertEquals(0, queueOnServer2.getConsumerCount());
  
        System.out.println("Queue on Server 1 = " + queueOnServer1.getMessageCount());
        System.out.println("Queue on Server 2 = " + queueOnServer2.getMessageCount());
  
        assertEquals(0, queueOnServer1.getDeliveringCount());
        assertEquals(0, queueOnServer2.getDeliveringCount());
        assertEquals(0, queueOnServer1.getDurableDeliveringCount());
        assertEquals(0, queueOnServer2.getDurableDeliveringCount());
        assertEquals(0, queueOnServer1.getDurableDeliveringSize());
        assertEquals(0, queueOnServer2.getDurableDeliveringSize());
        assertEquals(0, queueOnServer1.getDeliveringSize());
        assertEquals(0, queueOnServer2.getDeliveringSize());
  
        server_2.stop();
        server.stop();
     }
  
  
  
     @Test
     public void testStats() throws Exception {
        int NUMBER_OF_MESSAGES = 1;
        server.setIdentity("Server1");
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server.getConfiguration().addAMQPConnection(amqpConnection);
        }
        server.start();
  
        server_2 = createServer(AMQP_PORT_2, false);
        server_2.setIdentity("Server2");
  
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server_2.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server_2.start();
  
        server_2.addAddressInfo(new AddressInfo(getQueueName()).setAutoCreated(false).addRoutingType(RoutingType.ANYCAST));
        server_2.createQueue(QueueConfiguration.of(getQueueName()).setDurable(true).setRoutingType(RoutingType.ANYCAST));
  
        Wait.assertTrue(() -> server_2.locateQueue(getQueueName()) != null);
        Wait.assertTrue(() -> server.locateQueue(getQueueName()) != null, 5000);
  
        ConnectionFactory cf1 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
        Connection connection1 = cf1.createConnection();
        Session session1 = connection1.createSession(true, Session.SESSION_TRANSACTED);
        connection1.start();
  
        ConnectionFactory cf2 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);
        Connection connection2 = cf2.createConnection();
        Session session2 = connection2.createSession(true, Session.SESSION_TRANSACTED);
        connection2.start();
  
        Queue queue = session1.createQueue(getQueueName());
  
        MessageProducer producerServer1 = session1.createProducer(queue);
        MessageProducer producerServer2 = session2.createProducer(queue);
  
        for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
           TextMessage message = session1.createTextMessage("test " + i);
           message.setIntProperty("i", i);
           message.setStringProperty("server", server.getIdentity());
           producerServer1.send(message);
        }
        session1.commit();
  
        org.apache.activemq.artemis.core.server.Queue queueOnServer1 = server.locateQueue(getQueueName());
        org.apache.activemq.artemis.core.server.Queue queueOnServer2 = server_2.locateQueue(getQueueName());
        assertNotNull(queueOnServer1);
        assertNotNull(queueOnServer2);
  
        Wait.assertEquals(NUMBER_OF_MESSAGES, queueOnServer1::getMessageCount);
        Wait.assertEquals(NUMBER_OF_MESSAGES, queueOnServer2::getMessageCount);
  
        assertEquals(0, queueOnServer1.getDeliveringSize());
        assertEquals(0, queueOnServer2.getDeliveringSize());
  
        MessageConsumer consumerOn1 = session1.createConsumer(queue);
        for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
           TextMessage message = (TextMessage) consumerOn1.receive(5000);
           logger.debug("### Client acking message({}) on server 1, a message that was original sent on {} text = {}", i, message.getStringProperty("server"), message.getText());
           assertNotNull(message);
           assertEquals(i, message.getIntProperty("i"));
           assertEquals("test " + i, message.getText());
           session1.commit();
        }
  
        Wait.assertEquals(0L, queueOnServer1::getMessageCount, 2000, 100);
        Wait.assertEquals(0L, queueOnServer2::getMessageCount, 2000, 100);
        assertEquals(0, queueOnServer1.getDeliveringCount());
        assertEquals(0, queueOnServer2.getDeliveringCount());
        assertEquals(0, queueOnServer1.getDurableDeliveringCount());
        assertEquals(0, queueOnServer2.getDurableDeliveringCount());
        assertEquals(0, queueOnServer1.getDurableDeliveringSize());
        assertEquals(0, queueOnServer2.getDurableDeliveringSize());
        assertEquals(0, queueOnServer1.getDeliveringSize());
        assertEquals(0, queueOnServer2.getDeliveringSize());
  
        server_2.stop();
        server.stop();
     }
  
     private void debugData() throws Exception {
        StringPrintStream stringPrintStream = new StringPrintStream();
        PrintStream out = stringPrintStream.newStream();
        org.apache.activemq.artemis.core.server.Queue queueToDebugOn1 = server.locateQueue(getQueueName());
        org.apache.activemq.artemis.core.server.Queue queueToDebugOn2 = server_2.locateQueue(getQueueName());
        out.println("*******************************************************************************************************************************");
        out.println("Queue on Server 1 with count = " + queueToDebugOn1.getMessageCount());
        queueToDebugOn1.forEach((r) -> out.println("Server1 has reference " + r.getMessage()));
        out.println("*******************************************************************************************************************************");
        out.println("Queue on Server 2 with count = " + queueToDebugOn2.getMessageCount());
        queueToDebugOn2.forEach((r) -> out.println("Server2 has reference " + r.getMessage()));
        out.println("*******************************************************************************************************************************");
        out.println("PrintData Server 1");
        PrintData.printMessages(server.getConfiguration().getJournalLocation(), out, false, false, true, false);
        out.println("*******************************************************************************************************************************");
        out.println("PrintData Server 2");
        PrintData.printMessages(server_2.getConfiguration().getJournalLocation(), out, false, false, true, false);
        logger.debug("Data Available on Servers:\n{}", stringPrintStream.toString());
     }
  
     @Test
     public void testSyncDataNoSuppliedID() throws Exception {
        int NUMBER_OF_MESSAGES = 100;
        server.setIdentity("Server1");
  
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server.getConfiguration().addAMQPConnection(amqpConnection);
        }
        server.start();
  
        server_2 = createServer(AMQP_PORT_2, false);
        server_2.setIdentity("Server2");
  
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server_2.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server_2.start();
  
        server_2.addAddressInfo(new AddressInfo(getQueueName()).setAutoCreated(false).addRoutingType(RoutingType.ANYCAST));
        server_2.createQueue(QueueConfiguration.of(getQueueName()).setDurable(true).setRoutingType(RoutingType.ANYCAST));
  
        Wait.assertTrue(() -> server_2.locateQueue(getQueueName()) != null);
        Wait.assertTrue(() -> server.locateQueue(getQueueName()) != null);
  
        AmqpClient cf1 = new AmqpClient(new URI("tcp://localhost:" + AMQP_PORT), null, null);
        AmqpConnection connection1 = cf1.createConnection();
        connection1.connect();
        AmqpSession session1 = connection1.createSession();
  
        AmqpClient cf2 = new AmqpClient(new URI("tcp://localhost:" + AMQP_PORT_2), null, null);
        AmqpConnection connection2 = cf2.createConnection();
        connection2.connect();
        AmqpSession session2 = connection2.createSession();
  
        AmqpSender producerServer1 = session1.createSender(getQueueName());
        AmqpSender producerServer2 = session2.createSender(getQueueName());
  
        for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
           AmqpMessage message = new AmqpMessage();
           message.setDurable(true);
           message.setApplicationProperty("i", i);
           producerServer1.send(message);
        }
  
        org.apache.activemq.artemis.core.server.Queue queueOnServer1 = server.locateQueue(getQueueName());
        org.apache.activemq.artemis.core.server.Queue queueOnServer2 = server_2.locateQueue(getQueueName());
        assertNotNull(queueOnServer1);
        assertNotNull(queueOnServer2);
  
        Wait.assertEquals(NUMBER_OF_MESSAGES, queueOnServer1::getMessageCount);
        Wait.assertEquals(NUMBER_OF_MESSAGES, queueOnServer2::getMessageCount);
  
        for (int i = NUMBER_OF_MESSAGES; i < NUMBER_OF_MESSAGES * 2; i++) {
           AmqpMessage message = new AmqpMessage();
           message.setDurable(true);
           message.setApplicationProperty("i", i);
           producerServer2.send(message);
        }
  
        Wait.assertEquals(NUMBER_OF_MESSAGES * 2, queueOnServer1::getMessageCount);
        Wait.assertEquals(NUMBER_OF_MESSAGES * 2, queueOnServer2::getMessageCount);
  
        AmqpReceiver consumerOn1 = session1.createReceiver(getQueueName());
        consumerOn1.flow(NUMBER_OF_MESSAGES * 2 + 1);
        for (int i = 0; i < NUMBER_OF_MESSAGES * 2; i++) {
           AmqpMessage message = consumerOn1.receive(5, TimeUnit.SECONDS);
           assertNotNull(message);
           message.accept();
           assertEquals(i, (int) message.getApplicationProperty("i"));
        }
  
        Wait.assertEquals(0, queueOnServer1::getMessageCount);
        Wait.assertEquals(0, () -> {
           System.out.println(queueOnServer2.getMessageCount());
           return queueOnServer2.getMessageCount();
        });
  
        connection1.close();
        connection2.close();
        server_2.stop();
        server.stop();
     }
  
     @Test
     public void testLargeMessageInSync() throws Exception {
        String queueName = "testSyncLargeMessage";
        server.setIdentity("Server1");
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server.getConfiguration().addAMQPConnection(amqpConnection);
        }
        server.start();
  
        server_2 = createServer(AMQP_PORT_2, false);
        server_2.setIdentity("Server2");
  
        {
           AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
           amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
           server_2.getConfiguration().addAMQPConnection(amqpConnection);
        }
  
        server_2.start();
  
        server_2.addAddressInfo(new AddressInfo(queueName).setAutoCreated(false).addRoutingType(RoutingType.ANYCAST));
        server_2.createQueue(QueueConfiguration.of(queueName).setDurable(true).setRoutingType(RoutingType.ANYCAST));
  
        Wait.assertTrue(() -> server_2.locateQueue(queueName) != null);
        Wait.assertTrue(() -> server.locateQueue(queueName) != null);
  
        String bigString = RandomUtil.randomAlphaNumericString(200 * 1024);
  
        ConnectionFactory factory1 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
        ConnectionFactory factory2 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);
  
        try (Connection connection = factory1.createConnection()) {
           Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
           Queue queue = session.createQueue(queueName);
           MessageProducer producer = session.createProducer(queue);
           producer.send(session.createTextMessage(bigString));
        }
  
        try (Connection connection = factory2.createConnection()) {
           org.apache.activemq.artemis.core.server.Queue serverQueue = server.locateQueue(queueName);
           Wait.assertEquals(1, serverQueue::getMessageCount);
           Wait.assertEquals(1, () -> getNumberOfFiles(server_2.getConfiguration().getLargeMessagesLocation()), 5000, 100);
           Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
           Queue queue = session.createQueue(queueName);
           MessageConsumer consumer = session.createConsumer(queue);
           connection.start();
           TextMessage message = (TextMessage)consumer.receive(5000);
           assertNotNull(message);
           assertEquals(bigString, message.getText());
           Wait.assertEquals(0, () -> getNumberOfFiles(server_2.getConfiguration().getLargeMessagesLocation()), 5000, 100);
        }
  
        try (Connection connection = factory1.createConnection()) {
           org.apache.activemq.artemis.core.server.Queue serverQueue = server.locateQueue(queueName);
  
           Wait.assertEquals(0L, serverQueue::getMessageCount, 2000, 100);
  
           Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
           Queue queue = session.createQueue(queueName);
           MessageConsumer consumer = session.createConsumer(queue);
           connection.start();
           TextMessage message = (TextMessage)consumer.receiveNoWait();
           assertNull(message);
           server.stop();
           server_2.stop();
           Wait.assertEquals(0, () -> getNumberOfFiles(server.getConfiguration().getLargeMessagesLocation()), 1000, 100);
        }
     }
  
  }
  