/*
 * 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.cache.query.cq.dunit;

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.test.dunit.Assert.assertEquals;
import static org.apache.geode.test.dunit.Assert.assertFalse;
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.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;
import java.util.Set;

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

import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.MirrorType;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.PartitionAttributesFactory;
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.CqAttributesMutator;
import org.apache.geode.cache.query.CqExistsException;
import org.apache.geode.cache.query.CqListener;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.IndexType;
import org.apache.geode.cache.query.Query;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.Struct;
import org.apache.geode.cache.query.data.Portfolio;
import org.apache.geode.cache.query.internal.CqStateImpl;
import org.apache.geode.cache.query.internal.DefaultQueryService;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.cache30.CertifiableTestCacheListener;
import org.apache.geode.cache30.ClientServerTestCase;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.internal.cache.InitialImageOperation;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.CacheClientProxy;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.RMIException;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.dunit.WaitCriterion;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
import org.apache.geode.test.junit.categories.ClientSubscriptionTest;

/**
 * This class tests the ContinuousQuery mechanism in GemFire. It does so by creating a cache server
 * with a cache and a pre-defined region and a data loader. The client creates the same region and
 * attaches the connection pool.
 */
@Category({ClientSubscriptionTest.class})
public class CqQueryUsingPoolDUnitTest extends JUnit4CacheTestCase {

  /** The port on which the cache server was started in this VM */
  private static int bridgeServerPort;

  protected static int port = 0;
  protected static int port2 = 0;

  public static int noTest = -1;

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

  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;

  public static final String KEY = "key-";

  private static final String WAIT_PROPERTY = "CqQueryTest.maxWaitTime";

  private static final int WAIT_DEFAULT = (20 * 1000);

  public static final long MAX_TIME = Integer.getInteger(WAIT_PROPERTY, WAIT_DEFAULT);

