/*
 * 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.geode.internal.cache.ha;

import static org.apache.geode.cache.Region.Entry;
import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.distributed.ConfigurationProperties.DELTA_PROPAGATION;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
import static org.apache.geode.internal.cache.CacheServerImpl.generateNameForClientMsgsRegion;
import static org.apache.geode.internal.lang.SystemProperty.GEMFIRE_PREFIX;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.NetworkUtils.getServerHostName;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.apache.logging.log4j.Logger;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.GemFireException;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache30.ClientServerTestCase;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.internal.cache.CacheServerImpl;
import org.apache.geode.internal.cache.InternalCacheServer;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.tier.sockets.ClientUpdateMessage;
import org.apache.geode.internal.cache.tier.sockets.ConflationDistributedTestHelper;
import org.apache.geode.internal.cache.tier.sockets.HAEventWrapper;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.SerializableRunnableIF;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
import org.apache.geode.test.dunit.rules.DistributedRule;
import org.apache.geode.test.junit.categories.ClientSubscriptionTest;

/**
 * This DUnit contains various tests to ensure new implementation of ha region queues works as
 * expected.
 *
 * @since GemFire 5.7
 */
@Category({ClientSubscriptionTest.class})
public class HARQueueNewImplDistributedTest extends JUnit4DistributedTestCase {

  private static final String regionName = HARQueueNewImplDistributedTest.class.getSimpleName();
  private static final Map<Object, Object> map = new HashMap<>();

  private static Cache cache = null;
  private static VM serverVM0 = null;
  private static VM serverVM1 = null;
  private static VM clientVM1 = null;
  private static VM clientVM2 = null;

  private static final Logger logger = LogService.getLogger();
  private static int numOfCreates = 0;
  private static int numOfUpdates = 0;
  private static int numOfInvalidates = 0;
  private static Object[] deletedValues = null;

  private int PORT1;
  private int PORT2;

  @Rule
  public DistributedRule distributedRule = new DistributedRule();

  /**
   * Sets up the test.
   */
  @Before
  public void setUp() {
    map.clear();

    serverVM0 = VM.getVM(0);
    serverVM1 = VM.getVM(1);
    clientVM1 = VM.getVM(2);
    clientVM2 = VM.getVM(3);

    PORT1 = serverVM0.invoke(
        () -> HARQueueNewImplDistributedTest
            .createServerCache(HARegionQueue.HA_EVICTION_POLICY_MEMORY));
    PORT2 = serverVM1.invoke(
        () -> HARQueueNewImplDistributedTest
            .createServerCache(HARegionQueue.HA_EVICTION_POLICY_ENTRY));

    numOfCreates = 0;
    numOfUpdates = 0;
    numOfInvalidates = 0;
    clientVM1.invoke(() -> {
      numOfCreates = 0;
      numOfUpdates = 0;
      numOfInvalidates = 0;
    });
  }

  /**
   * Tears down the test.
   */
  @After
  public void tearDown() {
    map.clear();

    closeCache();
    clientVM1.invoke(HARQueueNewImplDistributedTest::closeCache);
    clientVM2.invoke(HARQueueNewImplDistributedTest::closeCache);

    // Unset the isSlowStartForTesting flag
    serverVM0.invoke(ConflationDistributedTestHelper::unsetIsSlowStart);
    serverVM1.invoke(ConflationDistributedTestHelper::unsetIsSlowStart);

    // then close the servers
    serverVM0.invoke(HARQueueNewImplDistributedTest::closeCache);
    serverVM1.invoke(HARQueueNewImplDistributedTest::closeCache);


    disconnectAllFromDS();
  }

  private void createCache(Properties props) throws Exception {
    props.setProperty(DELTA_PROPAGATION, "false");
    DistributedSystem ds = getSystem(props);
    ds.disconnect();
    ds = getSystem(props);
    assertThat(ds).isNotNull();
    cache = CacheFactory.create(ds);
    assertThat(cache).isNotNull();
  }

  public static Integer createServerCache() throws Exception {
    return createServerCache(null);
  }

  public static Integer createServerCache(String ePolicy) throws Exception {
    return createServerCache(ePolicy, 1);
  }

