/*
 * 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.SEPARATOR;
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.test.dunit.Assert.assertEquals;
import static org.apache.geode.test.dunit.Assert.assertNotNull;
import static org.apache.geode.test.dunit.Assert.assertTrue;
import static org.apache.geode.test.dunit.Assert.fail;

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

import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.LogWriter;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.ExpirationAttributes;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.query.CqAttributes;
import org.apache.geode.cache.query.CqAttributesFactory;
import org.apache.geode.cache.query.CqListener;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.cq.dunit.CqQueryTestListener;
import org.apache.geode.cache.query.data.Portfolio;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache30.CertifiableTestCacheListener;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.internal.cache.CacheServerImpl;
import org.apache.geode.internal.cache.InternalRegionArguments;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.tier.sockets.CacheServerTestUtil;
import org.apache.geode.internal.cache.tier.sockets.ClientUpdateMessageImpl;
import org.apache.geode.internal.cache.tier.sockets.ConflationDUnitTestHelper;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
import org.apache.geode.test.junit.categories.ClientSubscriptionTest;

/**
 * @since GemFire 5.7
 */
@Category({ClientSubscriptionTest.class})
public class CQListGIIDUnitTest extends JUnit4DistributedTestCase {

  private static final int CREATE = 0;
  private static final int UPDATE = 1;
  private static final int DESTROY = 2;
  private static final int INVALIDATE = 3;
  private static final int CLOSE = 4;
  private static final int REGION_CLEAR = 5;
  private static final int REGION_INVALIDATE = 6;

  protected static Cache cache = null;

  protected static VM serverVM0 = null;
  private static VM serverVM1 = null;
  protected static VM clientVM1 = null;
  protected static VM clientVM2 = null;

  private int PORT1;
  private int PORT2;

  private static final String regionName = "CQListGIIDUnitTest";

  private static final Map map = new HashMap();

  private static LogWriter logger = null;

  public static final String[] regions = new String[] {"regionA", "regionB"};

  public static final String KEY = "key-";

  public String[] cqs = new String[] {
      // 0 - Test for ">"
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0] + " p where p.ID > 0",

