/*
 * 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.replication;

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.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.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
import org.apache.activemq.artemis.api.core.ActiveMQNotConnectedException;
import org.apache.activemq.artemis.api.core.Interceptor;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
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.ServerLocator;
import org.apache.activemq.artemis.core.config.ClusterConnectionConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.SharedStoreBackupPolicyConfiguration;
import org.apache.activemq.artemis.core.io.IOCallback;
import org.apache.activemq.artemis.core.io.SequentialFileFactory;
import org.apache.activemq.artemis.core.journal.EncodingSupport;
import org.apache.activemq.artemis.core.journal.IOCompletion;
import org.apache.activemq.artemis.core.journal.Journal;
import org.apache.activemq.artemis.core.journal.JournalLoadInformation;
import org.apache.activemq.artemis.core.journal.JournalUpdateCallback;
import org.apache.activemq.artemis.core.journal.LoaderCallback;
import org.apache.activemq.artemis.core.journal.PreparedTransactionInfo;
import org.apache.activemq.artemis.core.journal.RecordInfo;
import org.apache.activemq.artemis.core.journal.TransactionFailureCallback;
import org.apache.activemq.artemis.core.journal.impl.JournalFile;
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
import org.apache.activemq.artemis.core.paging.PagedMessage;
import org.apache.activemq.artemis.core.paging.PagingManager;
import org.apache.activemq.artemis.core.paging.PagingStore;
import org.apache.activemq.artemis.core.paging.impl.PagedMessageImpl;
import org.apache.activemq.artemis.core.paging.impl.PagingManagerImpl;
import org.apache.activemq.artemis.core.paging.impl.PagingStoreFactoryNIO;
import org.apache.activemq.artemis.core.persistence.OperationContext;
import org.apache.activemq.artemis.core.persistence.Persister;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.persistence.impl.journal.JournalStorageManager;
import org.apache.activemq.artemis.core.persistence.impl.journal.LargeServerMessageImpl;
import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
import org.apache.activemq.artemis.core.protocol.core.CoreRemotingConnection;
import org.apache.activemq.artemis.core.protocol.core.Packet;
import org.apache.activemq.artemis.core.replication.ReplicatedJournal;
import org.apache.activemq.artemis.core.replication.ReplicationManager;
import org.apache.activemq.artemis.core.server.ActiveMQComponent;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.LargeServerMessage;
import org.apache.activemq.artemis.core.server.cluster.ClusterController;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.settings.HierarchicalRepository;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.lockmanager.file.FileBasedLockManager;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
import org.apache.activemq.artemis.tests.extensions.parameterized.Parameter;
import org.apache.activemq.artemis.tests.extensions.parameterized.ParameterizedTestExtension;
import org.apache.activemq.artemis.tests.extensions.parameterized.Parameters;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.tests.util.ReplicatedBackupUtils;
import org.apache.activemq.artemis.tests.util.TransportConfigurationUtils;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
import org.apache.activemq.artemis.utils.ExecutorFactory;
import org.apache.activemq.artemis.utils.actors.OrderedExecutorFactory;
import org.apache.activemq.artemis.utils.collections.SparseArrayLinkedList;
import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;

@ExtendWith(ParameterizedTestExtension.class)
public final class ReplicationTest extends ActiveMQTestBase {

   @Parameter(index = 0)
   public boolean pluggableQuorum;

   @Parameters(name = "PluggableQuorum={0}")
   public static Iterable<Object[]> data() {
      return Arrays.asList(new Object[][]{{true}, {false}});
   }

   private ThreadFactory tFactory;
   private ExecutorService executor;
   private ExecutorFactory factory;
   private ScheduledExecutorService scheduledExecutor;

   private ActiveMQServer backupServer;
   /**
    * This field is not always used.
    */
   private ActiveMQServer liveServer;

   private ServerLocator locator;

   private ReplicationManager manager;
   private static final SimpleString ADDRESS = SimpleString.of("foobar123");

   private void setupServer(boolean backup, String... interceptors) throws Exception {
      this.setupServer(false, backup, null, interceptors);
   }

   private void setupServer(boolean useNetty,
                            boolean backup,
                            ExtraConfigurer extraConfig,
                            String... incomingInterceptors) throws Exception {
      TransportConfiguration primaryConnector = null;
      TransportConfiguration primaryAcceptor = null;
      TransportConfiguration backupConnector = null;
      TransportConfiguration backupAcceptor = null;
      if (useNetty) {
         primaryConnector = TransportConfigurationUtils.getNettyConnector(true, 0);
         primaryAcceptor = TransportConfigurationUtils.getNettyAcceptor(true, 0);
         backupConnector = TransportConfigurationUtils.getNettyConnector(false, 0);
         backupAcceptor = TransportConfigurationUtils.getNettyAcceptor(false, 0);
      } else {
         primaryConnector = TransportConfigurationUtils.getInVMConnector(true);
         backupConnector = TransportConfigurationUtils.getInVMConnector(false);
         backupAcceptor = TransportConfigurationUtils.getInVMAcceptor(false);
      }

      Configuration primaryConfig = createDefaultInVMConfig();

      Configuration backupConfig = createDefaultInVMConfig().setHAPolicyConfiguration(new SharedStoreBackupPolicyConfiguration()).setBindingsDirectory(getBindingsDir(0, true)).setJournalDirectory(getJournalDir(0, true)).setPagingDirectory(getPageDir(0, true)).setLargeMessagesDirectory(getLargeMessagesDir(0, true)).setIncomingInterceptorClassNames(incomingInterceptors.length > 0 ? Arrays.asList(incomingInterceptors) : new ArrayList<>());

      if (!pluggableQuorum) {
         ReplicatedBackupUtils.configureReplicationPair(backupConfig, backupConnector, backupAcceptor, primaryConfig, primaryConnector, primaryAcceptor);
      } else {
         DistributedLockManagerConfiguration managerConfiguration =
            new DistributedLockManagerConfiguration(FileBasedLockManager.class.getName(),
                                                    Collections.singletonMap("locks-folder", newFolder(temporaryFolder, "manager").toString()));

         ReplicatedBackupUtils.configurePluggableQuorumReplicationPair(backupConfig, backupConnector, backupAcceptor, primaryConfig, primaryConnector, primaryAcceptor, managerConfiguration, managerConfiguration);
      }

      if (extraConfig != null) {
         extraConfig.config(primaryConfig, backupConfig);
      }

      if (backup) {
         liveServer = createServer(primaryConfig);
         liveServer.start();
         waitForComponent(liveServer);
      }

      backupServer = createServer(backupConfig);
      if (useNetty) {
         locator = createNettyNonHALocator();
      } else {
         locator = createInVMNonHALocator();
      }

      backupServer.start();
      if (backup) {
         ActiveMQTestBase.waitForRemoteBackup(null, 5, true, backupServer);
      }
      int count = 0;
      waitForReplication(count);
   }

   private void waitForReplication(int count) throws InterruptedException {
      if (liveServer == null)
         return;

      while (liveServer.getReplicationManager() == null && count < 10) {
         Thread.sleep(50);
         count++;
      }
   }

   private static void waitForComponent(ActiveMQComponent component) throws Exception {
      waitForComponent(component, 3);
   }

   @TestTemplate
   public void testBasicConnection() throws Exception {
      setupServer(true);
      waitForComponent(liveServer.getReplicationManager());
   }

   @TestTemplate
   public void testConnectIntoNonBackup() throws Exception {
      setupServer(false);
      try {
         ClientSessionFactory sf = createSessionFactory(locator);
         manager = new ReplicationManager(null, (CoreRemotingConnection) sf.getConnection(), sf.getServerLocator().getCallTimeout(), sf.getServerLocator().getCallTimeout(), factory);
         addActiveMQComponent(manager);
         manager.start();
         fail("Exception was expected");
      } catch (ActiveMQNotConnectedException nce) {
         // ok
      } catch (ActiveMQException expected) {
         fail("Invalid Exception type:" + expected.getType());
      }
   }

   @TestTemplate
   public void testSendPackets() throws Exception {
      setupServer(true);

      JournalStorageManager storage = getStorage();

      manager = liveServer.getReplicationManager();
      waitForComponent(manager);

      Journal replicatedJournal = new ReplicatedJournal((byte) 1, new FakeJournal(), manager);

      replicatedJournal.appendPrepareRecord(1, new FakeData(), false);

      replicatedJournal.appendAddRecord(1, (byte) 1, new FakeData(), false);
      replicatedJournal.appendUpdateRecord(1, (byte) 2, new FakeData(), false);
      replicatedJournal.appendDeleteRecord(1, false);
      replicatedJournal.appendAddRecordTransactional(2, 2, (byte) 1, new FakeData());
      replicatedJournal.appendUpdateRecordTransactional(2, 2, (byte) 2, new FakeData());
      replicatedJournal.appendCommitRecord(2, false);

      replicatedJournal.appendDeleteRecordTransactional(3, 4, new FakeData());
      replicatedJournal.appendPrepareRecord(3, new FakeData(), false);
      replicatedJournal.appendRollbackRecord(3, false);

      blockOnReplication(storage, manager);

      assertTrue(manager.getActiveTokens().isEmpty(), "Expecting no active tokens:" + manager.getActiveTokens());

      CoreMessage msg = new CoreMessage().initBuffer(1024).setMessageID(1);

      SimpleString dummy = SimpleString.of("dummy");
      msg.setAddress(dummy);

      replicatedJournal.appendAddRecordTransactional(23, 24, (byte) 1, new FakeData());

      PagedMessage pgmsg = new PagedMessageImpl(msg, new long[0]);
      manager.pageWrite(dummy, pgmsg, 1, true);
      manager.pageWrite(dummy, pgmsg, 2, true);
      manager.pageWrite(dummy, pgmsg, 3, true);
      manager.pageWrite(dummy, pgmsg, 4, true);

      blockOnReplication(storage, manager);

      PagingManager pagingManager = createPageManager(backupServer.getStorageManager(), backupServer.getConfiguration(), backupServer.getExecutorFactory(), backupServer.getScheduledPool(), backupServer.getAddressSettingsRepository());

      PagingStore store = pagingManager.getPageStore(dummy);
      store.start();
      assertEquals(4, store.getNumberOfPages());
      store.stop();

      manager.pageDeleted(dummy, 1);
      manager.pageDeleted(dummy, 2);
      manager.pageDeleted(dummy, 3);
      manager.pageDeleted(dummy, 4);
      manager.pageDeleted(dummy, 5);
      manager.pageDeleted(dummy, 6);

      blockOnReplication(storage, manager);

      CoreMessage serverMsg = new CoreMessage();
      serverMsg.setMessageID(500);
      serverMsg.setAddress(SimpleString.of("tttt"));

      ActiveMQBuffer buffer = ActiveMQBuffers.dynamicBuffer(100);
      serverMsg.encodeHeadersAndProperties(buffer.byteBuf());

      manager.largeMessageBegin(500);

      manager.largeMessageWrite(500, new byte[1024]);

      manager.largeMessageDelete(500L, storage);

      blockOnReplication(storage, manager);

      store.start();

      assertEquals(0, store.getNumberOfPages());
   }

   @TestTemplate
   public void testSendPacketsWithFailure() throws Exception {
      final int nMsg = 100;
      final int stop = 37;
      setupServer(true, TestInterceptor.class.getName());

      manager = liveServer.getReplicationManager();
      waitForComponent(manager);
      ClientSessionFactory sf = createSessionFactory(locator);
      final ClientSession session = sf.createSession();
      final ClientSession session2 = sf.createSession();
      session.createQueue(QueueConfiguration.of(ADDRESS));

      final ClientProducer producer = session.createProducer(ADDRESS);

      session.start();
      session2.start();
      try {
         final ClientConsumer consumer = session2.createConsumer(ADDRESS);
         for (int i = 0; i < nMsg; i++) {

            ClientMessage message = session.createMessage(true);
            setBody(i, message);
            message.putIntProperty("counter", i);
            producer.send(message);
            if (i == stop) {
               // Now we start intercepting the communication with the backup
               TestInterceptor.value.set(false);
            }
            ClientMessage msgRcvd = consumer.receive(1000);
            assertNotNull(msgRcvd, "Message should exist!");
            assertMessageBody(i, msgRcvd);
            assertEquals(i, msgRcvd.getIntProperty("counter").intValue());
            msgRcvd.acknowledge();
         }
      } finally {
         TestInterceptor.value.set(false);
         if (!session.isClosed())
            session.close();
         if (!session2.isClosed())
            session2.close();
      }
   }

   @TestTemplate
   public void testSendMessageWithLargeHeader() throws Exception {
      setupServer(true, TestInterceptor.class.getName());

      manager = liveServer.getReplicationManager();
      waitForComponent(manager);

      ClientSessionFactory sf = createSessionFactory(locator);

      try (ClientSession producerSession = sf.createSession();
           ClientSession consumerSession = sf.createSession()) {

         producerSession.start();
         producerSession.createQueue(QueueConfiguration.of(ADDRESS));
         ClientProducer producer = producerSession.createProducer(ADDRESS);

         consumerSession.start();
         ClientConsumer consumer = consumerSession.createConsumer(ADDRESS);

         ClientMessage messageBefore = producerSession.createMessage(true);
         setBody(0, messageBefore);
         messageBefore.putIntProperty("counter", 0);
         producer.send(messageBefore);

         ClientMessage messageReceivedBefore = consumer.receive(1000);
         assertNotNull(messageReceivedBefore, "Message should exist!");
         assertMessageBody(0, messageReceivedBefore);
         assertEquals(0, messageReceivedBefore.getIntProperty("counter").intValue());
         messageReceivedBefore.acknowledge();

         ClientMessage messageWithLargeHeader = producerSession.createMessage(true);
         setBody(1, messageWithLargeHeader);
         messageWithLargeHeader.putIntProperty("counter", 1);
         messageWithLargeHeader.putStringProperty("large-property", "z".repeat(512 * 1024));
         try {
            producer.send(messageWithLargeHeader);
            fail();
         } catch (Exception e) {
            assertTrue(e.getMessage().contains("AMQ149005"));
         }

         ClientMessage messageAfter = producerSession.createMessage(true);
         setBody(2, messageAfter);
         messageAfter.putIntProperty("counter", 2);
         producer.send(messageAfter);

         ClientMessage messageReceivedAfter = consumer.receive(1000);
         assertNotNull(messageReceivedAfter, "Message should exist!");
         assertMessageBody(2, messageReceivedAfter);
         assertEquals(2, messageReceivedAfter.getIntProperty("counter").intValue());
         messageReceivedAfter.acknowledge();

         assertNull(consumer.receiveImmediate());
      }
   }

   @TestTemplate
   public void testExceptionSettingActionBefore() throws Exception {
      OperationContext ctx = OperationContextImpl.getContext(factory);

      ctx.storeLineUp();

      String msg = "I'm an exception";

      ctx.onError(ActiveMQExceptionType.UNBLOCKED.getCode(), msg);

      final AtomicInteger lastError = new AtomicInteger(0);

      final List<String> msgsResult = new ArrayList<>();

      final CountDownLatch latch = new CountDownLatch(1);

      ctx.executeOnCompletion(new IOCallback() {
         @Override
         public void onError(final int errorCode, final String errorMessage) {
            lastError.set(errorCode);
            msgsResult.add(errorMessage);
            latch.countDown();
         }

         @Override
         public void done() {
         }
      });

      assertTrue(latch.await(5, TimeUnit.SECONDS));

      assertEquals(5, lastError.get());

      assertEquals(1, msgsResult.size());

      assertEquals(msg, msgsResult.get(0));

      final CountDownLatch latch2 = new CountDownLatch(1);

      // Adding the Task after the exception should still throw an exception
      ctx.executeOnCompletion(new IOCallback() {
         @Override
         public void onError(final int errorCode, final String errorMessage) {
            lastError.set(errorCode);
            msgsResult.add(errorMessage);
            latch2.countDown();
         }

         @Override
         public void done() {
         }
      });

      assertTrue(latch2.await(5, TimeUnit.SECONDS));

      assertEquals(2, msgsResult.size());

      assertEquals(msg, msgsResult.get(0));

      assertEquals(msg, msgsResult.get(1));

      final CountDownLatch latch3 = new CountDownLatch(1);

      ctx.executeOnCompletion(new IOCallback() {
         @Override
         public void onError(final int errorCode, final String errorMessage) {
         }

         @Override
         public void done() {
            latch3.countDown();
         }
      });

      assertTrue(latch2.await(5, TimeUnit.SECONDS));

   }

   @TestTemplate
   public void testClusterConnectionConfigs() throws Exception {
      final long ttlOverride = 123456789;
      final long checkPeriodOverride = 987654321;

      ExtraConfigurer configurer = (primaryConfig, backupConfig) -> {
         List<ClusterConnectionConfiguration> ccList = backupConfig.getClusterConfigurations();
         assertFalse(ccList.isEmpty());
         ClusterConnectionConfiguration cc = ccList.get(0);
         cc.setConnectionTTL(ttlOverride);
         cc.setClientFailureCheckPeriod(checkPeriodOverride);
      };
      this.setupServer(true, true, configurer);
      assertInstanceOf(ActiveMQServerImpl.class, backupServer);

      ClusterController controller = backupServer.getClusterManager().getClusterController();

      ServerLocator replicationLocator = controller.getReplicationLocator();

      assertNotNull(replicationLocator);

      assertEquals(ttlOverride, replicationLocator.getConnectionTTL());
      assertEquals(checkPeriodOverride, replicationLocator.getClientFailureCheckPeriod());
   }

   private JournalStorageManager getStorage() throws Exception {
      return new JournalStorageManager(createDefaultInVMConfig(), EmptyCriticalAnalyzer.getInstance(), factory, factory);
   }

   private void blockOnReplication(final StorageManager storage, final ReplicationManager manager1) throws Exception {
      final CountDownLatch latch = new CountDownLatch(1);
      storage.afterCompleteOperations(new IOCallback() {

         @Override
         public void onError(final int errorCode, final String errorMessage) {
         }

         @Override
         public void done() {
            latch.countDown();
         }
      });

      assertTrue(latch.await(30, TimeUnit.SECONDS));
   }

   @TestTemplate
   public void testNoActions() throws Exception {

      setupServer(true);
      StorageManager storage = getStorage();
      manager = liveServer.getReplicationManager();
      waitForComponent(manager);

      Journal replicatedJournal = new ReplicatedJournal((byte) 1, new FakeJournal(), manager);

      replicatedJournal.appendPrepareRecord(1, new FakeData(), false);

      final CountDownLatch latch = new CountDownLatch(1);
      storage.afterCompleteOperations(new IOCallback() {

         @Override
         public void onError(final int errorCode, final String errorMessage) {
         }

         @Override
         public void done() {
            latch.countDown();
         }
      });

      assertTrue(latch.await(1, TimeUnit.SECONDS));

      assertEquals(0, manager.getActiveTokens().size(), "should be empty " + manager.getActiveTokens());
   }

   @TestTemplate
   public void testOrderOnNonPersistency() throws Exception {

      setupServer(true);

      final ArrayList<Integer> executions = new ArrayList<>();

      StorageManager storage = getStorage();
      manager = liveServer.getReplicationManager();
      Journal replicatedJournal = new ReplicatedJournal((byte) 1, new FakeJournal(), manager);

      int numberOfAdds = 200;

      final CountDownLatch latch = new CountDownLatch(numberOfAdds);

      OperationContext ctx = storage.getContext();

      for (int i = 0; i < numberOfAdds; i++) {
         final int nAdd = i;

         if (i % 2 == 0) {
            replicatedJournal.appendPrepareRecord(i, new FakeData(), false);
         }

         ctx.executeOnCompletion(new IOCallback() {

            @Override
            public void onError(final int errorCode, final String errorMessage) {
            }

            @Override
            public void done() {
               executions.add(nAdd);
               latch.countDown();
            }
         });
      }

      assertTrue(latch.await(10, TimeUnit.SECONDS));

      for (int i = 0; i < numberOfAdds; i++) {
         assertEquals(i, executions.get(i).intValue());
      }

      assertEquals(0, manager.getActiveTokens().size());
   }

   @TestTemplate
   public void testReplicationLargeMessageFileClose() throws Exception {
      setupServer(true);

      JournalStorageManager storage = getStorage();

      manager = liveServer.getReplicationManager();
      waitForComponent(manager);

      CoreMessage msg = new CoreMessage().initBuffer(1024).setMessageID(1);
      LargeServerMessage largeMsg = liveServer.getStorageManager().createCoreLargeMessage(500, msg);
      largeMsg.addBytes(new byte[1024]);

      blockOnReplication(storage, manager);

      LargeServerMessageImpl message1 = (LargeServerMessageImpl) getReplicationEndpoint(backupServer).getLargeMessages().get(500L);


      assertNotNull(message1);
      assertTrue(largeMsg.getAppendFile().isOpen());
      assertTrue(message1.getAppendFile().isOpen());

      largeMsg.releaseResources(true, true);

      Wait.assertTrue(() -> getReplicationEndpoint(backupServer).getLargeMessages().get(500L) == null, 5000);

      assertFalse(largeMsg.getAppendFile().isOpen());
      assertFalse(message1.getAppendFile().isOpen());
   }

   class FakeData implements EncodingSupport {

      @Override
      public void decode(final ActiveMQBuffer buffer) {
      }

      @Override
      public void encode(final ActiveMQBuffer buffer) {
         buffer.writeBytes(new byte[5]);
      }

      @Override
      public int getEncodeSize() {
         return 5;
      }

   }

   @Override
   @BeforeEach
   public void setUp() throws Exception {
      super.setUp();

      tFactory = new ActiveMQThreadFactory("ActiveMQ-ReplicationTest", false, this.getClass().getClassLoader());

      executor = Executors.newCachedThreadPool(tFactory);

      scheduledExecutor = new ScheduledThreadPoolExecutor(10, tFactory);

      factory = new OrderedExecutorFactory(executor);
   }

   /**
    * We need to shutdown the executors before calling {@link ActiveMQTestBase#tearDown()} (which will check for leaking
    * threads). Due to that, we need to close/stop all components here.
    */
   @Override
   @AfterEach
   public void tearDown() throws Exception {
      stopComponent(manager);
      manager = null;
      closeServerLocator(locator);
      stopComponent(backupServer);
      backupServer = null;
      stopComponent(liveServer);
      liveServer = null;

      executor.shutdownNow();
      scheduledExecutor.shutdownNow();

      tFactory = null;
      scheduledExecutor = null;

      super.tearDown();
   }

   protected PagingManager createPageManager(final StorageManager storageManager,
                                             final Configuration configuration,
                                             final ExecutorFactory executorFactory,
                                             final ScheduledExecutorService scheduledExecutorService,
                                             final HierarchicalRepository<AddressSettings> addressSettingsRepository) throws Exception {

      PagingManager paging = new PagingManagerImpl(new PagingStoreFactoryNIO(storageManager, configuration.getPagingLocation(), 1000, scheduledExecutorService, executorFactory, false, null), addressSettingsRepository, configuration.getManagementAddress());

      paging.start();
      runAfter(paging::stop);
      return paging;
   }


   public static final class TestInterceptor implements Interceptor {

      static AtomicBoolean value = new AtomicBoolean(true);

      @Override
      public boolean intercept(final Packet packet, final RemotingConnection connection) throws ActiveMQException {
         return TestInterceptor.value.get();
      }

   }

   static final class FakeJournal implements Journal {

      @Override
      public void setRemoveExtraFilesOnLoad(boolean removeExtraFilesOnLoad) {

      }

      @Override
      public void appendAddEvent(long id,
                                 byte recordType,
                                 Persister persister,
                                 Object record,
                                 boolean sync,
                                 IOCompletion completionCallback) throws Exception {

      }

      @Override
      public boolean isRemoveExtraFilesOnLoad() {
         return false;
      }

      @Override
      public void appendAddRecord(long id,
                                  byte recordType,
                                  Persister persister,
                                  Object record,
                                  boolean sync) throws Exception {

      }

      @Override
      public void appendAddRecord(long id,
                                  byte recordType,
                                  Persister persister,
                                  Object record,
                                  boolean sync,
                                  IOCompletion completionCallback) throws Exception {

      }

      @Override
      public void appendUpdateRecord(long id,
                                     byte recordType,
                                     Persister persister,
                                     Object record,
                                     boolean sync) throws Exception {

      }

      @Override
      public void tryAppendUpdateRecord(long id,
                                           byte recordType,
                                           Persister persister,
                                           Object record, JournalUpdateCallback updateCallback,
                                           boolean sync,
                                           boolean repalceableUpdate) throws Exception {
      }

      @Override
      public void appendUpdateRecord(long id,
                                     byte recordType,
                                     Persister persister,
                                     Object record,
                                     boolean sync,
                                     IOCompletion callback) throws Exception {

      }

      @Override
      public void tryAppendUpdateRecord(long id,
                                           byte recordType,
                                           Persister persister,
                                           Object record,
                                           boolean sync, boolean replaceableUpdate, JournalUpdateCallback updateCallback,
                                           IOCompletion callback) throws Exception {
      }

      @Override
      public void appendAddRecordTransactional(long txID,
                                               long id,
                                               byte recordType,
                                               Persister persister,
                                               Object record) throws Exception {

      }

      @Override
      public void appendUpdateRecordTransactional(long txID,
                                                  long id,
                                                  byte recordType,
                                                  Persister persister,
                                                  Object record) throws Exception {

      }

      @Override
      public void appendAddRecord(final long id,
                                  final byte recordType,
                                  final byte[] record,
                                  final boolean sync) throws Exception {

      }

      @Override
      public void appendAddRecord(final long id,
                                  final byte recordType,
                                  final EncodingSupport record,
                                  final boolean sync) throws Exception {

      }

      @Override
      public void appendAddRecordTransactional(final long txID,
                                               final long id,
                                               final byte recordType,
                                               final byte[] record) throws Exception {

      }

      @Override
      public void appendAddRecordTransactional(final long txID,
                                               final long id,
                                               final byte recordType,
                                               final EncodingSupport record) throws Exception {

      }

      @Override
      public void flush() throws Exception {

      }

      @Override
      public long getMaxRecordSize() {
         return ActiveMQDefaultConfiguration.getDefaultJournalBufferSizeAio();
      }

      @Override
      public void appendCommitRecord(final long txID, final boolean sync) throws Exception {

      }

      @Override
      public void appendDeleteRecord(final long id, final boolean sync) throws Exception {

      }

      @Override
      public void tryAppendDeleteRecord(long id, JournalUpdateCallback updateConsumer, boolean sync) throws Exception {
      }

      @Override
      public void appendDeleteRecordTransactional(final long txID,
                                                  final long id,
                                                  final byte[] record) throws Exception {

      }

      @Override
      public void appendDeleteRecordTransactional(final long txID,
                                                  final long id,
                                                  final EncodingSupport record) throws Exception {

      }

      @Override
      public void appendDeleteRecordTransactional(final long txID, final long id) throws Exception {

      }

      @Override
      public void appendPrepareRecord(final long txID,
                                      final EncodingSupport transactionData,
                                      final boolean sync) throws Exception {

      }

      @Override
      public void appendPrepareRecord(final long txID,
                                      final byte[] transactionData,
                                      final boolean sync) throws Exception {

      }

      @Override
      public void appendRollbackRecord(final long txID, final boolean sync) throws Exception {

      }

      @Override
      public void appendUpdateRecord(final long id,
                                     final byte recordType,
                                     final byte[] record,
                                     final boolean sync) throws Exception {

      }

      @Override
      public void tryAppendUpdateRecord(long id, byte recordType, byte[] record, JournalUpdateCallback updateCallback, boolean sync, boolean replaceable) throws Exception {
      }

      @Override
      public void appendUpdateRecord(final long id,
                                     final byte recordType,
                                     final EncodingSupport record,
                                     final boolean sync) throws Exception {

      }

      @Override
      public void appendUpdateRecordTransactional(final long txID,
                                                  final long id,
                                                  final byte recordType,
                                                  final byte[] record) throws Exception {

      }

      @Override
      public void appendUpdateRecordTransactional(final long txID,
                                                  final long id,
                                                  final byte recordType,
                                                  final EncodingSupport record) throws Exception {

      }

      @Override
      public int getAlignment() throws Exception {

         return 0;
      }

      @Override
      public JournalLoadInformation load(final LoaderCallback reloadManager) throws Exception {

         return new JournalLoadInformation();
      }

      @Override
      public JournalLoadInformation load(final SparseArrayLinkedList<RecordInfo> committedRecords,
                                         final List<PreparedTransactionInfo> preparedTransactions,
                                         final TransactionFailureCallback transactionFailure,
                                         final boolean fixbadtx) throws Exception {

         return new JournalLoadInformation();
      }

      @Override
      public JournalLoadInformation load(final List<RecordInfo> committedRecords,
                                         final List<PreparedTransactionInfo> preparedTransactions,
                                         final TransactionFailureCallback transactionFailure,
                                         final boolean fixbadtx) throws Exception {

         return new JournalLoadInformation();
      }

      @Override
      public boolean isStarted() {

         return false;
      }

      @Override
      public void start() throws Exception {

      }

      @Override
      public void stop() throws Exception {

      }

      @Override
      public JournalLoadInformation loadInternalOnly() throws Exception {
         return new JournalLoadInformation();
      }

      @Override
      public int getNumberOfRecords() {
         return 0;
      }

      @Override
      public void appendAddRecord(final long id,
                                  final byte recordType,
                                  final EncodingSupport record,
                                  final boolean sync,
                                  final IOCompletion completionCallback) throws Exception {
      }

      @Override
      public void appendCommitRecord(final long txID,
                                     final boolean sync,
                                     final IOCompletion callback) throws Exception {
      }

      @Override
      public void appendDeleteRecord(final long id,
                                     final boolean sync,
                                     final IOCompletion completionCallback) throws Exception {
      }

      @Override
      public void tryAppendDeleteRecord(long id, boolean sync, JournalUpdateCallback updateCallback, IOCompletion completionCallback) throws Exception {
      }

      @Override
      public void appendPrepareRecord(final long txID,
                                      final EncodingSupport transactionData,
                                      final boolean sync,
                                      final IOCompletion callback) throws Exception {
      }

      @Override
      public void appendRollbackRecord(final long txID,
                                       final boolean sync,
                                       final IOCompletion callback) throws Exception {
      }

      @Override
      public void appendUpdateRecord(final long id,
                                     final byte recordType,
                                     final EncodingSupport record,
                                     final boolean sync,
                                     final IOCompletion completionCallback) throws Exception {
      }

      public void sync(final IOCompletion callback) {
      }

      @Override
      public int getUserVersion() {
         return 0;
      }

      @Override
      public void appendCommitRecord(long txID,
                                     boolean sync,
                                     IOCompletion callback,
                                     boolean lineUpContext) throws Exception {

      }

      @Override
      public void lineUpContext(IOCompletion callback) {

      }

      @Override
      public JournalLoadInformation loadSyncOnly(JournalState s) throws Exception {
         return null;
      }

      @Override
      public Map<Long, JournalFile> createFilesForBackupSync(long[] fileIds) throws Exception {
         return null;
      }

      @Override
      public void synchronizationLock() {

      }

      @Override
      public void synchronizationUnlock() {

      }

      @Override
      public void forceMoveNextFile() throws Exception {

      }

      @Override
      public JournalFile[] getDataFiles() {
         return null;
      }

      @Override
      public SequentialFileFactory getFileFactory() {
         return null;
      }

      @Override
      public int getFileSize() {
         return 0;
      }

      @Override
      public void scheduleCompactAndBlock(int timeout) throws Exception {
      }

      @Override
      public void replicationSyncPreserveOldFiles() {
         // no-op
      }

      @Override
      public void replicationSyncFinished() {
         // no-op
      }

      @Override
      public long getWarningRecordSize() {
         return getMaxRecordSize() - 2048;
      }
   }

   private interface ExtraConfigurer {

      void config(Configuration primaryConfig, Configuration backupConfig);
   }

   private static File newFolder(File root, String subFolder) throws IOException {
      File result = new File(root, subFolder);
      if (!result.mkdirs()) {
         throw new IOException("Couldn't create folders " + root);
      }
      return result;
   }
}