  public static Integer createServerCache(String ePolicy, Integer cap) throws Exception {
    new HARQueueNewImplDistributedTest().createCache(new Properties());
    RegionFactory<Object, Object> factory = cache.createRegionFactory();
    factory.setScope(Scope.DISTRIBUTED_ACK);
    factory.setDataPolicy(DataPolicy.REPLICATE);
    factory.create(regionName);

    int port = getRandomAvailableTCPPort();
    CacheServer server1 = cache.addCacheServer();
    server1.setPort(port);
    if (ePolicy != null) {
      File overflowDirectory = new File("bsi_overflow_" + port);
      overflowDirectory.mkdir();
      DiskStoreFactory dsf = cache.createDiskStoreFactory();
      File[] dirs1 = new File[] {overflowDirectory};

      server1.getClientSubscriptionConfig().setEvictionPolicy(ePolicy);
      server1.getClientSubscriptionConfig().setCapacity(cap);
      // specify disk store for this server
      server1.getClientSubscriptionConfig()
          .setDiskStoreName(dsf.setDiskDirs(dirs1).create("bsi").getName());
    }
    server1.start();
    return server1.getPort();
  }

  private static Integer createOneMoreBridgeServer(Boolean notifyBySubscription) throws Exception {
    int port = getRandomAvailableTCPPort();
    CacheServer server1 = cache.addCacheServer();
    server1.setPort(port);
    server1.setNotifyBySubscription(notifyBySubscription);
    server1.getClientSubscriptionConfig()
        .setEvictionPolicy(HARegionQueue.HA_EVICTION_POLICY_MEMORY);
    // let this server to use default disk store
    server1.start();
    return server1.getPort();
  }

  public static void createClientCache(String host, Integer port1, Integer port2, String rLevel,
      Boolean addListener) throws Exception {
    System.setProperty(GEMFIRE_PREFIX + "bridge.disableShufflingOfEndpoints", "true");

    Properties props = new Properties();
    props.setProperty(MCAST_PORT, "0");
    props.setProperty(LOCATORS, "");
    new HARQueueNewImplDistributedTest().createCache(props);
    AttributesFactory<Object, Object> factory = new AttributesFactory<>();
    ClientServerTestCase
        .configureConnectionPool(factory, host, port1, port2, true,
            Integer.parseInt(rLevel),
            2, null, 1000, 250,
            -2);

    factory.setScope(Scope.LOCAL);

    if (addListener) {
      factory.addCacheListener(new CacheListenerAdapter<Object, Object>() {
        @Override
        public void afterInvalidate(EntryEvent event) {
          logger.debug("Invalidate Event: <" + event.getKey() + ", " + event.getNewValue() + ">");
          numOfInvalidates++;
        }

        @Override
        public void afterCreate(EntryEvent event) {
          logger.debug("Create Event: <" + event.getKey() + ", " + event.getNewValue() + ">");
          numOfCreates++;
        }

        @Override
        public void afterUpdate(EntryEvent event) {
          logger.debug("Update Event: <" + event.getKey() + ", " + event.getNewValue() + ">");
          numOfUpdates++;
        }
      });
    }

    cache.createRegion(regionName, factory.create());
  }

  public static void createClientCache(String host, Integer port1, Integer port2, String rLevel)
      throws Exception {
    createClientCache(host, port1, port2, rLevel, Boolean.FALSE);
  }

  private static void registerInterestListAll() {
    try {
      Region<Object, Object> region = cache.getRegion(SEPARATOR + regionName);
      assertThat(region).isNotNull();
      region.registerInterest("ALL_KEYS");
    } catch (GemFireException ex) {
      fail("failed in registerInterestListAll", ex);
    }
  }

  private static void registerInterestList() {
    try {
      Region<Object, Object> region = cache.getRegion(SEPARATOR + regionName);
      assertThat(region).isNotNull();
      region.registerInterest("k1");
      region.registerInterest("k3");
      region.registerInterest("k5");
    } catch (GemFireException ex) {
      fail("failed while registering keys", ex);
    }
  }

  private static void putEntries() {
    try {

      Region<Object, Object> region = cache.getRegion(SEPARATOR + regionName);
      assertThat(region).isNotNull();

      region.put("k1", "pv1");
      region.put("k2", "pv2");
      region.put("k3", "pv3");
      region.put("k4", "pv4");
      region.put("k5", "pv5");
    } catch (GemFireException ex) {
      fail("failed in putEntries()", ex);
    }
  }

