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.cluster.reattach;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  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 java.util.ArrayList;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Set;
  import java.util.Timer;
  import java.util.TimerTask;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.TimeUnit;
  
  import org.apache.activemq.artemis.api.core.ActiveMQException;
  import org.apache.activemq.artemis.api.core.ActiveMQIllegalStateException;
  import org.apache.activemq.artemis.api.core.ActiveMQNotConnectedException;
  import org.apache.activemq.artemis.api.core.QueueConfiguration;
  import org.apache.activemq.artemis.api.core.SimpleString;
  import org.apache.activemq.artemis.api.core.client.ClientConsumer;
  import org.apache.activemq.artemis.api.core.client.ClientMessage;
  import org.apache.activemq.artemis.api.core.client.ClientProducer;
  import org.apache.activemq.artemis.api.core.client.ClientSession;
  import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
  import org.apache.activemq.artemis.api.core.client.MessageHandler;
  import org.apache.activemq.artemis.api.core.client.ServerLocator;
  import org.apache.activemq.artemis.core.client.impl.ClientSessionFactoryImpl;
  import org.apache.activemq.artemis.core.client.impl.ClientSessionInternal;
  import org.apache.activemq.artemis.core.remoting.impl.invm.InVMRegistry;
  import org.apache.activemq.artemis.core.server.ActiveMQServer;
  import org.apache.activemq.artemis.core.server.ActiveMQServers;
  import org.apache.activemq.artemis.jms.client.ActiveMQTextMessage;
  import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
  import org.apache.activemq.artemis.utils.Wait;
  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 final class RandomReattachIntegrationTest extends ActiveMQTestBase {
  
     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  
  
     private static final int RECEIVE_TIMEOUT = 10000;
  
  
     private static final SimpleString ADDRESS = SimpleString.of("FailoverTestAddress");
  
     private ActiveMQServer server;
  
     private Timer timer;
  
     private int getNumIterations() {
        return 2;
     }
  
  
     @Test
     public void testA() throws Exception {
        runTest(new RunnableT() {
           @Override
           public void run(final ClientSessionFactory sf) throws Exception {
              doTestA(sf);
           }
        });
     }
  
     @Test
     public void testB() throws Exception {
        runTest(new RunnableT() {
           @Override
           public void run(final ClientSessionFactory sf) throws Exception {
              doTestB(sf);
           }
        });
     }
  
     @Test
     public void testC() throws Exception {
        runTest(new RunnableT() {
           @Override
           public void run(final ClientSessionFactory sf) throws Exception {
              doTestC(sf);
           }
        });
     }
  
     @Test
     public void testD() throws Exception {
        runTest(new RunnableT() {
           @Override
           public void run(final ClientSessionFactory sf) throws Exception {
              doTestD(sf);
           }
        });
     }
  
     @Test
     public void testE() throws Exception {
        runTest(new RunnableT() {
           @Override
           public void run(final ClientSessionFactory sf) throws Exception {
              doTestE(sf);
           }
        });
     }
  
     @Test
     public void testF() throws Exception {
        runTest(new RunnableT() {
           @Override
           public void run(final ClientSessionFactory sf) throws Exception {
              doTestF(sf);
           }
        });
     }
  
     @Test
     public void testG() throws Exception {
        runTest(new RunnableT() {
           @Override
           public void run(final ClientSessionFactory sf) throws Exception {
              doTestG(sf);
           }
        });
     }
  
     @Test
     public void testH() throws Exception {
        runTest(new RunnableT() {
           @Override
           public void run(final ClientSessionFactory sf) throws Exception {
              doTestH(sf);
           }
        });
     }
  
     @Test
     public void testI() throws Exception {
        runTest(new RunnableT() {
           @Override
           public void run(final ClientSessionFactory sf) throws Exception {
              doTestI(sf);
           }
        });
     }
  
     @Test
     public void testJ() throws Exception {
        runTest(new RunnableT() {
           @Override
           public void run(final ClientSessionFactory sf) throws Exception {
              doTestJ(sf);
           }
        });
     }
  
     @Test
     public void testK() throws Exception {
        runTest(new RunnableT() {
           @Override
           public void run(final ClientSessionFactory sf) throws Exception {
              doTestK(sf);
           }
        });
     }
  
     @Test
     public void testL() throws Exception {
        runTest(new RunnableT() {
           @Override
           public void run(final ClientSessionFactory sf) throws Exception {
              doTestL(sf);
           }
        });
     }
  
     @Test
     public void testN() throws Exception {
        runTest(new RunnableT() {
           @Override
           public void run(final ClientSessionFactory sf) throws Exception {
              doTestN(sf);
           }
        });
     }
  
     public void runTest(final RunnableT runnable) throws Exception {
        int numIts = getNumIterations();
  
        for (int its = 0; its < numIts; its++) {
           logger.debug("####{} iteration #{}", getName(), its);
           start();
           ServerLocator locator = createInVMNonHALocator().setReconnectAttempts(15).setConfirmationWindowSize(1024 * 1024);
  
           ClientSessionFactory sf = createSessionFactory(locator);
  
           ClientSession session = sf.createSession(false, false, false);
  
           Failer failer = startFailer(250, session);
  
           try {
              do {
                 runnable.run(sf);
              }
              while (!failer.isExecuted());
           } catch (Exception e) {
              if (expectedException(e)) {
                 logger.info("an expected exception was thrown - {}", e.getMessage(), e);
                 // one more time
                 numIts++;
              } else {
                 throw e;
              }
           } finally {
              try {
                 sf.close();
                 session.close();
              } catch (Throwable ignored) {
              }
              try {
                 locator.close();
              } catch (Throwable ignored) {
              }
  
              try {
                 server.stop();
              } catch (Throwable ignored) {
              } finally {
                 server = null;
              }
           }
        }
     }
  
     private boolean expectedException(Throwable e) {
        // during a reattachment it's not possible to know if the previous ack was sent or not.
        // this is expected and clients should deal with this situation
        return e instanceof ActiveMQIllegalStateException && e.getMessage().contains("AMQ229027");
     }
  
  
  
     protected void doTestA(final ClientSessionFactory sf) throws Exception {
        long start = System.currentTimeMillis();
  
        ClientSession s = sf.createSession(false, false, false);
  
        final int numMessages = 100;
  
        final int numSessions = 10;
  
        Set<ClientConsumer> consumers = new HashSet<>();
        Set<ClientSession> sessions = new HashSet<>();
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           ClientSession sessConsume = sf.createSession(false, true, true);
  
           sessConsume.start();
  
           sessConsume.createQueue(QueueConfiguration.of(subName).setAddress(ADDRESS).setDurable(false));
  
           ClientConsumer consumer = sessConsume.createConsumer(subName);
  
           consumers.add(consumer);
  
           sessions.add(sessConsume);
        }
  
        ClientSession sessSend = sf.createSession(false, true, true);
  
        ClientProducer producer = sessSend.createProducer(ADDRESS);
  
        for (int i = 0; i < numMessages; i++) {
           ClientMessage message = sessSend.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
           message.putIntProperty(SimpleString.of("count"), i);
           producer.send(message);
        }
  
        class MyHandler extends AssertionCheckMessageHandler {
  
           final CountDownLatch latch = new CountDownLatch(1);
  
           int count;
  
           @Override
           public void onMessageAssert(final ClientMessage message) {
              if (count == numMessages) {
                 fail("Too many messages");
              }
  
              assertEquals(count, message.getObjectProperty(SimpleString.of("count")));
  
              count++;
  
              try {
                 message.acknowledge();
              } catch (ActiveMQException me) {
                 logger.error("Failed to process", me);
              }
  
              if (count == numMessages) {
                 latch.countDown();
              }
           }
        }
  
        Set<MyHandler> handlers = new HashSet<>();
  
        for (ClientConsumer consumer : consumers) {
           MyHandler handler = new MyHandler();
  
           consumer.setMessageHandler(handler);
  
           handlers.add(handler);
        }
  
        for (MyHandler handler : handlers) {
           boolean ok = handler.latch.await(5000, TimeUnit.MILLISECONDS);
  
           handler.checkAssertions();
  
           assertTrue(ok, "Didn't receive all messages");
        }
  
        sessSend.close();
        for (ClientSession session : sessions) {
           session.close();
        }
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           s.deleteQueue(subName);
        }
  
        s.close();
  
        long end = System.currentTimeMillis();
  
        logger.debug("duration {}", (end - start));
     }
  
     protected void doTestB(final ClientSessionFactory sf) throws Exception {
        long start = System.currentTimeMillis();
  
        ClientSession s = sf.createSession(false, false, false);
  
        final int numMessages = 100;
  
        final int numSessions = 50;
  
        Set<ClientConsumer> consumers = new HashSet<>();
        Set<ClientSession> sessions = new HashSet<>();
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           ClientSession sessConsume = sf.createSession(false, true, true);
  
           sessConsume.createQueue(QueueConfiguration.of(subName).setAddress(ADDRESS).setDurable(false));
  
           ClientConsumer consumer = sessConsume.createConsumer(subName);
  
           consumers.add(consumer);
  
           sessions.add(sessConsume);
        }
  
        ClientSession sessSend = sf.createSession(false, true, true);
  
        ClientProducer producer = sessSend.createProducer(ADDRESS);
  
        for (int i = 0; i < numMessages; i++) {
           ClientMessage message = sessSend.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
           message.putIntProperty(SimpleString.of("count"), i);
           producer.send(message);
        }
  
        for (ClientSession session : sessions) {
           session.start();
        }
  
        class MyHandler extends AssertionCheckMessageHandler {
  
           final CountDownLatch latch = new CountDownLatch(1);
  
           int count;
  
           @Override
           public void onMessageAssert(final ClientMessage message) {
              if (count == numMessages) {
                 fail("Too many messages");
              }
  
              assertEquals(count, message.getObjectProperty(SimpleString.of("count")));
  
              count++;
  
              if (count == numMessages) {
                 latch.countDown();
              }
           }
        }
  
        Set<MyHandler> handlers = new HashSet<>();
  
        for (ClientConsumer consumer : consumers) {
           MyHandler handler = new MyHandler();
  
           consumer.setMessageHandler(handler);
  
           handlers.add(handler);
        }
  
        for (MyHandler handler : handlers) {
           boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS);
  
           handler.checkAssertions();
  
           assertTrue(ok);
        }
  
        sessSend.close();
  
        for (ClientSession session : sessions) {
           session.close();
        }
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           s.deleteQueue(subName);
        }
  
        s.close();
  
        long end = System.currentTimeMillis();
  
        logger.debug("duration {}", (end - start));
  
     }
  
     protected void doTestC(final ClientSessionFactory sf) throws Exception {
        long start = System.currentTimeMillis();
  
        ClientSession s = sf.createSession(false, false, false);
  
        final int numMessages = 100;
  
        final int numSessions = 1;
  
        Set<ClientConsumer> consumers = new HashSet<>();
        Set<ClientSession> sessions = new HashSet<>();
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           ClientSession sessConsume = sf.createSession(false, false, false);
  
           sessConsume.start();
  
           sessConsume.createQueue(QueueConfiguration.of(subName).setAddress(ADDRESS).setDurable(false));
  
           ClientConsumer consumer = sessConsume.createConsumer(subName);
  
           consumers.add(consumer);
  
           sessions.add(sessConsume);
        }
  
        ClientSession sessSend = sf.createSession(false, false, true);
  
        ClientProducer producer = sessSend.createProducer(ADDRESS);
  
        for (int i = 0; i < numMessages; i++) {
           ClientMessage message = sessSend.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
           message.putIntProperty(SimpleString.of("count"), i);
           producer.send(message);
        }
  
        sessSend.rollback();
  
        for (int i = 0; i < numMessages; i++) {
           ClientMessage message = sessSend.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
           message.putIntProperty(SimpleString.of("count"), i);
           producer.send(message);
        }
  
        sessSend.commit();
  
        class MyHandler extends AssertionCheckMessageHandler {
  
           final CountDownLatch latch = new CountDownLatch(1);
  
           int count;
  
           @Override
           public void onMessageAssert(final ClientMessage message) {
              if (count == numMessages) {
                 fail("Too many messages, expected " + count);
              }
  
              assertEquals(count, message.getObjectProperty(SimpleString.of("count")));
  
              count++;
  
              try {
                 message.acknowledge();
              } catch (ActiveMQException e) {
                 e.printStackTrace();
                 throw new RuntimeException(e.getMessage(), e);
              }
  
              if (count == numMessages) {
                 latch.countDown();
              }
           }
        }
  
        Set<MyHandler> handlers = new HashSet<>();
  
        for (ClientConsumer consumer : consumers) {
           MyHandler handler = new MyHandler();
  
           consumer.setMessageHandler(handler);
  
           handlers.add(handler);
        }
  
        for (MyHandler handler : handlers) {
           boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS);
  
           assertTrue(ok);
  
           handler.checkAssertions();
        }
  
        handlers.clear();
  
        // New handlers
        for (ClientConsumer consumer : consumers) {
           MyHandler handler = new MyHandler();
  
           consumer.setMessageHandler(handler);
  
           handlers.add(handler);
        }
  
        for (ClientSession session : sessions) {
           session.rollback();
        }
  
        for (MyHandler handler : handlers) {
           boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS);
  
           assertTrue(ok);
  
           handler.checkAssertions();
        }
  
        for (ClientSession session : sessions) {
           session.commit();
        }
  
        sessSend.close();
        for (ClientSession session : sessions) {
           session.close();
        }
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           s.deleteQueue(subName);
        }
  
        s.close();
  
        long end = System.currentTimeMillis();
  
        logger.debug("duration {}", (end - start));
     }
  
     protected void doTestD(final ClientSessionFactory sf) throws Exception {
        long start = System.currentTimeMillis();
  
        ClientSession s = sf.createSession(false, false, false);
  
        final int numMessages = 100;
  
        final int numSessions = 10;
  
        Set<ClientConsumer> consumers = new HashSet<>();
        Set<ClientSession> sessions = new HashSet<>();
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           ClientSession sessConsume = sf.createSession(false, false, false);
  
           sessConsume.createQueue(QueueConfiguration.of(subName).setAddress(ADDRESS).setDurable(false));
  
           ClientConsumer consumer = sessConsume.createConsumer(subName);
  
           consumers.add(consumer);
  
           sessions.add(sessConsume);
        }
  
        ClientSession sessSend = sf.createSession(false, false, true);
  
        ClientProducer producer = sessSend.createProducer(ADDRESS);
  
        for (int i = 0; i < numMessages; i++) {
           ClientMessage message = sessSend.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
           message.putIntProperty(SimpleString.of("count"), i);
           producer.send(message);
        }
  
        sessSend.rollback();
  
        for (int i = 0; i < numMessages; i++) {
           ClientMessage message = sessSend.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
           message.putIntProperty(SimpleString.of("count"), i);
           producer.send(message);
        }
  
        sessSend.commit();
  
        for (ClientSession session : sessions) {
           session.start();
        }
  
        class MyHandler extends AssertionCheckMessageHandler {
  
           final CountDownLatch latch = new CountDownLatch(1);
  
           int count;
  
           @Override
           public void onMessageAssert(final ClientMessage message) {
              if (count == numMessages) {
                 fail("Too many messages, " + count);
              }
  
              assertEquals(count, message.getObjectProperty(SimpleString.of("count")));
  
              count++;
  
              if (count == numMessages) {
                 latch.countDown();
              }
           }
        }
  
        Set<MyHandler> handlers = new HashSet<>();
  
        for (ClientConsumer consumer : consumers) {
           MyHandler handler = new MyHandler();
  
           consumer.setMessageHandler(handler);
  
           handlers.add(handler);
        }
  
        for (MyHandler handler : handlers) {
           boolean ok = handler.latch.await(20000, TimeUnit.MILLISECONDS);
  
           assertTrue(ok);
  
           handler.checkAssertions();
        }
  
        handlers.clear();
  
        // New handlers
        for (ClientConsumer consumer : consumers) {
           MyHandler handler = new MyHandler();
  
           consumer.setMessageHandler(handler);
  
           handlers.add(handler);
        }
  
        for (ClientSession session : sessions) {
           session.rollback();
        }
  
        for (MyHandler handler : handlers) {
           boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS);
  
           assertTrue(ok);
  
           handler.checkAssertions();
        }
  
        for (ClientSession session : sessions) {
           session.commit();
        }
  
        sessSend.close();
        for (ClientSession session : sessions) {
           session.close();
        }
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           s.deleteQueue(subName);
        }
  
        s.close();
  
        long end = System.currentTimeMillis();
  
        logger.debug("duration {}", (end - start));
     }
  
     // Now with synchronous receive()
  
     protected void doTestE(final ClientSessionFactory sf) throws Exception {
        long start = System.currentTimeMillis();
  
        ClientSession s = sf.createSession(false, false, false);
  
        final int numMessages = 100;
  
        final int numSessions = 10;
  
        Set<ClientConsumer> consumers = new HashSet<>();
        Set<ClientSession> sessions = new HashSet<>();
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           ClientSession sessConsume = sf.createSession(false, true, true);
  
           sessConsume.start();
  
           sessConsume.createQueue(QueueConfiguration.of(subName).setAddress(ADDRESS).setDurable(false));
  
           ClientConsumer consumer = sessConsume.createConsumer(subName);
  
           consumers.add(consumer);
  
           sessions.add(sessConsume);
        }
  
        ClientSession sessSend = sf.createSession(false, true, true);
  
        ClientProducer producer = sessSend.createProducer(ADDRESS);
  
        for (int i = 0; i < numMessages; i++) {
           ClientMessage message = sessSend.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
           message.putIntProperty(SimpleString.of("count"), i);
           producer.send(message);
        }
  
        for (int i = 0; i < numMessages; i++) {
           for (ClientConsumer consumer : consumers) {
              ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT);
  
              assertNotNull(msg);
  
              assertEquals(i, msg.getObjectProperty(SimpleString.of("count")));
  
              msg.acknowledge();
           }
        }
  
        for (int i = 0; i < numMessages; i++) {
           for (ClientConsumer consumer : consumers) {
              ClientMessage msg = consumer.receiveImmediate();
  
              assertNull(msg);
           }
        }
  
        sessSend.close();
        for (ClientSession session : sessions) {
           session.close();
        }
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           s.deleteQueue(subName);
        }
  
        s.close();
  
        long end = System.currentTimeMillis();
  
        logger.debug("duration {}", (end - start));
     }
  
     protected void doTestF(final ClientSessionFactory sf) throws Exception {
        long start = System.currentTimeMillis();
  
        ClientSession s = sf.createSession(false, false, false);
  
        final int numMessages = 100;
  
        final int numSessions = 10;
  
        Set<ClientConsumer> consumers = new HashSet<>();
        Set<ClientSession> sessions = new HashSet<>();
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           ClientSession sessConsume = sf.createSession(false, true, true);
  
           sessConsume.createQueue(QueueConfiguration.of(subName).setAddress(ADDRESS).setDurable(false));
  
           ClientConsumer consumer = sessConsume.createConsumer(subName);
  
           consumers.add(consumer);
  
           sessions.add(sessConsume);
        }
  
        ClientSession sessSend = sf.createSession(false, true, true);
  
        ClientProducer producer = sessSend.createProducer(ADDRESS);
  
        for (int i = 0; i < numMessages; i++) {
           ClientMessage message = sessSend.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
           message.putIntProperty(SimpleString.of("count"), i);
           producer.send(message);
        }
  
        for (ClientSession session : sessions) {
           session.start();
        }
  
        for (int i = 0; i < numMessages; i++) {
           for (ClientConsumer consumer : consumers) {
              ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT);
  
              if (msg == null) {
                 throw new IllegalStateException("Failed to receive message " + i);
              }
  
              assertNotNull(msg);
  
              assertEquals(i, msg.getObjectProperty(SimpleString.of("count")));
  
              msg.acknowledge();
           }
        }
  
        for (int i = 0; i < numMessages; i++) {
           for (ClientConsumer consumer : consumers) {
              ClientMessage msg = consumer.receiveImmediate();
  
              assertNull(msg);
           }
        }
  
        sessSend.close();
        for (ClientSession session : sessions) {
           session.close();
        }
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           s.deleteQueue(subName);
        }
  
        s.close();
  
        assertEquals(1, ((ClientSessionFactoryImpl) sf).numSessions());
  
        long end = System.currentTimeMillis();
  
        logger.debug("duration {}", (end - start));
     }
  
     protected void doTestG(final ClientSessionFactory sf) throws Exception {
        long start = System.currentTimeMillis();
  
        ClientSession s = sf.createSession(false, false, false);
  
        final int numMessages = 100;
  
        final int numSessions = 10;
  
        Set<ClientConsumer> consumers = new HashSet<>();
        Set<ClientSession> sessions = new HashSet<>();
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           ClientSession sessConsume = sf.createSession(false, false, false);
  
           sessConsume.start();
  
           sessConsume.createQueue(QueueConfiguration.of(subName).setAddress(ADDRESS).setDurable(false));
  
           ClientConsumer consumer = sessConsume.createConsumer(subName);
  
           consumers.add(consumer);
  
           sessions.add(sessConsume);
        }
  
        ClientSession sessSend = sf.createSession(false, false, false);
  
        ClientProducer producer = sessSend.createProducer(ADDRESS);
  
        for (int i = 0; i < numMessages; i++) {
           ClientMessage message = sessSend.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
           message.putIntProperty(SimpleString.of("count"), i);
           producer.send(message);
        }
  
        sessSend.rollback();
  
        for (int i = 0; i < numMessages; i++) {
           ClientMessage message = sessSend.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
           message.putIntProperty(SimpleString.of("count"), i);
           producer.send(message);
        }
  
        sessSend.commit();
  
        for (int i = 0; i < numMessages; i++) {
           for (ClientConsumer consumer : consumers) {
              ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT);
  
              assertNotNull(msg);
  
              assertEquals(i, msg.getObjectProperty(SimpleString.of("count")));
  
              msg.acknowledge();
           }
        }
  
        for (ClientConsumer consumer : consumers) {
           ClientMessage msg = consumer.receiveImmediate();
  
           assertNull(msg);
        }
  
        for (ClientSession session : sessions) {
           session.rollback();
        }
  
        for (int i = 0; i < numMessages; i++) {
           for (ClientConsumer consumer : consumers) {
              ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT);
  
              assertNotNull(msg);
  
              assertEquals(i, msg.getObjectProperty(SimpleString.of("count")));
  
              msg.acknowledge();
           }
        }
  
        for (int i = 0; i < numMessages; i++) {
           for (ClientConsumer consumer : consumers) {
              ClientMessage msg = consumer.receiveImmediate();
  
              assertNull(msg);
           }
        }
  
        for (ClientSession session : sessions) {
           session.commit();
        }
  
        sessSend.close();
        for (ClientSession session : sessions) {
           session.close();
        }
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           s.deleteQueue(subName);
        }
  
        s.close();
  
        long end = System.currentTimeMillis();
  
        logger.debug("duration {}", (end - start));
     }
  
     protected void doTestH(final ClientSessionFactory sf) throws Exception {
        long start = System.currentTimeMillis();
  
        ClientSession s = sf.createSession(false, false, false);
  
        final int numMessages = 100;
  
        final int numSessions = 10;
  
        Set<ClientConsumer> consumers = new HashSet<>();
        Set<ClientSession> sessions = new HashSet<>();
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           ClientSession sessConsume = sf.createSession(false, false, false);
  
           sessConsume.createQueue(QueueConfiguration.of(subName).setAddress(ADDRESS).setDurable(false));
  
           ClientConsumer consumer = sessConsume.createConsumer(subName);
  
           consumers.add(consumer);
  
           sessions.add(sessConsume);
        }
  
        ClientSession sessSend = sf.createSession(false, false, false);
  
        ClientProducer producer = sessSend.createProducer(ADDRESS);
  
        for (int i = 0; i < numMessages; i++) {
           ClientMessage message = sessSend.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
           message.putIntProperty(SimpleString.of("count"), i);
           producer.send(message);
        }
  
        sessSend.rollback();
  
        for (int i = 0; i < numMessages; i++) {
           ClientMessage message = sessSend.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
           message.putIntProperty(SimpleString.of("count"), i);
           producer.send(message);
        }
  
        sessSend.commit();
  
        for (ClientSession session : sessions) {
           session.start();
        }
  
        for (int i = 0; i < numMessages; i++) {
           for (ClientConsumer consumer : consumers) {
              ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT);
  
              assertNotNull(msg);
  
              assertEquals(i, msg.getObjectProperty(SimpleString.of("count")));
  
              msg.acknowledge();
           }
        }
  
        for (int i = 0; i < numMessages; i++) {
           for (ClientConsumer consumer : consumers) {
              ClientMessage msg = consumer.receiveImmediate();
  
              assertNull(msg);
           }
        }
  
        for (ClientSession session : sessions) {
           session.rollback();
        }
  
        for (int i = 0; i < numMessages; i++) {
           for (ClientConsumer consumer : consumers) {
              ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT);
  
              assertNotNull(msg);
  
              assertEquals(i, msg.getObjectProperty(SimpleString.of("count")));
  
              msg.acknowledge();
           }
        }
  
        for (int i = 0; i < numMessages; i++) {
           for (ClientConsumer consumer : consumers) {
              ClientMessage msg = consumer.receiveImmediate();
  
              assertNull(msg);
           }
        }
  
        for (ClientSession session : sessions) {
           session.commit();
        }
  
        sessSend.close();
        for (ClientSession session : sessions) {
           session.close();
        }
  
        for (int i = 0; i < numSessions; i++) {
           SimpleString subName = SimpleString.of("sub" + i);
  
           s.deleteQueue(subName);
        }
  
        s.close();
  
        long end = System.currentTimeMillis();
  
        logger.debug("duration {}", (end - start));
     }
  
     protected void doTestI(final ClientSessionFactory sf) throws Exception {
        ClientSession sessCreate = sf.createSession(false, true, true);
  
        sessCreate.createQueue(QueueConfiguration.of(ADDRESS).setDurable(false));
  
        ClientSession sess = sf.createSession(false, true, true);
  
        sess.start();
  
        ClientConsumer consumer = sess.createConsumer(ADDRESS);
  
        ClientProducer producer = sess.createProducer(ADDRESS);
  
        ClientMessage message = sess.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
        producer.send(message);
  
        ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
  
        assertNotNull(message2);
  
        message2.acknowledge();
  
        sess.close();
  
        sessCreate.deleteQueue(ADDRESS);
  
        sessCreate.close();
     }
  
     protected void doTestJ(final ClientSessionFactory sf) throws Exception {
        ClientSession sessCreate = sf.createSession(false, true, true);
  
        sessCreate.createQueue(QueueConfiguration.of(ADDRESS).setDurable(false));
  
        ClientSession sess = sf.createSession(false, true, true);
  
        sess.start();
  
        ClientConsumer consumer = sess.createConsumer(ADDRESS);
  
        ClientProducer producer = sess.createProducer(ADDRESS);
  
        ClientMessage message = sess.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
        producer.send(message);
  
        ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
  
        assertNotNull(message2);
  
        message2.acknowledge();
  
        sess.close();
  
        sessCreate.deleteQueue(ADDRESS);
  
        sessCreate.close();
     }
  
     protected void doTestK(final ClientSessionFactory sf) throws Exception {
        ClientSession s = sf.createSession(false, false, false);
  
        s.createQueue(QueueConfiguration.of(ADDRESS).setDurable(false));
  
        final int numConsumers = 100;
  
        for (int i = 0; i < numConsumers; i++) {
           ClientConsumer consumer = s.createConsumer(ADDRESS);
  
           consumer.close();
        }
  
        s.deleteQueue(ADDRESS);
  
        s.close();
     }
  
     protected void doTestL(final ClientSessionFactory sf) throws Exception {
        final int numSessions = 10;
  
        for (int i = 0; i < numSessions; i++) {
           ClientSession session = sf.createSession(false, false, false);
  
           session.close();
        }
     }
  
     protected void doTestN(final ClientSessionFactory sf) throws Exception {
        ClientSession sessCreate = sf.createSession(false, true, true);
  
        sessCreate.createQueue(QueueConfiguration.of(ADDRESS).setDurable(false));
  
        ClientSession sess = sf.createSession(false, true, true);
  
        sess.stop();
  
        sess.start();
  
        sess.stop();
  
        ClientConsumer consumer = sess.createConsumer(ADDRESS);
  
        ClientProducer producer = sess.createProducer(ADDRESS);
  
        ClientMessage message = sess.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);
        producer.send(message);
  
        sess.start();
  
        ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
  
        assertNotNull(message2);
  
        message2.acknowledge();
  
        sess.stop();
  
        sess.start();
  
        sess.close();
  
        sessCreate.deleteQueue(SimpleString.of(ADDRESS.toString()));
  
        sessCreate.close();
     }
  
     @Override
     @BeforeEach
     public void setUp() throws Exception {
        super.setUp();
  
        timer = new Timer(true);
     }
  
     @Override
     @AfterEach
     public void tearDown() throws Exception {
        timer.cancel();
  
        super.tearDown();
     }
  
  
     private Failer startFailer(final long time, final ClientSession session) {
        Failer failer = new Failer((ClientSessionInternal) session);
  
        timer.schedule(failer, (long) (time * Math.random()), 100);
  
        return failer;
     }
  
     private void start() throws Exception {
        server = addServer(ActiveMQServers.newActiveMQServer(createDefaultInVMConfig(), false));
        server.start();
     }
  
     private void stop() throws Exception {
        server.stop();
        Wait.assertFalse(server::isStarted, 5_000, 100);
  
        Wait.assertEquals(0, () -> InVMRegistry.instance.size(), 5_000, 100);
  
        server = null;
     }
  
  
     class Failer extends TimerTask {
  
        private final ClientSessionInternal session;
  
        private boolean executed;
  
        Failer(final ClientSessionInternal session) {
           this.session = session;
        }
  
        @Override
        public synchronized void run() {
           logger.debug("** Failing connection");
  
           session.getConnection().fail(new ActiveMQNotConnectedException("oops"));
  
           logger.debug("** Fail complete");
  
           cancel();
  
           executed = true;
        }
  
        public synchronized boolean isExecuted() {
           return executed;
        }
     }
  
     public abstract class RunnableT {
  
        abstract void run(ClientSessionFactory sf) throws Exception;
     }
  
     abstract static class AssertionCheckMessageHandler implements MessageHandler {
  
        public void checkAssertions() {
           for (AssertionError e : errors) {
              // it will throw the first error
              throw e;
           }
        }
  
        private final List<AssertionError> errors = new ArrayList<>();
  
        @Override
        public void onMessage(ClientMessage message) {
           try {
              onMessageAssert(message);
           } catch (AssertionError e) {
              e.printStackTrace(); // System.out -> junit reports
              errors.add(e);
           }
        }
  
        public abstract void onMessageAssert(ClientMessage message);
     }
  }
  