      // 1 - Test for "=" and "and".
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0]
          + " p where p.ID = 2 and p.status='active'",

      // 2 - Test for "<" and "and".
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[1]
          + " p where p.ID < 5 and p.status='active'",

      // FOLLOWING CQS ARE NOT TESTED WITH VALUES; THEY ARE USED TO TEST PARSING
      // LOGIC WITHIN CQ.
      // 3
      "SELECT * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0] + " ;",
      // 4
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0],
      // 5
      "import org.apache.geode.cache.\"query\".data.Portfolio; " + "SELECT ALL * FROM " + SEPARATOR
          + "root" + SEPARATOR
          + regions[0] + " TYPE Portfolio",
      // 6
      "import org.apache.geode.cache.\"query\".data.Portfolio; " + "SELECT ALL * FROM " + SEPARATOR
          + "root" + SEPARATOR
          + regions[0] + " p TYPE Portfolio",
      // 7
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[1]
          + " p where p.ID < 5 and p.status='active';",
      // 8
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0] + "  ;",
      // 9
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0]
          + " p where p.description = NULL",
      // 10
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0]
          + " p where p.ID > 0 and p.status='active'",};

  @Override
  public final void postSetUp() throws Exception {
    final Host host = Host.getHost(0);
    serverVM0 = host.getVM(0);
    serverVM1 = host.getVM(1);
    clientVM1 = host.getVM(2);
    clientVM2 = host.getVM(3);

    PORT1 = ((Integer) serverVM0.invoke(
        () -> CQListGIIDUnitTest.createServerCache(HARegionQueue.HA_EVICTION_POLICY_MEMORY)))
            .intValue();
    PORT2 = ((Integer) serverVM1
        .invoke(() -> CQListGIIDUnitTest.createServerCache(HARegionQueue.HA_EVICTION_POLICY_ENTRY)))
            .intValue();
  }

  @Override
  public final void preTearDown() throws Exception {
    serverVM0.invoke(() -> ConflationDUnitTestHelper.unsetIsSlowStart());
    serverVM1.invoke(() -> ConflationDUnitTestHelper.unsetIsSlowStart());
    closeCache();
    clientVM1.invoke(() -> CQListGIIDUnitTest.closeCache());
    clientVM2.invoke(() -> CQListGIIDUnitTest.closeCache());
    // then close the servers
    serverVM0.invoke(() -> CQListGIIDUnitTest.closeCache());
    serverVM1.invoke(() -> CQListGIIDUnitTest.closeCache());
    disconnectAllFromDS();
  }

  private void createCache(Properties props) throws Exception {
    DistributedSystem ds = getSystem(props);
    ds.disconnect();
    ds = getSystem(props);
    assertNotNull(ds);
    cache = CacheFactory.create(ds);
    assertNotNull(cache);
  }

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

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

  public static Integer createServerCache(String ePolicy, Integer cap) throws Exception {
    new CQListGIIDUnitTest().createCache(new Properties());
    AttributesFactory factory = new AttributesFactory();
    factory.setScope(Scope.DISTRIBUTED_ACK);
    factory.setDataPolicy(DataPolicy.REPLICATE);
    RegionAttributes attrs = factory.create();
    createRegion(regions[0], "root", attrs);
    createRegion(regions[1], "root", attrs);
    Thread.sleep(2000);
    logger = cache.getLogger();

    int port = getRandomAvailableTCPPort();
    CacheServer server1 = cache.addCacheServer();
    server1.setPort(port);
    server1.setNotifyBySubscription(true);
    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.intValue());
      // specify diskstore for this server
      server1.getClientSubscriptionConfig()
          .setDiskStoreName(dsf.setDiskDirs(dirs1).create("bsi").getName());
    }
    server1.start();
    Thread.sleep(2000);
    return Integer.valueOf(server1.getPort());
  }

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

  public static Region createRegion(String name, String rootName, RegionAttributes attrs)
      throws CacheException {
    Region root = cache.getRegion(rootName);
    if (root == null) {
      // don't put listeners on root region
      RegionAttributes rootAttrs = attrs;
      AttributesFactory fac = new AttributesFactory(attrs);
      ExpirationAttributes expiration = ExpirationAttributes.DEFAULT;

      // fac.setCacheListener(null);
      fac.setCacheLoader(null);
      fac.setCacheWriter(null);
      fac.setPoolName(null);
      fac.setPartitionAttributes(null);
      fac.setRegionTimeToLive(expiration);
      fac.setEntryTimeToLive(expiration);
      fac.setRegionIdleTimeout(expiration);
      fac.setEntryIdleTimeout(expiration);
      rootAttrs = fac.create();
      root = cache.createRegion(rootName, rootAttrs);
    }

    return createSubregion(root, name, attrs, null);
  }

  /**
   * A helper for creating a subregion, potentially using a package protected method to do so.
   *
   * @param root the parent region
   * @param name the name of the subregion to create
   * @param attrs the attributes used to create the subregion
   * @param internalArgs if not null, then use the package protected creation mechanism
   * @return the subregion whose parent is the provided root
   * @see Region#createSubregion(String, RegionAttributes)
   * @see LocalRegion#createSubregion(String, RegionAttributes, InternalRegionArguments)
   */
  public static Region createSubregion(Region root, String name, RegionAttributes attrs,
      final InternalRegionArguments internalArgs) throws CacheException {
    Region value = null;
    if (internalArgs == null) {
      value = root.createSubregion(name, attrs);
    } else {
      try {
        LocalRegion lr = (LocalRegion) root;
        value = lr.createSubregion(name, attrs, internalArgs);
      } catch (IOException e) {
        fail("unexpected exception", e);
      } catch (ClassNotFoundException e) {
        fail("unexpected exception", e);
      }
    }
    return value;
  }

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

  public static void createClientCache(Integer port1, Integer port2, String rLevel,
      Boolean addListener) throws Exception {
    createClientCache(port1, port2, Integer.valueOf(-1), rLevel, addListener);
  }

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

  public static void destroyClientPool() {
    cache.getRegion("root").getSubregion(regions[0]).close();
    cache.getRegion("root").getSubregion(regions[1]).close();
    PoolManager.find("clientPool").destroy();
  }

  public static void createClientCache(Integer port1, Integer port2, Integer port3, String rLevel,
      Boolean addListener) throws Exception {
    CacheServerTestUtil.disableShufflingOfEndpoints();
    String host = NetworkUtils.getIPLiteral();

    Properties props = new Properties();
    props.setProperty(MCAST_PORT, "0");
    props.setProperty(LOCATORS, "");
    new CQListGIIDUnitTest().createCache(props);

    PoolFactory pf = PoolManager.createFactory();
    int endPointCount = 1;
    pf.addServer(host, port1);
    if (port2.intValue() != -1) {
      pf.addServer(host, port2);
      endPointCount++;
    }
    if (port3.intValue() != -1) {
      pf.addServer(host, port3);
      endPointCount++;
    }
    pf.setRetryAttempts(5);
    pf.setReadTimeout(2500);
    pf.setSocketBufferSize(32768);
    pf.setPingInterval(1000);
    pf.setMinConnections(endPointCount * 2);
    pf.setSubscriptionRedundancy(Integer.parseInt(rLevel));
    pf.setSubscriptionEnabled(true).create("clientPool");

    try {
      cache.getQueryService();
    } catch (Exception cqe) {
      Assert.fail("Failed to getCQService.", cqe);
    }

    AttributesFactory factory = new AttributesFactory();
    factory.setScope(Scope.LOCAL);
    factory.setPoolName("clientPool");

    RegionAttributes attrs = factory.create();
    createRegion(regions[0], "root", attrs);
    createRegion(regions[1], "root", attrs);
    logger = cache.getLogger();
  }

  /* Register CQs */
  public static void createCQ(String cqName, String queryStr) {
    org.apache.geode.test.dunit.LogWriterUtils.getLogWriter().info("### Create CQ. ###" + cqName);
    // Get CQ Service.
    QueryService cqService = null;
    try {
      cqService = cache.getQueryService();
    } catch (Exception cqe) {
      Assert.fail("Failed to getCQService.", cqe);
    }
    // Create CQ Attributes.
    CqAttributesFactory cqf = new CqAttributesFactory();
    CqListener[] cqListeners =
        {new CqQueryTestListener(org.apache.geode.test.dunit.LogWriterUtils.getLogWriter())};
    ((CqQueryTestListener) cqListeners[0]).cqName = cqName;

    cqf.initCqListeners(cqListeners);
    CqAttributes cqa = cqf.create();

    // Create CQ.
    try {
      CqQuery cq1 = cqService.newCq(cqName, queryStr, cqa);
      assertTrue("newCq() state mismatch", cq1.getState().isStopped());
    } catch (Exception ex) {
      fail("Failed to create CQ " + cqName + " . ", ex);
    }
  }

  public static void executeCQ(String cqName, Boolean initialResults) {
    org.apache.geode.test.dunit.LogWriterUtils.getLogWriter()
        .info("### DEBUG EXECUTE CQ START ####");
    // Get CQ Service.
    QueryService cqService = null;
    CqQuery cq1 = null;
    cqService = cache.getQueryService();

    // Get CqQuery object.
    try {
      cq1 = cqService.getCq(cqName);
      if (cq1 == null) {
        org.apache.geode.test.dunit.LogWriterUtils.getLogWriter()
            .info("Failed to get CqQuery object for CQ name: " + cqName);
        Assert.fail("Failed to get CQ " + cqName, new Exception("Failed to get CQ " + cqName));
      } else {
        org.apache.geode.test.dunit.LogWriterUtils.getLogWriter()
            .info("Obtained CQ, CQ name: " + cq1.getName());
        assertTrue("newCq() state mismatch", cq1.getState().isStopped());
      }
    } catch (Exception ex) {
      fail("Failed to execute  CQ " + cqName, ex);
    }

    if (initialResults.booleanValue()) {
      SelectResults cqResults = null;

      try {
        cqResults = cq1.executeWithInitialResults();
      } catch (Exception ex) {
        fail("Failed to execute  CQ " + cqName, ex);
      }
      org.apache.geode.test.dunit.LogWriterUtils.getLogWriter()
          .info("initial result size = " + cqResults.size());
      assertTrue("executeWithInitialResults() state mismatch", cq1.getState().isRunning());
    } else {

      try {
        cq1.execute();
      } catch (Exception ex) {
        fail("Failed to execute  CQ " + cqName, ex);
      }
      assertTrue("execute() state mismatch", cq1.getState().isRunning());
    }
  }

  public static void registerInterestListCQ(String regionName, int keySize) {
    // Get CQ Service.
    Region region = null;
    try {
      region = cache.getRegion("root").getSubregion(regionName);
      region.getAttributesMutator().addCacheListener(new CertifiableTestCacheListener());
    } catch (Exception e) {
      fail("Failed to get Region.", e);
    }

    try {
      List list = new ArrayList();
      for (int i = 1; i <= keySize; i++) {
        list.add(KEY + i);
      }
      region.registerInterest(list);
    } catch (Exception ex) {
      fail("Failed to Register InterestList", ex);
    }
  }

  public static void waitForCreated(String cqName, String key) {
    waitForEvent(0, cqName, key);
  }

  public static void waitForEvent(int event, String cqName, String key) {
    // Get CQ Service.
    QueryService cqService = null;
    try {
      cqService = cache.getQueryService();
    } catch (Exception cqe) {
      Assert.fail("Failed to getCQService.", cqe);
    }

    CqQuery cQuery = cqService.getCq(cqName);
    if (cQuery == null) {
      Assert.fail("Failed to get CqQuery for CQ : " + cqName,
          new Exception("Failed to get CqQuery for CQ : " + cqName));
    }

    CqAttributes cqAttr = cQuery.getCqAttributes();
    CqListener[] cqListener = cqAttr.getCqListeners();
    CqQueryTestListener listener = (CqQueryTestListener) cqListener[0];

    switch (event) {
      case CREATE:
        listener.waitForCreated(key);
        break;

      case UPDATE:
        listener.waitForUpdated(key);
        break;

      case DESTROY:
        listener.waitForDestroyed(key);
        break;

      case INVALIDATE:
        listener.waitForInvalidated(key);
        break;

      case CLOSE:
        listener.waitForClose();
        break;

      case REGION_CLEAR:
        listener.waitForRegionClear();
        break;

      case REGION_INVALIDATE:
        listener.waitForRegionInvalidate();
        break;
    }
  }

  public static void registerInterestListAll() {
    try {
      Region r = cache.getRegion(SEPARATOR + regionName);
      assertNotNull(r);
      r.registerInterest("ALL_KEYS");
    } catch (Exception ex) {
      Assert.fail("failed in registerInterestListAll", ex);
    }
  }

  public static void registerInterestList() {
    try {
      Region r = cache.getRegion(SEPARATOR + regionName);
      assertNotNull(r);
      r.registerInterest("k1");
      r.registerInterest("k3");
      r.registerInterest("k5");
    } catch (Exception ex) {
      Assert.fail("failed while registering keys", ex);
    }
  }

  public static void putEntries(String rName, Integer num) {
    try {
      Region r = cache.getRegion("root").getSubregion(rName);
      assertNotNull(r);
      for (int i = 0; i < num.longValue(); i++) {
        r.put(KEY + i, new Portfolio(i + 1));
      }
      org.apache.geode.test.dunit.LogWriterUtils.getLogWriter()
          .info("### Number of Entries in Region " + rName + ": " + r.keySet().size());
    } catch (Exception ex) {
      Assert.fail("failed in putEntries()", ex);
    }
  }

  @Ignore("TODO: test is disabled")
  @Test
  public void testSpecificClientCQIsGIIedPart1() throws Exception {
    Integer size = Integer.valueOf(10);
    // slow start for dispatcher
    serverVM0.invoke(() -> ConflationDUnitTestHelper.setIsSlowStart("30000"));
    serverVM1.invoke(() -> ConflationDUnitTestHelper.setIsSlowStart("30000"));

    // createClientCache(Integer.valueOf(PORT1), Integer.valueOf(PORT2), "1");
    clientVM1.invoke(() -> CQListGIIDUnitTest.createClientCache(Integer.valueOf(PORT1),
        Integer.valueOf(PORT2), "1"));
    clientVM2.invoke(() -> CQListGIIDUnitTest.createClientCache(Integer.valueOf(PORT1),
        Integer.valueOf(PORT2), "0"));

    clientVM1.invoke(() -> CQListGIIDUnitTest.createCQ("testSpecificClientCQIsGIIed_0", cqs[0]));
    clientVM1
        .invoke(() -> CQListGIIDUnitTest.executeCQ("testSpecificClientCQIsGIIed_0", Boolean.FALSE));
    clientVM2.invoke(() -> CQListGIIDUnitTest.createCQ("testSpecificClientCQIsGIIed_0", cqs[0]));
    clientVM2
        .invoke(() -> CQListGIIDUnitTest.executeCQ("testSpecificClientCQIsGIIed_0", Boolean.FALSE));

    serverVM1.invoke(() -> CQListGIIDUnitTest.stopServer());

    serverVM0.invoke(() -> CQListGIIDUnitTest.putEntries(regions[0], size));

    serverVM1.invoke(() -> CQListGIIDUnitTest.startServer());
    Thread.sleep(3000); // TODO: Find a better 'n reliable alternative

    serverVM0.invoke(() -> CQListGIIDUnitTest.VerifyCUMCQList(size, Integer.valueOf(2)));
    serverVM1.invoke(() -> CQListGIIDUnitTest.VerifyCUMCQList(size, Integer.valueOf(1)));
    serverVM0.invoke(() -> ConflationDUnitTestHelper.unsetIsSlowStart());
    serverVM1.invoke(() -> ConflationDUnitTestHelper.unsetIsSlowStart());
  }

  /**
   * This test asserts that cq list of a client for an event is not lost if that client's queue has
   * been GII'ed to a server where that event already existed.
   */
  @Test
  public void testClientCQNotLostAtGIIReceiver() throws Exception {
    Integer size = Integer.valueOf(10);
    VM serverVM2 = clientVM2;

    int port3 = ((Integer) serverVM2.invoke(
        () -> CQListGIIDUnitTest.createServerCache(HARegionQueue.HA_EVICTION_POLICY_MEMORY)))
            .intValue();

    // slow start for dispatcher
    serverVM0.invoke(() -> ConflationDUnitTestHelper.setIsSlowStart("45000"));

    // createClientCache(Integer.valueOf(PORT1), Integer.valueOf(PORT2), "1");
    createClientCache(Integer.valueOf(PORT1), Integer.valueOf(PORT2), Integer.valueOf(port3), "1");
    try {
      clientVM1.invoke(() -> CQListGIIDUnitTest.createClientCache(Integer.valueOf(PORT1),
          Integer.valueOf(port3), Integer.valueOf(PORT2), "1"));
      try {
        createCQ("testSpecificClientCQIsGIIed_0", cqs[0]);
        executeCQ("testSpecificClientCQIsGIIed_0", Boolean.FALSE);
        clientVM1
            .invoke(() -> CQListGIIDUnitTest.createCQ("testSpecificClientCQIsGIIed_0", cqs[0]));
        clientVM1.invoke(
            () -> CQListGIIDUnitTest.executeCQ("testSpecificClientCQIsGIIed_0", Boolean.FALSE));

        serverVM0.invoke(() -> CQListGIIDUnitTest.putEntries(regions[0], size));

        serverVM1.invoke(() -> CQListGIIDUnitTest.VerifyCUMCQList(size, Integer.valueOf(1)));

        serverVM2.invoke(() -> CQListGIIDUnitTest.stopServer());
        Thread.sleep(3000); // TODO: Find a better 'n reliable alternative

        serverVM0.invoke(() -> CQListGIIDUnitTest.VerifyCUMCQList(size, Integer.valueOf(2)));
        serverVM1.invoke(() -> CQListGIIDUnitTest.VerifyCUMCQList(size, Integer.valueOf(2)));
      } finally {
        clientVM1.invoke(() -> CQListGIIDUnitTest.destroyClientPool());
      }

    } finally {
      destroyClientPool();
    }
  }

  public static void VerifyCUMCQList(Integer numOfKeys, Integer numOfClients) {
    try {
      Iterator iter = cache.getCacheServers().iterator();
      if (iter.hasNext()) {
        CacheServerImpl server = (CacheServerImpl) iter.next();
        Map haContainer = server.getAcceptor().getCacheClientNotifier().getHaContainer();
        Object[] keys = haContainer.keySet().toArray();
        logger.fine("### numOfKeys :" + numOfKeys.intValue() + " keys.length : " + keys.length
            + " haContainer size : " + haContainer.size());
        assertEquals(numOfKeys.intValue(), keys.length);
        for (int i = 0; i < numOfKeys.intValue(); i++) {
          logger.fine("i=: " + i);
          ClientUpdateMessageImpl cum = (ClientUpdateMessageImpl) haContainer.get(keys[i]);
          assertNotNull(cum);
          assertNotNull(cum.getClientCqs());
          assertEquals(
              "This test may fail if the image provider gets an ack from client before providing image",
              numOfClients.intValue(), cum.getClientCqs().size());
        }
      }
    } catch (Exception e) {
      Assert.fail("failed in VerifyCUMCQList()" + e, e);
    }
  }

  private static void stopOneBridgeServer(Integer port) {
    try {
      Iterator iter = cache.getCacheServers().iterator();
      if (iter.hasNext()) {
        CacheServer server = (CacheServer) iter.next();
        if (server.getPort() == port.intValue()) {
          server.stop();
        }
      }
    } catch (Exception e) {
      fail("failed in stopOneBridgeServer()", e);
    }
  }

  public static void stopServer() {
    try {
      Iterator iter = cache.getCacheServers().iterator();
      if (iter.hasNext()) {
        CacheServer server = (CacheServer) iter.next();
        server.stop();
      }
    } catch (Exception e) {
      fail("failed in stopServer()", e);
    }
  }

  public static void startServer() {
    try {
      Iterator iter = cache.getCacheServers().iterator();
      if (iter.hasNext()) {
        CacheServer server = (CacheServer) iter.next();
        server.start();
      }
    } catch (Exception e) {
      fail("failed in startServer()", e);
    }
  }

  public static void waitTillMessagesAreDispatched(Integer port, Long waitLimit) {
    try {
      boolean dispatched = false;
      Map haContainer = null;
      haContainer = cache.getRegion(
          SEPARATOR + CacheServerImpl.generateNameForClientMsgsRegion(port.intValue()));
      if (haContainer == null) {
        Object[] servers = cache.getCacheServers().toArray();
        for (int i = 0; i < servers.length; i++) {
          if (port.intValue() == ((CacheServerImpl) servers[i]).getPort()) {
            haContainer = ((CacheServerImpl) servers[i]).getAcceptor().getCacheClientNotifier()
                .getHaContainer();
            break;
          }
        }
      }
      long startTime = System.currentTimeMillis();
      while (waitLimit.longValue() > (System.currentTimeMillis() - startTime)) {
        if (haContainer.size() == 0) {
          dispatched = true;
          break;
        }
        try {
          Thread.sleep(50);
        } catch (InterruptedException ie) {
          fail("interrupted", ie);
        }
      }
      logger.fine("Exiting sleep, time elapsed was: " + (System.currentTimeMillis() - startTime));
      if (!dispatched) {
        throw new Exception(
            "Test tuning issue: The HARegionQueue is not fully drained, so cannot continue the test.");
      }
    } catch (Exception e) {
      fail("failed in waitTillMessagesAreDispatched()", e);
    }
  }

  public static void closeCache() {
    if (cache != null && !cache.isClosed()) {
      cache.close();
      cache.getDistributedSystem().disconnect();
    }
  }
}