  public static void createEntries() {
    try {
      Region<Object, Object> region = cache.getRegion(SEPARATOR + regionName);
      assertThat(region).isNotNull();

      region.create("k1", "v1");
      region.create("k2", "v2");
      region.create("k3", "v3");
      region.create("k4", "v4");
      region.create("k5", "v5");
    } catch (GemFireException ex) {
      fail("failed in createEntries()", ex);
    }
  }

  public static void createEntries(Long num) {
    try {
      Region<Object, Object> region = cache.getRegion(SEPARATOR + regionName);
      assertThat(region).isNotNull();
      for (long i = 0; i < num; i++) {
        region.create("k" + i, "v" + i);
      }
    } catch (GemFireException ex) {
      fail("failed in createEntries(Long)", ex);
    }
  }

  private static void putHeavyEntries(Integer num) {
    try {
      byte[] val;
      Region<Object, Object> region = cache.getRegion(SEPARATOR + regionName);
      assertThat(region).isNotNull();
      for (long i = 0; i < num; i++) {
        val = new byte[1024 * 1024 * 5]; // 5 MB
        region.put("k0", val);
      }
    } catch (GemFireException ex) {
      fail("failed in putHeavyEntries(Long)", ex);
    }
  }

  /**
   * This test verifies that the client-messages-region does not store duplicate
   * ClientUpdateMessageImpl instances, during a normal put path as well as the GII path.
   */
  @Test
  public void testClientMsgsRegionSize() throws Exception {
    // slow start for dispatcher
    serverVM0.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("30000"));
    serverVM1.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("30000"));

    createClientCache(getServerHostName(), PORT1, PORT2,
        "1");
    final String client1Host = getServerHostName();
    clientVM1.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client1Host,
        PORT1, PORT2, "1"));
    final String client2Host = getServerHostName();
    clientVM2.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client2Host,
        PORT1, PORT2, "1"));

    registerInterestListAll();
    clientVM1.invoke(HARQueueNewImplDistributedTest::registerInterestList);
    clientVM2.invoke(HARQueueNewImplDistributedTest::registerInterestList);

    serverVM1.invoke(HARQueueNewImplDistributedTest::stopServer);

    serverVM0.invoke((SerializableRunnableIF) HARQueueNewImplDistributedTest::createEntries);

    serverVM1.invoke(HARQueueNewImplDistributedTest::startServer);

    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.verifyRegionSize(5, 5));
    serverVM1.invoke(() -> HARQueueNewImplDistributedTest.verifyRegionSize(5, 5));
  }

  /**
   * This test verifies that the ha-region-queues increment the reference count of their respective
   * HAEventWrapper instances in the client-messages-region correctly, during put as well as GII
   * path.
   */
  @Test
  public void testRefCountForNormalAndGIIPut() throws Exception {
    // slow start for dispatcher
    serverVM0.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("240000"));
    serverVM1.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("240000"));

    createClientCache(getServerHostName(), PORT1, PORT2,
        "1");
    final String client1Host = getServerHostName();
    clientVM1.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client1Host,
        PORT1, PORT2, "1"));
    final String client2Host = getServerHostName();
    clientVM2.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client2Host,
        PORT1, PORT2, "1"));

    registerInterestListAll();
    clientVM1.invoke(HARQueueNewImplDistributedTest::registerInterestList);
    clientVM2.invoke(HARQueueNewImplDistributedTest::registerInterestList);

    serverVM1.invoke(HARQueueNewImplDistributedTest::stopServer);

    serverVM0.invoke((SerializableRunnableIF) HARQueueNewImplDistributedTest::createEntries);

    serverVM1.invoke(HARQueueNewImplDistributedTest::startServer);

    serverVM1.invoke(() -> ValidateRegionSizes(PORT2));
    serverVM0.invoke(() -> ValidateRegionSizes(PORT1));


    serverVM0.invoke(HARQueueNewImplDistributedTest::updateMapForVM0);
    serverVM1.invoke(HARQueueNewImplDistributedTest::updateMapForVM1);

    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.verifyQueueData(
        PORT1));
    serverVM1.invoke(() -> HARQueueNewImplDistributedTest.verifyQueueData(
        PORT2));
  }

  private void ValidateRegionSizes(int port) {
    await().untilAsserted(() -> {
      Region region = cache.getRegion(SEPARATOR + regionName);
      Region<Object, Object> msgsRegion =
          cache.getRegion(CacheServerImpl.generateNameForClientMsgsRegion(port));
      int clientMsgRegionSize = msgsRegion.size();
      int regionSize = region.size();
      assertThat(((5 == clientMsgRegionSize) && (5 == regionSize))).describedAs(
          "Region sizes were not as expected after 60 seconds elapsed. Actual region size = "
              + regionSize + "Actual client msg region size = " + clientMsgRegionSize)
          .isTrue();
    });
  }

  /**
   * This test verifies that the ha-region-queues decrement the reference count of their respective
   * HAEventWrapper instances in the client-messages-region correctly, after the events have been
   * peeked and removed from the queue.
   */
  @Test
  public void testRefCountForPeekAndRemove() throws Exception {
    serverVM0.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("30000"));

    createClientCache(getServerHostName(), PORT1, PORT2,
        "1");
    final String client1Host = getServerHostName();
    clientVM1.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client1Host,
        PORT1, PORT2, "1"));
    final String client2Host = getServerHostName();
    clientVM2.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client2Host,
        PORT1, PORT2, "1"));

    registerInterestListAll();
    clientVM1.invoke(HARQueueNewImplDistributedTest::registerInterestList);
    clientVM2.invoke(HARQueueNewImplDistributedTest::registerInterestList);

    serverVM0.invoke((SerializableRunnableIF) HARQueueNewImplDistributedTest::createEntries);

    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.verifyRegionSize(5, 5));

    serverVM0.invoke(ConflationDistributedTestHelper::unsetIsSlowStart);
    serverVM0.invoke(() -> HARQueueNewImplDistributedTest
        .waitTillMessagesAreDispatched(PORT1));

    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.verifyRegionSize(5, 0));
  }

  /**
   * This test verifies that the processing of the QRM messages results in decrementing the
   * reference count of corresponding HAEventWrapper instances, correctly.
   */
  @Test
  public void testRefCountForQRM() throws Exception {
    serverVM0.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("30000"));

    createClientCache(getServerHostName(), PORT1, PORT2,
        "1");
    final String client1Host = getServerHostName();
    clientVM1.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client1Host,
        PORT1, PORT2, "1"));
    final String client2Host = getServerHostName();
    clientVM2.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client2Host,
        PORT1, PORT2, "1"));

    registerInterestListAll();
    clientVM1.invoke(HARQueueNewImplDistributedTest::registerInterestList);
    clientVM2.invoke(HARQueueNewImplDistributedTest::registerInterestList);

    serverVM1.invoke(HARQueueNewImplDistributedTest::stopServer);

    serverVM0.invoke((SerializableRunnableIF) HARQueueNewImplDistributedTest::createEntries);

    serverVM1.invoke(HARQueueNewImplDistributedTest::startServer);

    serverVM1.invoke(() -> HARQueueNewImplDistributedTest.verifyRegionSize(5, 5));

    serverVM0.invoke(ConflationDistributedTestHelper::unsetIsSlowStart);

    serverVM1.invoke(() -> HARQueueNewImplDistributedTest.verifyRegionSize(5, 0));
  }

  /**
   * This test verifies that the destruction of a ha-region (caused by proxy/client disconnect),
   * causes the reference count of all HAEventWrapper instances belonging to the ha-region-queue to
   * be decremented by one, and makes it visible to the client-messages-region.
   */
  @Test
  public void testRefCountForDestroy() throws Exception {
    // slow start for dispatcher
    serverVM0.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("30000"));
    serverVM1.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("30000"));

    createClientCache(getServerHostName(), PORT1, PORT2,
        "1");
    final String client1Host = getServerHostName();
    clientVM1.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client1Host,
        PORT1, PORT2, "1"));
    final String client2Host = getServerHostName();
    clientVM2.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client2Host,
        PORT1, PORT2, "1"));

    registerInterestListAll();
    clientVM1.invoke(HARQueueNewImplDistributedTest::registerInterestList);
    clientVM2.invoke(HARQueueNewImplDistributedTest::registerInterestList);
    // 1. stop the second server
    serverVM1.invoke(HARQueueNewImplDistributedTest::stopServer);

    serverVM0.invoke((SerializableRunnableIF) HARQueueNewImplDistributedTest::createEntries);
    // 3. start the second server.
    serverVM1.invoke(HARQueueNewImplDistributedTest::startServer);
    Thread.sleep(3000);

    clientVM1.invoke(HARQueueNewImplDistributedTest::closeCache);

    Thread.sleep(1000);
    serverVM0.invoke(HARQueueNewImplDistributedTest::updateMap1);
    serverVM1.invoke(HARQueueNewImplDistributedTest::updateMap1);

    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.verifyQueueData(
        PORT1));
    serverVM1.invoke(() -> HARQueueNewImplDistributedTest.verifyQueueData(
        PORT2));

    clientVM2.invoke(HARQueueNewImplDistributedTest::closeCache);

    serverVM0.invoke(HARQueueNewImplDistributedTest::updateMap2);
    serverVM1.invoke(HARQueueNewImplDistributedTest::updateMap2);

    Thread.sleep(1000);
    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.verifyQueueData(
        PORT1));
    serverVM1.invoke(() -> HARQueueNewImplDistributedTest.verifyQueueData(
        PORT2));
  }

  /**
   * Addresses the bug 39179. If a clientUpdateMessage is dispatched to the client while its GII was
   * under way, then it should not be put into the HARegionQueue of a client at receiving server
   * side.
   */
  @Test
  public void testConcurrentGIIAndDispatch() throws Exception {
    serverVM0.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("40000"));
    serverVM1.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("40000"));

    createClientCache(getServerHostName(), PORT1, PORT2,
        "1");
    final String client1Host = getServerHostName();
    clientVM1.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client1Host,
        PORT1, PORT2, "1"));
    final String client2Host = getServerHostName();
    clientVM2.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client2Host,
        PORT1, PORT2, "1"));

    registerInterestListAll();
    clientVM1.invoke(HARQueueNewImplDistributedTest::registerInterestListAll);
    clientVM2.invoke(HARQueueNewImplDistributedTest::registerInterestListAll);
    // 1. stop the second server
    serverVM1.invoke(HARQueueNewImplDistributedTest::stopServer);

    serverVM0.invoke((SerializableRunnableIF) HARQueueNewImplDistributedTest::createEntries);
    serverVM0.invoke(() -> HARQueueNewImplDistributedTest
        .makeValuesOfSomeKeysNullInClientMsgsRegion(PORT1, new String[] {"k1", "k3"}));
    // 3. start the second server.
    serverVM1.invoke(HARQueueNewImplDistributedTest::startServer);
    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.verifyRegionSize(5, 3));

    serverVM1.invoke(() -> HARQueueNewImplDistributedTest.verifyNullValuesInCMR(
        PORT2, new String[] {"k1", "k3"}));
    serverVM1.invoke(() -> HARQueueNewImplDistributedTest.verifyRegionSize(5, 3));

    serverVM0.invoke(() -> HARQueueNewImplDistributedTest
        .populateValuesOfSomeKeysInClientMsgsRegion(PORT1, new String[] {"k1", "k3"}));

    serverVM0.invoke(ConflationDistributedTestHelper::unsetIsSlowStart);
    serverVM1.invoke(ConflationDistributedTestHelper::unsetIsSlowStart);
  }

  /**
   * This test verifies that when two BridgeServerImpl instances are created in a single VM, they do
   * share the client-messages-region.
   */
  @Test
  public void testTwoBridgeServersInOneVMDoShareCMR() throws Exception {
    // slow start for dispatcher
    serverVM0.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("30000"));

    Integer port3 = serverVM0
        .invoke(() -> HARQueueNewImplDistributedTest.createOneMoreBridgeServer(Boolean.TRUE));

    createClientCache(getServerHostName(), PORT1, port3, "0");
    final String client1Host = getServerHostName();
    clientVM1.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client1Host,
        PORT1, PORT2, "1"));
    final String client2Host = getServerHostName();
    clientVM2.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client2Host,
        PORT1, PORT2, "1"));

    registerInterestListAll();
    clientVM1.invoke(HARQueueNewImplDistributedTest::registerInterestList);
    clientVM2.invoke(HARQueueNewImplDistributedTest::registerInterestList);

    serverVM0.invoke((SerializableRunnableIF) HARQueueNewImplDistributedTest::createEntries);

    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.verifyRegionSize(5, 5));
    serverVM0.invoke(
        () -> HARQueueNewImplDistributedTest.verifyRegionSize(5, 5));
  }

  /**
   * This test verifies that two clients, connected to two cache servers with different
   * notifyBySubscription values, on a single VM, receive updates/invalidates depending upon their
   * notifyBySubscription value.
   */
  @Test
  public void testUpdatesWithTwoBridgeServersInOneVM() throws Exception {
    Integer port3 = serverVM0
        .invoke(() -> HARQueueNewImplDistributedTest.createOneMoreBridgeServer(Boolean.FALSE));

    createClientCache(getServerHostName(), PORT1, PORT2,
        "1", Boolean.TRUE);
    final String client1Host = getServerHostName();
    clientVM1.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client1Host, port3,
        PORT2, "1", Boolean.TRUE));

    registerInterestListAll();
    clientVM1.invoke(HARQueueNewImplDistributedTest::registerInterestListAll);

    clientVM1.invoke((SerializableRunnableIF) HARQueueNewImplDistributedTest::createEntries);
    serverVM0.invoke(HARQueueNewImplDistributedTest::putEntries);

    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.waitTillMessagesAreDispatched(PORT1));
    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.waitTillMessagesAreDispatched(port3));

    // expect updates
    verifyUpdatesReceived();
    // expect invalidates
    clientVM1.invoke(HARQueueNewImplDistributedTest::verifyUpdatesReceived);
  }

  /**
   * This test verifies that the HAEventWrapper instances present in the client-messages-region give
   * up the references to their respective ClientUpdateMessageImpl instances.
   */
  @Test
  public void testHAEventWrapperDoesNotHoldCUMOnceInsideCMR() throws Exception {
    // slow start for dispatcher
    serverVM0.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("30000"));

    createClientCache(getServerHostName(), PORT1, PORT2,
        "1");
    final String client1Host = getServerHostName();
    clientVM1.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client1Host,
        PORT1, PORT2, "1"));
    final String client2Host = getServerHostName();
    clientVM2.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client2Host,
        PORT1, PORT2, "1"));

    registerInterestListAll();
    clientVM1.invoke(HARQueueNewImplDistributedTest::registerInterestList);
    clientVM2.invoke(HARQueueNewImplDistributedTest::registerInterestList);

    serverVM1.invoke(HARQueueNewImplDistributedTest::stopServer);

    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.createEntries(1000L));

    serverVM1.invoke(HARQueueNewImplDistributedTest::startServer);
    Thread.sleep(2000);

    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.verifyNullCUMReference(PORT1));
    serverVM1.invoke(() -> HARQueueNewImplDistributedTest.verifyNullCUMReference(PORT2));
  }

  /**
   * This test verifies that client-messages-regions are not created for the cache servers who have
   * eviction policy as 'none'. Instead, such cache servers will have simple HashMap structures.
   * Also, it verifies that such a structure (referred to as haContainer, in general) is destroyed
   * when its cache server is stopped.
   */
  @Test
  public void testCMRNotCreatedForNoneEvictionPolicy() throws Exception {
    serverVM0.invoke(HARQueueNewImplDistributedTest::closeCache);
    serverVM1.invoke(HARQueueNewImplDistributedTest::closeCache);
    Thread.sleep(2000);
    PORT1 = serverVM0.invoke(
        () -> HARQueueNewImplDistributedTest
            .createServerCache(HARegionQueue.HA_EVICTION_POLICY_NONE));
    PORT2 = serverVM1.invoke(
        () -> HARQueueNewImplDistributedTest
            .createServerCache(HARegionQueue.HA_EVICTION_POLICY_NONE));
    Boolean isRegion = Boolean.FALSE;
    // slow start for dispatcher
    serverVM0.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("30000"));

    createClientCache(getServerHostName(), PORT1, PORT2,
        "1");
    final String client1Host = getServerHostName();
    clientVM1.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client1Host,
        PORT1, PORT2, "1"));
    final String client2Host = getServerHostName();
    clientVM2.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client2Host,
        PORT1, PORT2, "1"));

    registerInterestListAll();
    clientVM1.invoke(HARQueueNewImplDistributedTest::registerInterestList);
    clientVM2.invoke(HARQueueNewImplDistributedTest::registerInterestList);

    serverVM0
        .invoke(() -> HARQueueNewImplDistributedTest.verifyHaContainerType(isRegion, PORT1));
    serverVM1
        .invoke(() -> HARQueueNewImplDistributedTest.verifyHaContainerType(isRegion, PORT2));

    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.stopOneBridgeServer(PORT1));
    serverVM1.invoke(() -> HARQueueNewImplDistributedTest.stopOneBridgeServer(PORT2));

    serverVM0.invoke(
        () -> HARQueueNewImplDistributedTest.verifyHaContainerDestroyed(isRegion, PORT1));
    serverVM1.invoke(
        () -> HARQueueNewImplDistributedTest.verifyHaContainerDestroyed(isRegion, PORT2));
  }

  /**
   * This test verifies that client-messages-regions are created for the cache servers who have
   * eviction policy either as 'mem' or as 'entry'. Also, it verifies that such a
   * client-messages-region is destroyed when its cache server is stopped.
   */
  @Test
  public void testCMRCreatedForMemOrEntryEvictionPolicy() throws Exception {
    Boolean isRegion = Boolean.TRUE;
    // slow start for dispatcher
    serverVM0.invoke(() -> ConflationDistributedTestHelper.setIsSlowStart("30000"));

    createClientCache(getServerHostName(), PORT1, PORT2,
        "1");
    final String client1Host = getServerHostName();
    clientVM1.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client1Host,
        PORT1, PORT2, "1"));
    final String client2Host = getServerHostName();
    clientVM2.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client2Host,
        PORT1, PORT2, "1"));

    registerInterestListAll();
    clientVM1.invoke(HARQueueNewImplDistributedTest::registerInterestList);
    clientVM2.invoke(HARQueueNewImplDistributedTest::registerInterestList);

    serverVM0
        .invoke(() -> HARQueueNewImplDistributedTest.verifyHaContainerType(isRegion, PORT1));
    serverVM1
        .invoke(() -> HARQueueNewImplDistributedTest.verifyHaContainerType(isRegion, PORT2));

    serverVM0.invoke(() -> HARQueueNewImplDistributedTest.stopOneBridgeServer(PORT1));
    serverVM1.invoke(() -> HARQueueNewImplDistributedTest.stopOneBridgeServer(PORT2));

    serverVM0.invoke(
        () -> HARQueueNewImplDistributedTest.verifyHaContainerDestroyed(isRegion, PORT1));
    serverVM1.invoke(
        () -> HARQueueNewImplDistributedTest.verifyHaContainerDestroyed(isRegion, PORT2));
  }

  /**
   * This test verifies that the Cache.rootRegions() method does not return the
   * client-messages-region of any of the cache's attached cache servers.
   */
  @Test
  public void testCMRNotReturnedByRootRegionsMethod() throws Exception {

    createClientCache(getServerHostName(), PORT1, PORT2,
        "1");
    final String client1Host = getServerHostName();
    clientVM1.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client1Host,
        PORT1, PORT2, "1"));
    final String client2Host = getServerHostName();
    clientVM2.invoke(() -> HARQueueNewImplDistributedTest.createClientCache(client2Host,
        PORT1, PORT2, "1"));

    registerInterestListAll();
    clientVM1.invoke(HARQueueNewImplDistributedTest::registerInterestList);
    clientVM2.invoke(HARQueueNewImplDistributedTest::registerInterestList);

    serverVM0.invoke((SerializableRunnableIF) HARQueueNewImplDistributedTest::createEntries);

    serverVM0.invoke(
        () -> HARQueueNewImplDistributedTest.verifyRootRegionsDoesNotReturnCMR(PORT1));
    serverVM1.invoke(
        () -> HARQueueNewImplDistributedTest.verifyRootRegionsDoesNotReturnCMR(PORT2));
  }

  /**
   * This test verifies that the memory footprint of the ha region queues is less when ha-overflow
   * is enabled (with an appropriate value of haCapacity) compared to when it is disabled, for the
   * same amount of data feed.
   */
  @Ignore("TODO")
  @Test
  public void testMemoryFootprintOfHARegionQueuesWithAndWithoutOverflow() throws Exception {
    serverVM0.invoke(HARQueueNewImplDistributedTest::closeCache);
