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.openwire;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  import javax.jms.Connection;
  import javax.jms.ConnectionFactory;
  import javax.jms.JMSException;
  import javax.jms.MessageConsumer;
  import javax.jms.MessageProducer;
  import javax.jms.Queue;
  import javax.jms.Session;
  import javax.jms.TextMessage;
  
  import java.io.PrintStream;
  import java.lang.invoke.MethodHandles;
  import java.util.ArrayList;
  import java.util.ConcurrentModificationException;
  import java.util.Map;
  import java.util.concurrent.CountDownLatch;
  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.ActiveMQConnection;
  import org.apache.activemq.ActiveMQConnectionFactory;
  import org.apache.activemq.ActiveMQMessageConsumer;
  import org.apache.activemq.ActiveMQPrefetchPolicy;
  import org.apache.activemq.RedeliveryPolicy;
  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.Configuration;
  import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
  import org.apache.activemq.artemis.utils.Wait;
  import org.apache.activemq.command.ActiveMQMessage;
  import org.apache.activemq.command.ActiveMQQueue;
  import org.apache.activemq.command.LocalTransactionId;
  import org.apache.activemq.command.MessageAck;
  import org.apache.activemq.command.TransactionId;
  import org.apache.activemq.command.TransactionInfo;
  import org.apache.activemq.transport.Transport;
  import org.apache.activemq.transport.failover.FailoverTransport;
  import org.apache.activemq.transport.tcp.TcpTransport;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.Timeout;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  public class PrefetchRedeliveryCountOpenwireTest extends OpenWireTestBase {
  
     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  
     @BeforeEach
     @Override
     public void setUp() throws Exception {
        realStore = true;
        super.setUp();
     }
  
     @Override
     protected void configureAddressSettings(Map<String, AddressSettings> addressSettingsMap) {
        super.configureAddressSettings(addressSettingsMap);
        // force send to dlq early
        addressSettingsMap.put("exampleQueue", new AddressSettings().setAutoCreateQueues(false).setAutoCreateAddresses(false).setDeadLetterAddress(SimpleString.of("ActiveMQ.DLQ")).setAutoCreateAddresses(true).setMaxDeliveryAttempts(2));
        // force send to dlq late
        addressSettingsMap.put("exampleQueueTwo", new AddressSettings().setAutoCreateQueues(false).setAutoCreateAddresses(false).setDeadLetterAddress(SimpleString.of("ActiveMQ.DLQ")).setAutoCreateAddresses(true).setMaxDeliveryAttempts(-1));
     }
  
     @Override
     protected void extraServerConfig(Configuration serverConfig) {
        // useful to debug if there is some contention that causes a concurrent rollback,
        // when true, the rollback journal update uses the operation context and the failure propagates
        //serverConfig.setJournalSyncTransactional(true);
     }
  
     @Test
     @Timeout(60)
     public void testConsumerSingleMessageLoopExclusive() throws Exception {
        doTestConsumerSingleMessageLoop(true);
     }
  
     @Test
     @Timeout(60)
     public void testConsumerSingleMessageLoopNonExclusive() throws Exception {
        doTestConsumerSingleMessageLoop(false);
     }
  
     public void  doTestConsumerSingleMessageLoop(boolean exclusive) throws Exception {
        Connection exConn = null;
  
        SimpleString durableQueue = SimpleString.of("exampleQueue");
        this.server.createQueue(QueueConfiguration.of(durableQueue).setRoutingType(RoutingType.ANYCAST).setExclusive(exclusive));
  
        try {
           ActiveMQConnectionFactory exFact = new ActiveMQConnectionFactory();
           exFact.setWatchTopicAdvisories(false);
  
           Queue queue = new ActiveMQQueue("exampleQueue");
  
           exConn = exFact.createConnection();
  
           exConn.start();
  
           Session session = exConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
  
           MessageProducer producer = session.createProducer(queue);
  
           TextMessage message = session.createTextMessage("This is a text message");
  
           int numMessages = 20;
           for (int i = 0; i < numMessages; i++) {
              producer.send(message);
           }
  
           for (int i = 0; i < numMessages; i++) {
              // consumer per message
              MessageConsumer messageConsumer = session.createConsumer(queue);
  
              TextMessage messageReceived = (TextMessage) messageConsumer.receive(5000);
              assertNotNull(messageReceived);
  
              assertEquals("This is a text message", messageReceived.getText());
              messageConsumer.close();
           }
        } finally {
           if (exConn != null) {
              exConn.close();
           }
        }
     }
  
     @Test
     @Timeout(60)
     public void testExclusiveConsumerOrderOnReconnectionLargePrefetch() throws Exception {
        Connection exConn = null;
  
        SimpleString durableQueue = SimpleString.of("exampleQueueTwo");
        this.server.createQueue(QueueConfiguration.of(durableQueue).setRoutingType(RoutingType.ANYCAST).setExclusive(true));
  
        try {
           ActiveMQConnectionFactory exFact = new ActiveMQConnectionFactory();
           exFact.setWatchTopicAdvisories(false);
  
           ActiveMQPrefetchPolicy prefetchPastMaxDeliveriesInLoop = new ActiveMQPrefetchPolicy();
           prefetchPastMaxDeliveriesInLoop.setAll(2000);
           exFact.setPrefetchPolicy(prefetchPastMaxDeliveriesInLoop);
  
           RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
           redeliveryPolicy.setMaximumRedeliveries(4000);
           exFact.setRedeliveryPolicy(redeliveryPolicy);
  
           Queue queue = new ActiveMQQueue("exampleQueueTwo");
  
           exConn = exFact.createConnection();
  
           exConn.start();
  
           Session session = exConn.createSession(true, Session.AUTO_ACKNOWLEDGE);
  
           MessageProducer producer = session.createProducer(queue);
  
           TextMessage message = session.createTextMessage("This is a text message");
  
           int numMessages = 10000;
           for (int i = 0; i < numMessages; i++) {
              message.setIntProperty("SEQ", i);
              producer.send(message);
           }
           session.commit();
           exConn.close();
  
           final int batch = 200;
           for (int i = 0; i < numMessages; i += batch) {
              // connection per batch
              exConn = exFact.createConnection();
              exConn.start();
  
              session = exConn.createSession(true, Session.SESSION_TRANSACTED);
  
              MessageConsumer messageConsumer = session.createConsumer(queue);
              TextMessage messageReceived;
              for (int j = 0; j < batch; j++) { // a small batch
                 messageReceived = (TextMessage) messageConsumer.receive(5000);
                 assertNotNull(messageReceived, "null @ i=" + i);
                 assertEquals(i + j, messageReceived.getIntProperty("SEQ"));
  
                 assertEquals("This is a text message", messageReceived.getText());
              }
              session.commit();
  
              // force a local socket close such that the broker sees an exception on the connection and fails the consumer via close
              ((org.apache.activemq.ActiveMQConnection)exConn).getTransport().narrow(FailoverTransport.class).stop();
              exConn.close();
           }
        } finally {
           if (exConn != null) {
              exConn.close();
           }
        }
     }
  
     @Test
     @Timeout(60)
     public void testServerSideRollbackOnCloseOrder() throws Exception {
  
        final ArrayList<Throwable> errors = new ArrayList<>();
        SimpleString durableQueue = SimpleString.of("exampleQueueTwo");
        this.server.createQueue(QueueConfiguration.of(durableQueue).setRoutingType(RoutingType.ANYCAST).setExclusive(true));
  
        Queue queue = new ActiveMQQueue(durableQueue.toString());
  
        final ActiveMQConnectionFactory exFact = new ActiveMQConnectionFactory("failover:(tcp://localhost:61616?closeAsync=false)?startupMaxReconnectAttempts=10&maxReconnectAttempts=0&timeout=1000");
        exFact.setWatchTopicAdvisories(false);
        exFact.setConnectResponseTimeout(10000);
  
        ActiveMQPrefetchPolicy prefetchPastMaxDeliveriesInLoop = new ActiveMQPrefetchPolicy();
        prefetchPastMaxDeliveriesInLoop.setAll(2000);
        exFact.setPrefetchPolicy(prefetchPastMaxDeliveriesInLoop);
  
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setRedeliveryDelay(0);
        redeliveryPolicy.setMaximumRedeliveries(-1);
        exFact.setRedeliveryPolicy(redeliveryPolicy);
  
        Connection exConn = exFact.createConnection();
        exConn.start();
  
        Session session = exConn.createSession(true, Session.AUTO_ACKNOWLEDGE);
        MessageProducer producer = session.createProducer(queue);
        TextMessage message = session.createTextMessage("This is a text message");
  
        int numMessages = 1000;
  
        for (int i = 0; i < numMessages; i++) {
           message.setIntProperty("SEQ", i);
           producer.send(message);
        }
        session.commit();
        exConn.close();
  
        final int numConsumers = 2;
        ExecutorService executorService = Executors.newCachedThreadPool();
        // consume under load
        final int numLoadProducers = 2;
        underLoad(numLoadProducers, ()-> {
  
           // a bunch of concurrent batch consumers, expecting order
           AtomicBoolean done = new AtomicBoolean(false);
           AtomicInteger receivedCount = new AtomicInteger();
           AtomicInteger inProgressBatch = new AtomicInteger();
  
           final int batch = 100;
  
           final  ExecutorService commitExecutor = Executors.newCachedThreadPool();
  
           Runnable consumerTask = () -> {
  
              Connection toCloseOnError = null;
              while (!done.get() && receivedCount.get() < 20 * numMessages) {
                 try (Connection consumerConnection = exFact.createConnection()) {
  
                    toCloseOnError = consumerConnection;
                    ((ActiveMQConnection) consumerConnection).setCloseTimeout(1); // so rollback on close won't block after socket close exception
  
                    consumerConnection.start();
  
                    Session consumerConnectionSession = consumerConnection.createSession(true, Session.SESSION_TRANSACTED);
                    MessageConsumer messageConsumer = consumerConnectionSession.createConsumer(queue);
                    TextMessage messageReceived = null;
  
                    int i = 0;
                    for (; i < batch; i++) {
                       messageReceived = (TextMessage) messageConsumer.receive(2000);
                       if (messageReceived == null) {
                          break;
                       }
  
                       receivedCount.incrementAndGet();
                       // need to infer batch from seq number and adjust - client never gets commit response
                       int receivedSeq = messageReceived.getIntProperty("SEQ");
                       int currentBatch = (receivedSeq / batch);
                       if (i == 0) {
                          if (inProgressBatch.get() != currentBatch) {
                             if (inProgressBatch.get() + 1 == currentBatch) {
                                inProgressBatch.incrementAndGet(); // all good, next batch
                                logger.info("@:" + receivedCount.get() + ", current batch increment to: " + inProgressBatch.get() + ", Received Seq: " + receivedSeq + ", Message: " + messageReceived);
                             } else {
                                // we have an order problem
                                done.set(true);
                                throw new AssertionError("@:" + receivedCount.get() + ", batch out of sequence, expected: " + (inProgressBatch.get() + 1) + ", but have: " + currentBatch + " @" + receivedSeq + ", Message: " + messageReceived);
                             }
                          }
                       }
                       // verify within batch order
                       assertEquals(((long) batch * inProgressBatch.get()) + i, receivedSeq, "@:" + receivedCount.get() + " batch out of order");
                    }
  
                    if (i != batch) {
                       continue;
                    }
  
                    // manual ack in tx to setup server for rollback work on fail
                    Transport transport = ((ActiveMQConnection) consumerConnection).getTransport();
                    TransactionId txId =  new LocalTransactionId(((ActiveMQConnection) consumerConnection).getConnectionInfo().getConnectionId(), receivedCount.get());
                    TransactionInfo tx = new TransactionInfo(((ActiveMQConnection) consumerConnection).getConnectionInfo().getConnectionId(), txId, TransactionInfo.BEGIN);
                    transport.request(tx);
                    MessageAck ack = new MessageAck();
                    ActiveMQMessage mqMessage = (ActiveMQMessage) messageReceived;
                    ack.setDestination(mqMessage.getDestination());
                    ack.setMessageID(mqMessage.getMessageId());
                    ack.setMessageCount(batch);
                    ack.setTransactionId(tx.getTransactionId());
                    ack.setConsumerId(((ActiveMQMessageConsumer)messageConsumer).getConsumerId());
  
                    transport.request(ack);
  
                    try {
                       // force a local socket close such that the broker sees an exception on the connection and fails the consumer via serverConsumer close
                       ((ActiveMQConnection) consumerConnection).getTransport().narrow(TcpTransport.class).stop();
                    } catch (Throwable expected) {
                    }
  
                 } catch (ConcurrentModificationException | NullPointerException ignored) {
                 } catch (JMSException ignored) {
                    // expected on executor stop
                 } catch (Throwable unexpected) {
                    unexpected.printStackTrace();
                    errors.add(unexpected);
                    done.set(true);
                 } finally {
                    if (toCloseOnError != null) {
                       try {
                          toCloseOnError.close();
                       } catch (Throwable ignored) {
                       }
                    }
                 }
              }
           };
  
           for (int i = 0; i < numConsumers; i++) {
              executorService.submit(consumerTask);
           }
           executorService.shutdown();
  
  
           try {
              assertTrue(executorService.awaitTermination(30, TimeUnit.SECONDS));
              assertTrue(errors.isEmpty());
           } catch (Throwable t) {
              errors.add(t);
           } finally {
              done.set(true);
              commitExecutor.shutdownNow();
              executorService.shutdownNow();
           }
  
           assertTrue(errors.isEmpty(), "errors: " + errors);
        });
  
        assertTrue(errors.isEmpty());
     }
  
     @Test
     @Timeout(60)
     public void testExclusiveConsumerBatchOrderUnderLoad() throws Exception {
  
        final ArrayList<Throwable> errors = new ArrayList<>();
        SimpleString durableQueue = SimpleString.of("exampleQueueTwo");
        this.server.createQueue(QueueConfiguration.of(durableQueue).setRoutingType(RoutingType.ANYCAST).setExclusive(true));
  
        Queue queue = new ActiveMQQueue(durableQueue.toString());
  
        final ActiveMQConnectionFactory exFact = new ActiveMQConnectionFactory("failover:(tcp://localhost:61616?closeAsync=false)?startupMaxReconnectAttempts=10&maxReconnectAttempts=0&timeout=1000");
        exFact.setWatchTopicAdvisories(false);
        exFact.setConnectResponseTimeout(10000);
  
        ActiveMQPrefetchPolicy prefetchPastMaxDeliveriesInLoop = new ActiveMQPrefetchPolicy();
        prefetchPastMaxDeliveriesInLoop.setAll(2000);
        exFact.setPrefetchPolicy(prefetchPastMaxDeliveriesInLoop);
  
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setRedeliveryDelay(0);
        redeliveryPolicy.setMaximumRedeliveries(-1);
        exFact.setRedeliveryPolicy(redeliveryPolicy);
  
        Connection exConn = exFact.createConnection();
        exConn.start();
  
        Session session = exConn.createSession(true, Session.AUTO_ACKNOWLEDGE);
        MessageProducer producer = session.createProducer(queue);
        TextMessage message = session.createTextMessage("This is a text message");
  
        int numMessages = 10000;
  
        for (int i = 0; i < numMessages; i++) {
           message.setIntProperty("SEQ", i);
           producer.send(message);
        }
        session.commit();
        exConn.close();
  
        final int numConsumers = 4;
        ExecutorService executorService = Executors.newCachedThreadPool();
        // consume under load
        final int numLoadProducers = 4;
        underLoad(numLoadProducers, ()-> {
  
           // a bunch of concurrent batch consumers, expecting order
           AtomicBoolean done = new AtomicBoolean(false);
           AtomicInteger receivedCount = new AtomicInteger();
           AtomicInteger inProgressBatch = new AtomicInteger();
  
           final int batch = 200;
  
           final  ExecutorService commitExecutor = Executors.newCachedThreadPool();
  
           Runnable consumerTask = () -> {
  
              Connection toCloseOnError = null;
              while (!done.get() && server.locateQueue(durableQueue).getMessageCount() > 0L) {
                 try (Connection consumerConnection = exFact.createConnection()) {
  
                    toCloseOnError = consumerConnection;
                    ((ActiveMQConnection) consumerConnection).setCloseTimeout(1); // so rollback on close won't block after socket close exception
  
                    consumerConnection.start();
  
                    Session consumerConnectionSession = consumerConnection.createSession(true, Session.SESSION_TRANSACTED);
                    MessageConsumer messageConsumer = consumerConnectionSession.createConsumer(queue);
                    TextMessage messageReceived;
  
                    int i = 0;
                    for (; i < batch; i++) {
                       messageReceived = (TextMessage) messageConsumer.receive(2000);
                       if (messageReceived == null) {
                          break;
                       }
  
                       receivedCount.incrementAndGet();
                       // need to infer batch from seq number and adjust - client never gets commit response
                       int receivedSeq = messageReceived.getIntProperty("SEQ");
                       int currentBatch = (receivedSeq / batch);
                       if (i == 0) {
                          if (inProgressBatch.get() != currentBatch) {
                             if (inProgressBatch.get() + 1 == currentBatch) {
                                inProgressBatch.incrementAndGet(); // all good, next batch
                                logger.info("@:" + receivedCount.get() + ", current batch increment to: " + inProgressBatch.get() + ", Received Seq: " + receivedSeq + ", Message: " + messageReceived);
                             } else {
                                // we have an order problem
                                done.set(true);
                                throw new AssertionError("@:" + receivedCount.get() + ", batch out of sequence, expected: " + (inProgressBatch.get() + 1) + ", but have: " + currentBatch + " @" + receivedSeq + ", Message: " + messageReceived);
                             }
                          }
                       }
                       // verify within batch order
                       assertEquals(((long) batch * inProgressBatch.get()) + i, receivedSeq, "@:" + receivedCount.get() + " batch out of order");
                    }
  
                    if (i != batch) {
                       continue;
                    }
  
                    // arrange concurrent commit - ack/commit of batch
                    // with server side error, potential for ack/commit and close-on-fail to contend
                    final CountDownLatch latch = new CountDownLatch(1);
                    final Session finalSession = consumerConnectionSession;
                    commitExecutor.submit(() -> {
                       try {
                          latch.countDown();
                          finalSession.commit();
  
                       } catch (Throwable expected) {
                       }
                    });
  
                    latch.await(1, TimeUnit.SECONDS);
  
                    // give a chance to have a batch complete to make progress!
                    TimeUnit.MILLISECONDS.sleep(15);
  
                    try {
                       // force a local socket close such that the broker sees an exception on the connection and fails the consumer via serverConsumer close
                       ((ActiveMQConnection) consumerConnection).getTransport().narrow(TcpTransport.class).stop();
                    } catch (Throwable expected) {
                    }
  
                 } catch (InterruptedException | ConcurrentModificationException | NullPointerException ignored) {
                 } catch (JMSException ignored) {
                    // expected on executor stop
                 } catch (Throwable unexpected) {
                    unexpected.printStackTrace();
                    errors.add(unexpected);
                    done.set(true);
                 } finally {
                    if (toCloseOnError != null) {
                       try {
                          toCloseOnError.close();
                       } catch (Throwable ignored) {
                       }
                    }
                 }
              }
           };
  
           for (int i = 0; i < numConsumers; i++) {
              executorService.submit(consumerTask);
           }
           executorService.shutdown();
  
           try {
              Wait.assertEquals(0L, () -> {
                 if (!errors.isEmpty()) {
                    return -1;
                 }
                 return server.locateQueue(durableQueue).getMessageCount();
              }, 30 * 1000);
           } catch (Throwable t) {
              errors.add(t);
           } finally {
              done.set(true);
              commitExecutor.shutdownNow();
              executorService.shutdownNow();
           }
  
           assertTrue(errors.isEmpty());
        });
  
        assertTrue(errors.isEmpty());
     }
  
     public void underLoad(final int numProducers, Runnable r) throws Exception {
        // produce some load with a producer(s)/consumer
        SimpleString durableQueue = SimpleString.of("exampleQueue");
        this.server.createQueue(QueueConfiguration.of(durableQueue).setRoutingType(RoutingType.ANYCAST));
  
        ExecutorService executor = Executors.newFixedThreadPool(numProducers + 1);
  
        Queue queue;
        ConnectionFactory cf;
        boolean useCoreForLoad = true;
  
        if (useCoreForLoad) {
           org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory connectionFactory = new org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory();
           connectionFactory.setConfirmationWindowSize(1000000);
           connectionFactory.setBlockOnDurableSend(true);
           connectionFactory.setBlockOnNonDurableSend(true);
           cf = connectionFactory;
  
           queue = connectionFactory.createContext().createQueue(durableQueue.toString());
  
        } else {
           ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:(tcp://localhost:61616)?startupMaxReconnectAttempts=0&maxReconnectAttempts=0&timeout=1000");
           connectionFactory.setWatchTopicAdvisories(false);
           connectionFactory.setCloseTimeout(1);
           connectionFactory.setSendTimeout(2000);
  
           cf = connectionFactory;
           queue = new ActiveMQQueue(durableQueue.toString());
        }
  
  
        final Queue destination = queue;
        final ConnectionFactory connectionFactory = cf;
        final AtomicBoolean done = new AtomicBoolean();
        Runnable producerTask = ()-> {
  
           try (Connection exConn = connectionFactory.createConnection()) {
  
              exConn.start();
  
              final Session session = exConn.createSession(true, Session.SESSION_TRANSACTED);
              final MessageProducer producer = session.createProducer(destination);
              final TextMessage message = session.createTextMessage("This is a text message");
  
              int count = 1;
              while (!done.get()) {
                 producer.send(message);
                 if ((count++ % 100) == 0) {
                    session.commit();
                 }
              }
           } catch (Exception ignored) {
           }
        };
  
        for (int i = 0; i < numProducers; i++) {
           executor.submit(producerTask);
        }
        // one consumer
        executor.submit(()-> {
  
           try (Connection exConn = connectionFactory.createConnection()) {
              exConn.start();
  
              Session session = exConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
              MessageConsumer messageConsumer = session.createConsumer(destination);
  
              while (!done.get()) {
                 messageConsumer.receive(200);
              }
           } catch (Exception ignored) {
           }
        });
  
  
        try {
           r.run();
        } finally {
           done.set(true);
           executor.shutdown();
           if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
              executor.shutdownNow();
           }
           logger.info("LOAD ADDED: " + server.locateQueue(durableQueue).getMessagesAdded());
        }
     }
  
     @Test
     @Timeout(120)
     public void testExclusiveConsumerTransactionalBatchOnReconnectionLargePrefetch() throws Exception {
        doTestExclusiveConsumerTransactionalBatchOnReconnectionLargePrefetch();
     }
  
     public void doTestExclusiveConsumerTransactionalBatchOnReconnectionLargePrefetch() throws Exception {
        Connection exConn = null;
  
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        SimpleString durableQueue = SimpleString.of("exampleQueueTwo");
        this.server.createQueue(QueueConfiguration.of(durableQueue).setRoutingType(RoutingType.ANYCAST).setExclusive(true));
        AtomicInteger batchConsumed = new AtomicInteger(0);
  
        try {
           ActiveMQConnectionFactory exFact = new ActiveMQConnectionFactory();
           exFact.setWatchTopicAdvisories(false);
  
           RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
           redeliveryPolicy.setMaximumRedeliveries(4000);
           exFact.setRedeliveryPolicy(redeliveryPolicy);
  
           Queue queue = new ActiveMQQueue("exampleQueueTwo");
  
           exConn = exFact.createConnection();
  
           exConn.start();
  
           Session session = exConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
  
           MessageProducer producer = session.createProducer(queue);
  
           TextMessage message = session.createTextMessage("This is a text message");
  
           int numMessages = 1000;
           for (int i = 0; i < numMessages; i++) {
              message.setIntProperty("SEQ", i);
              producer.send(message);
           }
           session.close();
           exConn.close();
  
           final int batch = 200;
           AtomicBoolean done = new AtomicBoolean(false);
           while (!done.get()) {
              // connection per batch attempt
              exConn = exFact.createConnection();
              ((ActiveMQConnection) exConn).setCloseTimeout(1); // so rollback on close won't block after socket close exception
  
              exConn.start();
  
              session = exConn.createSession(true, Session.SESSION_TRANSACTED);
  
              MessageConsumer messageConsumer = session.createConsumer(queue);
              TextMessage messageReceived;
              int received = 0;
              for (int j = 0; j < batch; j++) {
                 messageReceived = (TextMessage) messageConsumer.receive(2000);
                 if (messageReceived == null) {
                    done.set(true);
                    break;
                 }
                 received++;
                 batchConsumed.incrementAndGet();
                 assertEquals("This is a text message", messageReceived.getText());
  
                 int receivedSeq = messageReceived.getIntProperty("SEQ");
                 // need to infer batch from seq number and adjust - client never gets commit response
                 assertEquals((batch * (receivedSeq / batch)) + j, receivedSeq, "@:" + received + ", out of order");
              }
  
              // arrange concurrent commit - ack/commit
              // with server side error, potential for ack/commit and close-on-fail to contend
              final CountDownLatch latch = new CountDownLatch(1);
              Session finalSession = session;
              executorService.submit(() -> {
                 try {
                    latch.countDown();
                    finalSession.commit();
  
                 } catch (JMSException ignored) {
                 }
              });
  
              latch.await(1, TimeUnit.SECONDS);
              // force a local socket close such that the broker sees an exception on the connection and fails the consumer via serverConsumer close
              ((org.apache.activemq.ActiveMQConnection) exConn).getTransport().narrow(FailoverTransport.class).stop();
              // retry asap, not waiting for client close
              final Connection finalConToClose = exConn;
              executorService.submit(() -> {
                 try {
                    finalConToClose.close();
                 } catch (JMSException ignored) {
                 }
              });
           }
        } finally {
           if (exConn != null) {
              exConn.close();
           }
           executorService.shutdownNow();
        }
  
        logger.info("Done after: {}, queue: {}", batchConsumed.get(), server.locateQueue(durableQueue));
        try {
           Wait.assertEquals(0L, () -> server.locateQueue(durableQueue).getDeliveringCount(), 1000);
        } catch (Throwable e) {
  
           final AtomicBoolean doOut = new AtomicBoolean(false);
           PrintStream out = new PrintStream(System.out) {
  
              @Override
              public void println(String s) {
                 if (doOut.get()) {
                    super.println(s);
                 } else {
                    if (s.startsWith("### Failed Transactions")) {
                       doOut.set(true);
                       super.println(s);
                    }
                 }
              }
           };
           PrintData.printData(server.getConfiguration().getBindingsLocation(), server.getConfiguration().getJournalLocation(), server.getConfiguration().getPagingLocation(), out, true, true, true, false, -1);
  
           throw e;
        }
     }
  
  }
  