  public final 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'",
      // 11 - Test for "No Alias"
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0] + " where ID > 0",};

  private final String[] invalidCQs = new String[] {
      // Test for ">"
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + "invalidRegion p where p.ID > 0"};

  @Override
  public final void postSetUp() throws Exception {
    // We're seeing this on the server when the client
    // disconnects.
    IgnoredException.addIgnoredException("Connection reset");
    IgnoredException.addIgnoredException("SocketTimeoutException");
    IgnoredException.addIgnoredException("ServerConnectivityException");
    IgnoredException.addIgnoredException("Socket Closed");
    IgnoredException.addIgnoredException("SocketException");
    // avoid IllegalStateException from HandShake by connecting all vms tor
    // system before creating connection pools
    getSystem();
    Invoke.invokeInEveryVM(new SerializableRunnable("getSystem") {
      @Override
      public void run() {
        getSystem();
      }
    });
    postSetUpCqQueryUsingPoolDUnitTest();
  }

  protected void postSetUpCqQueryUsingPoolDUnitTest() throws Exception {}

  /* Returns Cache Server Port */
  static int getCacheServerPort() {
    return bridgeServerPort;
  }

  /* Create Cache Server */
  public void createServer(VM server) {
    createServer(server, 0);
  }

  public void createServer(VM server, final int p) {
    createServer(server, p, false);
  }

  public void createServer(VM server, final int thePort, final boolean eviction) {
    MirrorType mirrorType = MirrorType.KEYS_VALUES;
    createServer(server, thePort, eviction, mirrorType);
  }

  public void createServer(VM server, final int port, final boolean eviction,
      final MirrorType mirrorType) {

    // Don't use ephemeral ports since those will probably fall out of the range as dictated by
    // AvailablePort.lowerBound and AvailablePort.upperBound. Usually that's OK unless the server
    // intends to be restarted using the same port as originally assigned ephemerally. There's
    // always a race when restarting servers, but this should reduce the problem just a bit.
    int actualPort = port == 0 ? AvailablePortHelper.getRandomAvailableTCPPort() : port;

    SerializableRunnable createServer = new CacheSerializableRunnable("Create Cache Server") {
      @Override
      public void run2() throws CacheException {
        LogWriterUtils.getLogWriter().info("### Create Cache Server. ###");
        AttributesFactory factory = new AttributesFactory();
        factory.setScope(Scope.DISTRIBUTED_ACK);
        factory.setMirrorType(mirrorType);

        // setting the eviction attributes.
        if (eviction) {
          EvictionAttributes evictAttrs =
              EvictionAttributes.createLRUEntryAttributes(100000, EvictionAction.OVERFLOW_TO_DISK);
          factory.setEvictionAttributes(evictAttrs);
        }

        for (final String region : regions) {
          createRegion(region, factory.createRegionAttributes());
        }

        try {
          startBridgeServer(actualPort, true);
        }

        catch (Exception ex) {
          Assert.fail("While starting CacheServer", ex);
        }
      }
    };

    server.invoke(createServer);
  }

  /**
   * Create a cache server with partitioned region.
   *
   * @param server VM where to create the cache server.
   * @param port cache server port.
   * @param isAccessor if true the under lying partitioned region will not host data on this vm.
   * @param redundantCopies number of redundant copies for the primary bucket.
   */
  public void createServerWithPR(VM server, final int port, final boolean isAccessor,
      final int redundantCopies) {
    SerializableRunnable createServer = new CacheSerializableRunnable("Create Cache Server") {
      @Override
      public void run2() throws CacheException {
        LogWriterUtils.getLogWriter().info("### Create Cache Server. ###");

        AttributesFactory attr = new AttributesFactory();
        PartitionAttributesFactory paf = new PartitionAttributesFactory();
        if (isAccessor) {
          paf.setLocalMaxMemory(0);
        }
        PartitionAttributes prAttr =
            paf.setTotalNumBuckets(197).setRedundantCopies(redundantCopies).create();
        attr.setPartitionAttributes(prAttr);

        assertFalse(getSystem().isLoner());
        for (final String region : regions) {
          Region r = createRegion(region, attr.create());
          LogWriterUtils.getLogWriter().info("Server created the region: " + r);
        }
        try {
          startBridgeServer(port, true);
        } catch (Exception ex) {
          Assert.fail("While starting CacheServer", ex);
        }
      }
    };

    server.invoke(createServer);
  }

  /* Close Cache Server */
  public void closeServer(VM server) {
    server.invoke(new SerializableRunnable("Close CacheServer") {
      @Override
      public void run() {
        LogWriterUtils.getLogWriter().info("### Close CacheServer. ###");
        stopBridgeServer(getCache());
      }
    });
  }

  /* Create Client */
  public void createClient(VM client, final int serverPort, final String serverHost) {
    int[] serverPorts = new int[] {serverPort};
    createClient(client, serverPorts, serverHost, null, null);
  }

  /* Create Client */
  public void createClient(VM client, final int[] serverPorts, final String serverHost,
      final String redundancyLevel, final String poolName) {
    SerializableRunnable createQService = new CacheSerializableRunnable("Create Client") {
      @Override
      public void run2() throws CacheException {
        LogWriterUtils.getLogWriter().info("### Create Client. ###");
        // Region region1 = null;
        // Initialize CQ Service.
        try {
          getCache().getQueryService();
          IgnoredException.addIgnoredException("java.net.ConnectException");
        } catch (Exception cqe) {
          Assert.fail("Failed to getCQService.", cqe);
        }

        AttributesFactory regionFactory = new AttributesFactory();
        regionFactory.setScope(Scope.LOCAL);

        if (poolName != null) {
          regionFactory.setPoolName(poolName);
        } else {
          if (redundancyLevel != null) {
            ClientServerTestCase.configureConnectionPool(regionFactory, serverHost, serverPorts,
                true, Integer.parseInt(redundancyLevel), -1, null);
          } else {
            ClientServerTestCase.configureConnectionPool(regionFactory, serverHost, serverPorts,
                true, -1, -1, null);
          }
        }
        for (final String region : regions) {
          createRegion(region, regionFactory.createRegionAttributes());
          LogWriterUtils.getLogWriter()
              .info("### Successfully Created Region on Client :" + region);
          // region1.getAttributesMutator().setCacheListener(new CqListener());
        }
      }
    };

    client.invoke(createQService);
  }

  /* Close Client */
  public void closeClient(VM client) {
    SerializableRunnable closeCQService = new CacheSerializableRunnable("Close Client") {
      @Override
      public void run2() throws CacheException {
        LogWriterUtils.getLogWriter().info("### Close Client. ###");
        try {
          ((DefaultQueryService) getCache().getQueryService()).closeCqService();
        } catch (Exception ex) {
          LogWriterUtils.getLogWriter()
              .info("### Failed to get CqService during ClientClose() ###");
          // TODO: fix eaten exception
        }
      }
    };

    client.invoke(closeCQService);
    Wait.pause(1000);
  }

  public void createFunctionalIndex(VM vm, final String indexName, final String indexedExpression,
      final String fromClause) {
    vm.invoke(new CacheSerializableRunnable("Create Functional Index") {
      @Override
      public void run2() throws CacheException {
        QueryService qs = null;
        try {
          qs = getCache().getQueryService();
        } catch (Exception ex) {
          LogWriterUtils.getLogWriter()
              .info("### Failed to get CqService during ClientClose() ###");
          // TODO: fix eaten exception
        }
        try {
          qs.createIndex(indexName, IndexType.FUNCTIONAL, indexedExpression, fromClause);
        } catch (Exception ex) {
          LogWriterUtils.getLogWriter().info("### Failed to create Index :" + indexName);
          // TODO: fix eaten exception
        }
      }
    });
  }

  /* Create/Init values */
  public void createValues(VM vm, final String regionName, final int size) {
    vm.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(regionName);
        for (int i = 1; i <= size; i++) {
          region1.put(KEY + i, new Portfolio(i, i));
        }
        LogWriterUtils.getLogWriter()
            .info("### Number of Entries in Region :" + region1.keySet().size());
      }
    });
  }

  /* Create/Init values */
  public void createValuesWithTime(VM vm, final String regionName, final int size) {
    vm.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(regionName);
        for (int i = 1; i <= size; i++) {
          Portfolio portfolio = new Portfolio(i);
          portfolio.createTime = System.currentTimeMillis();
          region1.put(KEY + i, portfolio);
        }
        LogWriterUtils.getLogWriter()
            .info("### Number of Entries in Region :" + region1.keySet().size());
      }
    });
  }

  /* delete values */
  public void deleteValues(VM vm, final String regionName, final int size) {
    vm.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(regionName);
        for (int i = 1; i <= size; i++) {
          region1.destroy(KEY + i);
        }
        LogWriterUtils.getLogWriter()
            .info("### Number of Entries In Region after Delete :" + region1.keySet().size());
      }

    });
  }

  /**
   * support for invalidating values.
   */
  public void invalidateValues(VM vm, final String regionName, final int size) {
    vm.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(regionName);
        for (int i = 1; i <= size; i++) {
          region1.invalidate(KEY + i);
        }
        LogWriterUtils.getLogWriter()
            .info("### Number of Entries In Region after Delete :" + region1.keySet().size());
      }

    });
  }

  public void createPool(VM vm, String poolName, String server, int port) {
    createPool(vm, poolName, new String[] {server}, new int[] {port});
  }

  public void createPool(VM vm, final String poolName, final String[] servers, final int[] ports) {
    createPool(vm, poolName, servers, ports, null);
  }

  public void createPool(VM vm, final String poolName, final String[] servers, final int[] ports,
      final String redundancyLevel) {
    vm.invoke(new CacheSerializableRunnable("createPool :" + poolName) {
      @Override
      public void run2() throws CacheException {
        // Create Cache.
        getCache();
        IgnoredException.addIgnoredException("java.net.ConnectException");

        PoolFactory cpf = PoolManager.createFactory();
        cpf.setSubscriptionEnabled(true);

        if (redundancyLevel != null) {
          int redundancy = Integer.parseInt(redundancyLevel);
          cpf.setSubscriptionRedundancy(redundancy);
        }

        for (int i = 0; i < servers.length; i++) {
          LogWriterUtils.getLogWriter()
              .info("### Adding to Pool. ### Server : " + servers[i] + " Port : " + ports[i]);
          cpf.addServer(servers[i], ports[i]);
        }

        cpf.create(poolName);
      }
    });
  }

  /* Register CQs */
  public void createCQ(VM vm, final String poolName, final String cqName, final String queryStr) {
    vm.invoke(new CacheSerializableRunnable("Create CQ :" + cqName) {
      @Override
      public void run2() throws CacheException {
        LogWriterUtils.getLogWriter().info("### Create CQ. ###" + cqName);
        // Get CQ Service.
        QueryService qService = null;
        try {
          qService = (PoolManager.find(poolName)).getQueryService();
        } catch (Exception cqe) {
          Assert.fail("Failed to getCQService.", cqe);
        }
        // Create CQ Attributes.
        CqAttributesFactory cqf = new CqAttributesFactory();
        CqListener[] cqListeners = {new CqQueryTestListener(LogWriterUtils.getLogWriter())};
        ((CqQueryTestListener) cqListeners[0]).cqName = cqName;

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

        // Create CQ.
        try {
          CqQuery cq1 = qService.newCq(cqName, queryStr, cqa);
          assertTrue("newCq() state mismatch", cq1.getState().isStopped());
        } catch (Exception ex) {
          LogWriterUtils.getLogWriter().info("QueryService is :" + qService, ex);
          Assert.fail("Failed to create CQ " + cqName + " . ", ex);
        }
      }
    });
  }

  // REMOVE..........
  public void createCQ(VM vm, final String cqName, final String queryStr) {
    vm.invoke(new CacheSerializableRunnable("Create CQ :" + cqName) {
      @Override
      public void run2() throws CacheException {
        // pause(60 * 1000);
        // getLogWriter().info("### DEBUG CREATE CQ START ####");
        // pause(20 * 1000);

        LogWriterUtils.getLogWriter().info("### Create CQ. ###" + cqName);
        // Get CQ Service.
        QueryService qService = null;
        try {
          qService = getCache().getQueryService();
        } catch (Exception cqe) {
          Assert.fail("Failed to getCQService.", cqe);
        }
        // Create CQ Attributes.
        CqAttributesFactory cqf = new CqAttributesFactory();
        CqListener[] cqListeners = {new CqQueryTestListener(LogWriterUtils.getLogWriter())};
        ((CqQueryTestListener) cqListeners[0]).cqName = cqName;

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

        // Create CQ.
        try {
          CqQuery cq1 = qService.newCq(cqName, queryStr, cqa);
          assertTrue("newCq() state mismatch", cq1.getState().isStopped());
        } catch (Exception ex) {
          LogWriterUtils.getLogWriter().info("QueryService is :" + qService, ex);
          Assert.fail("Failed to create CQ " + cqName + " . ", ex);
        }
      }
    });
  }

  /* Register CQs with no name, execute, and close */
  public void createAndExecCQNoName(VM vm, final String poolName, final String queryStr) {
    vm.invoke(new CacheSerializableRunnable("Create CQ with no name:") {
      @Override
      public void run2() throws CacheException {
        // pause(60 * 1000);
        LogWriterUtils.getLogWriter().info("### DEBUG CREATE CQ START ####");
        // pause(20 * 1000);

        LogWriterUtils.getLogWriter().info("### Create CQ with no name. ###");
        // Get CQ Service.
        QueryService qService = null;
        CqQuery cq1 = null;
        String cqName = null;

        try {
          qService = (PoolManager.find(poolName)).getQueryService();
        } catch (Exception cqe) {
          Assert.fail("Failed to getCQService.", cqe);
        }

        SelectResults cqResults = null;
        for (int i = 0; i < 20; ++i) {
          // Create CQ Attributes.
          CqAttributesFactory cqf = new CqAttributesFactory();
          CqListener[] cqListeners = {new CqQueryTestListener(LogWriterUtils.getLogWriter())};

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

          // Create CQ with no name and execute with initial results.
          try {
            cq1 = qService.newCq(queryStr, cqa);
            ((CqQueryTestListener) cqListeners[0]).cqName = cq1.getName();
          } catch (Exception ex) {
            LogWriterUtils.getLogWriter().info("CQService is :" + qService);
            Assert.fail("Failed to create CQ with no name" + " . ", ex);
          }

          if (cq1 == null) {
            LogWriterUtils.getLogWriter().info("Failed to get CqQuery object for CQ with no name.");
          } else {
            cqName = cq1.getName();
            LogWriterUtils.getLogWriter().info("Created CQ with no name, generated CQ name: "
                + cqName + " CQ state:" + cq1.getState());
            assertTrue("Create CQ with no name illegal state", cq1.getState().isStopped());
          }
          if (i % 2 == 0) {
            try {
              cqResults = cq1.executeWithInitialResults();
            } catch (Exception ex) {
              LogWriterUtils.getLogWriter().info("CqService is :" + qService);
              Assert.fail("Failed to execute CQ with initial results, cq name: " + cqName + " . ",
                  ex);
            }
            LogWriterUtils.getLogWriter().info("initial result size = " + cqResults.size());
            LogWriterUtils.getLogWriter()
                .info("CQ state after execute with initial results = " + cq1.getState());
            assertTrue("executeWithInitialResults() state mismatch", cq1.getState().isRunning());
          } else {
            try {
              cq1.execute();
            } catch (Exception ex) {
              LogWriterUtils.getLogWriter().info("CQService is :" + qService);
              Assert.fail("Failed to execute CQ " + cqName + " . ", ex);
            }
            LogWriterUtils.getLogWriter().info("CQ state after execute = " + cq1.getState());
            assertTrue("execute() state mismatch", cq1.getState().isRunning());
          }

          // Close the CQ
          try {
            cq1.close();
          } catch (Exception ex) {
            LogWriterUtils.getLogWriter().info("CqService is :" + qService);
            Assert.fail("Failed to close CQ " + cqName + " . ", ex);
          }
          assertTrue("closeCq() state mismatch", cq1.getState().isClosed());
        }
      }
    });
  }

  public void executeCQ(VM vm, final String cqName, final boolean initialResults,
      String expectedErr) {
    executeCQ(vm, cqName, initialResults, noTest, null, expectedErr);
  }

  /**
   * Execute/register CQ as running.
   *
   * @param initialResults true if initialResults are requested
   * @param expectedResultsSize if >= 0, validate results against this size
   * @param expectedErr if not null, an error we expect
   */
  public void executeCQ(VM vm, final String cqName, final boolean initialResults,
      final int expectedResultsSize, final String[] expectedKeys, final String expectedErr) {

    vm.invoke(new CacheSerializableRunnable("Execute CQ :" + cqName) {

      private void work() throws CacheException {
        LogWriterUtils.getLogWriter().info("### DEBUG EXECUTE CQ START ####");

        // Get CQ Service.
        QueryService cqService = null;
        CqQuery cq1 = null;
        cqService = getCache().getQueryService();

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

        if (initialResults) {
          SelectResults cqResults = null;

          try {
            cqResults = cq1.executeWithInitialResults();
          } catch (Exception ex) {
            fail("Failed to execute  CQ " + cqName, ex);
          }
          LogWriterUtils.getLogWriter().info("initial result size = " + cqResults.size());
          assertTrue("executeWithInitialResults() state mismatch", cq1.getState().isRunning());
          if (expectedResultsSize >= 0) {
            assertEquals(
                "Unexpected results size for CQ: " + cqName + " CQ Query :" + cq1.getQueryString(),
                expectedResultsSize, cqResults.size());
          }

          if (expectedKeys != null) {
            HashSet resultKeys = new HashSet();
            for (Object o : cqResults.asList()) {
              Struct s = (Struct) o;
              resultKeys.add(s.get("key"));
            }
            for (final String expectedKey : expectedKeys) {
              assertTrue(
                  "Expected key :" + expectedKey + " Not found in CqResults for CQ: " + cqName
                      + " CQ Query :" + cq1.getQueryString() + " Keys in CqResults :" + resultKeys,
                  resultKeys.contains(expectedKey));
            }
          }
        } else {
          try {
            cq1.execute();
          } catch (Exception ex) {
            if (expectedErr == null) {
              LogWriterUtils.getLogWriter().info("CqService is :" + cqService, ex);
            }
            Assert.fail("Failed to execute  CQ " + cqName, ex);
          }
          assertTrue("execute() state mismatch", cq1.getState().isRunning());
        }
      }

      @Override
      public void run2() throws CacheException {
        if (expectedErr != null) {
          getCache().getLogger()
              .info("<ExpectedException action=add>" + expectedErr + "</ExpectedException>");
        }
        try {
          work();
        } finally {
          if (expectedErr != null) {
            getCache().getLogger()
                .info("<ExpectedException action=remove>" + expectedErr + "</ExpectedException>");
          }
        }
      }
    });
  }

  /* Stop/pause CQ */
  public void stopCQ(VM vm, final String cqName) throws Exception {
    vm.invoke(new CacheSerializableRunnable("Stop CQ :" + cqName) {
      @Override
      public void run2() throws CacheException {
        LogWriterUtils.getLogWriter().info("### Stop CQ. ###" + cqName);
        // Get CQ Service.
        QueryService cqService = null;
        try {
          cqService = getCache().getQueryService();
        } catch (Exception cqe) {
          Assert.fail("Failed to getCQService.", cqe);
        }

        // Stop CQ.
        CqQuery cq1 = null;
        try {
          cq1 = cqService.getCq(cqName);
          cq1.stop();
        } catch (Exception ex) {
          Assert.fail("Failed to stop CQ " + cqName + " . ", ex);
        }
        assertTrue("Stop CQ state mismatch", cq1.getState().isStopped());
      }
    });
  }

  // Stop and execute CQ repeatedly
  /* Stop/pause CQ */
  private void stopExecCQ(VM vm, final String cqName, final int count) throws Exception {
    vm.invoke(new CacheSerializableRunnable("Stop CQ :" + cqName) {
      @Override
      public void run2() throws CacheException {
        CqQuery cq1 = null;
        LogWriterUtils.getLogWriter().info("### Stop and Exec CQ. ###" + cqName);
        // Get CQ Service.
        QueryService cqService = null;
        try {
          cqService = getCache().getQueryService();
        } catch (Exception cqe) {
          Assert.fail("Failed to getCqService.", cqe);
        }

        // Get CQ.
        try {
          cq1 = cqService.getCq(cqName);
        } catch (Exception ex) {
          Assert.fail("Failed to get CQ " + cqName + " . ", ex);
        }

        for (int i = 0; i < count; ++i) {
          // Stop CQ.
          try {
            cq1.stop();
          } catch (Exception ex) {
            Assert.fail("Count = " + i + "Failed to stop CQ " + cqName + " . ", ex);
          }
          assertTrue("Stop CQ state mismatch, count = " + i, cq1.getState().isStopped());
          LogWriterUtils.getLogWriter()
              .info("After stop in Stop and Execute loop, ran successfully, loop count: " + i);
          LogWriterUtils.getLogWriter().info("CQ state: " + cq1.getState());

          // Re-execute CQ
          try {
            cq1.execute();
          } catch (Exception ex) {
            Assert.fail("Count = " + i + "Failed to execute CQ " + cqName + " . ", ex);
          }
          assertTrue("Execute CQ state mismatch, count = " + i, cq1.getState().isRunning());
          LogWriterUtils.getLogWriter()
              .info("After execute in Stop and Execute loop, ran successfully, loop count: " + i);
          LogWriterUtils.getLogWriter().info("CQ state: " + cq1.getState());
        }
      }
    });
  }

  /* UnRegister CQs */
  public void closeCQ(VM vm, final String cqName) throws Exception {
    vm.invoke(new CacheSerializableRunnable("Close CQ :" + cqName) {
      @Override
      public void run2() throws CacheException {
        LogWriterUtils.getLogWriter().info("### Close CQ. ###" + cqName);
        // Get CQ Service.
        QueryService cqService = null;
        try {
          cqService = getCache().getQueryService();
        } catch (Exception cqe) {
          Assert.fail("Failed to getCqService.", cqe);
        }

        // Close CQ.
        CqQuery cq1 = null;
        try {
          cq1 = cqService.getCq(cqName);
          cq1.close();
