/*
 * 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.LOG_LEVEL;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
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.assertNull;
import static org.apache.geode.test.dunit.Assert.assertTrue;
import static org.apache.geode.test.dunit.Assert.fail;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.assertj.core.api.Assertions.assertThat;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
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.CacheFactory;
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.ClientCache;
import org.apache.geode.cache.client.ClientCacheFactory;
import org.apache.geode.cache.client.ClientRegionShortcut;
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.Struct;
import org.apache.geode.cache.query.data.Portfolio;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.cache30.ClientServerTestCase;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.NetworkUtils;
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.cache.internal.JUnit4CacheTestCase;
import org.apache.geode.test.junit.categories.ClientSubscriptionTest;

/**
 * Test class for Partitioned Region and CQs
 *
 * @since GemFire 5.5
 */
@Category({ClientSubscriptionTest.class})
public class PartitionedRegionCqQueryDUnitTest extends JUnit4CacheTestCase {

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

  public static final String KEY = "key-";

  protected final CqQueryDUnitTest cqHelper = new CqQueryDUnitTest();

  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[1] + " p where p.ID > 0",};

  public final String[] cqsWithoutRoot = new String[] {
      // 0 - Test for ">"
      "SELECT ALL * FROM " + SEPARATOR + regions[0] + " p where p.ID > 0"};

  private static int bridgeServerPort;

  @Test
  public void testPutAllWithCQLocalDestroy() {
    VM server1 = getVM(0);
    VM server2 = getVM(1);
    VM client = getVM(2);

    final String cqName = "testPutAllWithCQLocalDestroy_0";
    createServer(server1);
    createServer(server2);
    final String host = VM.getHostName();
    final int port = server2.invoke(PartitionedRegionCqQueryDUnitTest::getCacheServerPort);
    createClient(client, port, host);
    createCQ(client, cqName, cqs[0]);

    int numObjects = 1000;

    server1.invoke(() -> {
      Region<String, Object> region =
          getCache().getRegion(SEPARATOR + "root" + SEPARATOR + regions[0]);
      Map<String, Object> buffer = new HashMap();
      for (int i = 1; i < numObjects; i++) {
        Portfolio p = new Portfolio(i);
        buffer.put("" + i, p);
      }
      region.putAll(buffer);
      assertThat(region.size()).isEqualTo(numObjects - 1);
    });

    client.invoke(() -> {
      QueryService cqService = getCache().getQueryService();
      CqQuery cqQuery = cqService.getCq(cqName);
      assertThat(cqQuery)
          .withFailMessage("Failed to get CQ " + cqName)
          .isNotNull();
      cqQuery.executeWithInitialResults();
    });

    server1.invoke(() -> {
      Region<String, Object> region =
          getCache().getRegion(SEPARATOR + "root" + SEPARATOR + regions[0]);
      // PutAll with entries that do not satisfy CQ. This is to generate LOCAL_DESTROY CQ event
      Map<String, Object> buffer = new HashMap();
      for (int i = 1; i < numObjects; i++) {
        Portfolio p = new Portfolio(-1 * i);
        buffer.put("" + i, p);
      }
      region.putAll(buffer);
      assertThat(region.size()).isEqualTo(numObjects - 1);
    });

    client.invoke(() -> {
      QueryService cqService = getCache().getQueryService();
      CqQuery cqQuery = cqService.getCq(cqName);
      assertThat(cqQuery)
          .withFailMessage("Failed to get CQ " + cqName)
          .isNotNull();
      CqQueryTestListener cqListener =
          (CqQueryTestListener) cqQuery.getCqAttributes().getCqListener();
      await().untilAsserted(
          () -> assertThat(cqListener.getTotalEventCount()).isEqualTo(numObjects - 1));
    });

    cqHelper.closeClient(client);
    cqHelper.closeServer(server2);
    cqHelper.closeServer(server1);
  }

  @Test
  public void testRemoveAllWithCQLocalDestroy() {
    VM server1 = getVM(0);
    VM server2 = getVM(1);
    VM client = getVM(2);

    final String cqName = "testRemoveAllWithCQLocalDestroy_0";
    createServer(server1);
    createServer(server2);
    final String host = VM.getHostName();
    final int port = server2.invoke(PartitionedRegionCqQueryDUnitTest::getCacheServerPort);
    createClient(client, port, host);
    createCQ(client, cqName, cqs[0]);

    int numObjects = 1000;

    server1.invoke(() -> {
      Region<String, Object> region =
          getCache().getRegion(SEPARATOR + "root" + SEPARATOR + regions[0]);
      Map<String, Object> buffer = new HashMap();
      for (int i = 1; i < numObjects; i++) {
        Portfolio p = new Portfolio(i);
        buffer.put("" + i, p);
      }
      region.putAll(buffer);
      assertThat(region.size()).isEqualTo(numObjects - 1);
    });

    client.invoke(() -> {
      QueryService cqService = getCache().getQueryService();
      CqQuery cqQuery = cqService.getCq(cqName);
      assertThat(cqQuery)
          .withFailMessage("Failed to get CQ " + cqName)
          .isNotNull();
      cqQuery.executeWithInitialResults();
    });

    server1.invoke(() -> {
      Region<String, Object> region =
          getCache().getRegion(SEPARATOR + "root" + SEPARATOR + regions[0]);
      Set<String> keys = new HashSet<>();
      for (int i = 1; i < numObjects; i++) {
        keys.add("" + i);
      }
      // This is to generate LOCAL_DESTROY CQ event
      region.removeAll(keys);
      assertThat(region.size()).isEqualTo(0);
    });

    client.invoke(() -> {
      QueryService cqService = getCache().getQueryService();
      CqQuery cqQuery = cqService.getCq(cqName);
      assertThat(cqQuery)
          .withFailMessage("Failed to get CQ " + cqName)
          .isNotNull();
      CqQueryTestListener cqListener =
          (CqQueryTestListener) cqQuery.getCqAttributes().getCqListener();
      await().untilAsserted(
          () -> assertThat(cqListener.getTotalEventCount()).isEqualTo(numObjects - 1));
    });

    cqHelper.closeClient(client);
    cqHelper.closeServer(server2);
    cqHelper.closeServer(server1);
  }

  @Test
  public void testCQLeakWithPartitionedRegion() throws Exception {
    // creating servers.
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM server2 = host.getVM(1);
    VM client = host.getVM(2);

    createServer(server1);
    createServer(server2);

    // create client

    final int port = server1.invoke(PartitionedRegionCqQueryDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName(server1.getHost());

    createClient(client, port, host0);

    // register cq.
    createCQ(client, "testCQEvents_0", cqs[0]);
    cqHelper.executeCQ(client, "testCQEvents_0", false, null);
    Wait.pause(2 * 1000);

    // create values
    int size = 40;
    createValues(server1, regions[0], size);

    // wait for last creates...

    cqHelper.waitForCreated(client, "testCQEvents_0", KEY + size);

    // validate cq..
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ 0, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ 0, /* queryDeletes: */ 0, /* totalEvents: */ size);

    int cc1 = server1.invoke(PartitionedRegionCqQueryDUnitTest::getCqCountFromRegionProfile);
    int cc2 = server2.invoke(PartitionedRegionCqQueryDUnitTest::getCqCountFromRegionProfile);
    assertEquals("Should have one", 1, cc1);
    assertEquals("Should have one", 1, cc2);

    server1.bounce();

    cqHelper.closeClient(client);
    Wait.pause(10 * 1000);
    cc2 = server2.invoke(PartitionedRegionCqQueryDUnitTest::getCqCountFromRegionProfile);

    // assertIndexDetailsEquals("Should have one", 0, cc1);
    assertEquals("Should have one", 0, cc2);

    cqHelper.closeServer(server2);
    // cqHelper.closeServer(server1);
  }

  @Test
  public void testCQAndPartitionedRegion() throws Exception {
    // creating servers.
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM server2 = host.getVM(1);
    VM client = host.getVM(2);

    createServer(server1);

    createServer(server2);

    // create client

    final int port = server1.invoke(PartitionedRegionCqQueryDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName(server1.getHost());

    createClient(client, port, host0);

    // register cq.
    createCQ(client, "testCQEvents_0", cqs[0]);
    cqHelper.executeCQ(client, "testCQEvents_0", false, null);
    Wait.pause(2 * 1000);

    // create values
    int size = 40;
    createValues(server1, regions[0], size);

    // wait for last creates...

    cqHelper.waitForCreated(client, "testCQEvents_0", KEY + size);

    // validate cq..
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ 0, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ 0, /* queryDeletes: */ 0, /* totalEvents: */ size);

    // do updates
    createValues(server1, regions[0], size);

    cqHelper.waitForUpdated(client, "testCQEvents_0", KEY + size);

    // validate cqs again
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ size, /* queryDeletes: */ 0, /* totalEvents: */ (size + size));

    // destroy all the values.
    int numDestroys = size;
    cqHelper.deleteValues(server2, regions[0], numDestroys);

    cqHelper.waitForDestroyed(client, "testCQEvents_0", KEY + numDestroys);

    // validate cqs after destroys on server2.

    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ numDestroys,
        /* queryInserts: */ size, /* queryUpdates: */ size, /* queryDeletes: */ numDestroys,
        /* totalEvents: */ (size + size + numDestroys));

    cqHelper.closeClient(client);
    cqHelper.closeServer(server2);
    cqHelper.closeServer(server1);
  }

  /**
   * test for registering cqs on a cache server with local max memory zero.
   */
  @Test
  public void testPartitionedCqOnAccessorBridgeServer() throws Exception {
    // creating servers.
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM server2 = host.getVM(1);
    VM client = host.getVM(2);

    // creating an accessor vm with cache server installed.
    createServer(server1, true);

    createServer(server2);

    // create client

    final int port = server1.invoke(PartitionedRegionCqQueryDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName(server1.getHost());

    createClient(client, port, host0);

    // register cq.
    createCQ(client, "testCQEvents_0", cqs[0]);
    cqHelper.executeCQ(client, "testCQEvents_0", false, null);


    // create values
    final int size = 1000;
    createValues(server1, regions[0], size);

    // wait for last creates...

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForCreated(client, "testCQEvents_0", KEY + i);
    }

    // validate cq..
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ 0, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ 0, /* queryDeletes: */ 0, /* totalEvents: */ size);

    // do updates
    createValues(server1, regions[0], size);

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForUpdated(client, "testCQEvents_0", KEY + i);
    }

    // validate cqs again.
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ size, /* queryDeletes: */ 0, /* totalEvents: */ (size + size));


    // destroy all the values.
    int numDestroys = size;
    cqHelper.deleteValues(server2, regions[0], numDestroys);

    for (int i = 1; i <= numDestroys; i++) {
      cqHelper.waitForDestroyed(client, "testCQEvents_0", KEY + i);
    }

    // validate cqs after destroys on server2.

    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ numDestroys,
        /* queryInserts: */ size, /* queryUpdates: */ size, /* queryDeletes: */ numDestroys,
        /* totalEvents: */ (size + size + numDestroys));

    cqHelper.closeClient(client);
    cqHelper.closeServer(server2);
    cqHelper.closeServer(server1);
  }

  /**
   * test for registering cqs on single cache server hosting all the data. This will generate all
   * the events locally and should always have the old value and should not sent the profile update
   * on wire.
   */
  @Test
  public void testPartitionedCqOnSingleBridgeServer() throws Exception {
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM client = host.getVM(2);

    // creating an accessor vm with cache server installed.
    createServer(server1);
    final int port = server1.invoke(PartitionedRegionCqQueryDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName(server1.getHost());

    createClient(client, port, host0);

    // register cq.
    createCQ(client, "testCQEvents_0", cqs[0]);
    cqHelper.executeCQ(client, "testCQEvents_0", false, null);


    // create values
    final int size = 400;
    createValues(server1, regions[0], size);

    // wait for last creates...

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForCreated(client, "testCQEvents_0", KEY + i);
    }

    // validate cq..
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ 0, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ 0, /* queryDeletes: */ 0, /* totalEvents: */ size);

    // do updates
    createValues(server1, regions[0], size);

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForUpdated(client, "testCQEvents_0", KEY + i);
    }

    // validate cqs again.
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ size, /* queryDeletes: */ 0, /* totalEvents: */ (size + size));

    // destroy all the values.
    int numDestroys = size;
    cqHelper.deleteValues(server1, regions[0], numDestroys);

    for (int i = 1; i <= numDestroys; i++) {
      cqHelper.waitForDestroyed(client, "testCQEvents_0", KEY + i);
    }

    // validate cqs after destroys on server2.

    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ numDestroys,
        /* queryInserts: */ size, /* queryUpdates: */ size, /* queryDeletes: */ numDestroys,
        /* totalEvents: */ (size + size + numDestroys));

    cqHelper.closeClient(client);
    cqHelper.closeServer(server1);
  }

  /**
   * test for registering cqs on single cache server hosting all the data. This will generate all
   * the events locally but the puts, updates and destroys originate at an accessor vm.
   */
  @Test
  public void testPRCqOnSingleBridgeServerUpdatesOriginatingAtAccessor() throws Exception {
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM server2 = host.getVM(1);
    VM client = host.getVM(2);

    // creating an accessor vm with cache server installed.
    createServer(server1, true);

    assertLocalMaxMemory(server1);

    createServer(server2);

    // create client

    final int port = server2.invoke(PartitionedRegionCqQueryDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName(server2.getHost());

    createClient(client, port, host0);

    // register cq.
    createCQ(client, "testCQEvents_0", cqs[0]);
    cqHelper.executeCQ(client, "testCQEvents_0", false, null);


    // create values
    final int size = 400;
    createValues(server1, regions[0], size);

    // wait for last creates...

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForCreated(client, "testCQEvents_0", KEY + i);
    }

    // validate cq..
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ 0, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ 0, /* queryDeletes: */ 0, /* totalEvents: */ size);

    // do updates
    createValues(server1, regions[0], size);

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForUpdated(client, "testCQEvents_0", KEY + i);
    }

    // validate cqs again.
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ size, /* queryDeletes: */ 0, /* totalEvents: */ (size + size));

    // destroy all the values.
    int numDestroys = size;
    cqHelper.deleteValues(server1, regions[0], numDestroys);

    for (int i = 1; i <= numDestroys; i++) {
      cqHelper.waitForDestroyed(client, "testCQEvents_0", KEY + i);
    }

    // validate cqs after destroys on server2.

    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ numDestroys,
        /* queryInserts: */ size, /* queryUpdates: */ size, /* queryDeletes: */ numDestroys,
        /* totalEvents: */ (size + size + numDestroys));

    cqHelper.closeClient(client);
    cqHelper.closeServer(server2);
    cqHelper.closeServer(server1);
  }

  /**
   * test to check invalidates on cache server hosting datastores as well.
   */
  @Test
  public void testPRCqWithInvalidatesOnBridgeServer() {
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM server2 = host.getVM(1);
    VM client = host.getVM(2);

    // creating cache server with data store. clients will connect to this
    // cache server.
    createServer(server1);

    // create another server with data store.
    createServer(server2);

    final int port = server1.invoke(PartitionedRegionCqQueryDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName(server1.getHost());

    createClient(client, port, host0);

    // register cq.
    createCQ(client, "testCQEvents_0", cqs[0]);
    cqHelper.executeCQ(client, "testCQEvents_0", false, null);


    // create values
    final int size = 400;
    createValues(server1, regions[0], size);

    // wait for last creates...

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForCreated(client, "testCQEvents_0", KEY + i);
    }

    // validate cq..
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ 0, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ 0, /* queryDeletes: */ 0, /* totalEvents: */ size);

    // do updates
    createValues(server1, regions[0], size);

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForUpdated(client, "testCQEvents_0", KEY + i);
    }

    // validate cqs again.
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ size, /* queryDeletes: */ 0, /* totalEvents: */ (size + size));

    // invalidate all the values.
    int numInvalidates = size;
    cqHelper.invalidateValues(server2, regions[0], numInvalidates);

    for (int i = 1; i <= numInvalidates; i++) {
      cqHelper.waitForInvalidated(client, "testCQEvents_0", KEY + i);
    }

    // validate cqs after invalidates on server2.

    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ numInvalidates,
        /* queryInserts: */ size, /* queryUpdates: */ size, /* queryDeletes: */ numInvalidates,
        /* totalEvents: */ (size + size + numInvalidates));

    cqHelper.closeClient(client);
    cqHelper.closeServer(server2);
    cqHelper.closeServer(server1);
  }

  /**
   * test cqs with invalidates on cache server not hosting datastores.
   */
  @Test
  public void testPRCqWithInvalidatesOnAccessorBridgeServer() throws Exception {
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM server2 = host.getVM(1);
    VM client = host.getVM(2);

    // creating cache server with data store. clients will connect to this
    // cache server.
    createServer(server1, true);

    // create another server with data store.
    createServer(server2);

    final int port = server1.invoke(PartitionedRegionCqQueryDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName(server1.getHost());

    createClient(client, port, host0);

    // register cq.
    createCQ(client, "testCQEvents_0", cqs[0]);
    cqHelper.executeCQ(client, "testCQEvents_0", false, null);


    // create values
    final int size = 400;
    createValues(server1, regions[0], size);

    // wait for last creates...

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForCreated(client, "testCQEvents_0", KEY + i);
    }

    // validate cq..
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ 0, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ 0, /* queryDeletes: */ 0, /* totalEvents: */ size);

    // do updates
    createValues(server1, regions[0], size);

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForUpdated(client, "testCQEvents_0", KEY + i);
    }

    // validate cqs again.
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ size, /* queryDeletes: */ 0, /* totalEvents: */ (size + size));

    // invalidate all the values.
    int numInvalidates = size;
    cqHelper.invalidateValues(server1, regions[0], numInvalidates);

    for (int i = 1; i <= numInvalidates; i++) {
      cqHelper.waitForInvalidated(client, "testCQEvents_0", KEY + i);
    }
    // validate cqs after invalidates on server2.

    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ numInvalidates,
        /* queryInserts: */ size, /* queryUpdates: */ size, /* queryDeletes: */ numInvalidates,
        /* totalEvents: */ (size + size + numInvalidates));

    cqHelper.closeClient(client);
    cqHelper.closeServer(server2);
    cqHelper.closeServer(server1);
  }

  /**
   * test cqs with create updates and destroys from client on cache server hosting datastores.
   */
  @Test
  public void testPRCqWithUpdatesFromClients() throws Exception {
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM server2 = host.getVM(1);
    VM client = host.getVM(2);
    VM client2 = host.getVM(3);

    // creating cache server with data store. clients will connect to this
    // cache server.
    createServer(server1, false, 1);

    // create another server with data store.
    createServer(server2, false, 1);

    final int port = server1.invoke(PartitionedRegionCqQueryDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName(server1.getHost());

    createClient(client, port, host0);
    createClient(client2, port, host0);

    // register cq.
    createCQ(client, "testCQEvents_0", cqs[0]);
    cqHelper.executeCQ(client, "testCQEvents_0", false, null);


    // create values
    final int size = 400;
    createValues(client2, regions[0], size);

    // wait for last creates...

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForCreated(client, "testCQEvents_0", KEY + i);
    }

    // validate cq..
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ 0, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ 0, /* queryDeletes: */ 0, /* totalEvents: */ size);

    // do updates
    createValues(client2, regions[0], size);

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForUpdated(client, "testCQEvents_0", KEY + i);
    }

    // validate cqs again.
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ size, /* queryDeletes: */ 0, /* totalEvents: */ (size + size));

    // invalidate all the values.
    int numDelets = size;

    cqHelper.deleteValues(client2, regions[0], numDelets);

    for (int i = 1; i <= numDelets; i++) {
      cqHelper.waitForDestroyed(client, "testCQEvents_0", KEY + i);
    }

    // validate cqs after invalidates on server2.

    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ numDelets,
        /* queryInserts: */ size, /* queryUpdates: */ size, /* queryDeletes: */ numDelets,
        /* totalEvents: */ (size + size + numDelets));

    cqHelper.closeClient(client);
    cqHelper.closeClient(client2);
    cqHelper.closeServer(server2);
    cqHelper.closeServer(server1);
  }

  /**
   * test cqs on multiple partitioned region hosted by cache servers.
   */
  @Test
  public void testPRCqWithMultipleRegionsOnServer() throws Exception {
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM server2 = host.getVM(1);
    VM client = host.getVM(2);
    VM client2 = host.getVM(3);

    // creating cache server with data store. clients will connect to this
    // cache server.
    createServer(server1, false, 1);

    // create another server with data store.
    createServer(server2, false, 1);

    // Wait for server to initialize.
    Wait.pause(2000);

    final int port = server1.invoke(PartitionedRegionCqQueryDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName(server1.getHost());

    createClient(client, port, host0);
    createClient(client2, port, host0);

    // register cq.
    createCQ(client, "testCQEvents_0", cqs[0]);
    createCQ(client, "testCQEvents_1", cqs[10]);
    cqHelper.executeCQ(client, "testCQEvents_0", false, null);
    cqHelper.executeCQ(client, "testCQEvents_1", false, null);

    // create values
    final int size = 400;
    createValues(client2, regions[0], size);
    createValues(client2, regions[1], size);

    // wait for last creates...

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForCreated(client, "testCQEvents_0", KEY + i);
      cqHelper.waitForCreated(client, "testCQEvents_1", KEY + i);
    }

    // validate cq..
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ 0, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ 0, /* queryDeletes: */ 0, /* totalEvents: */ size);
    cqHelper.validateCQ(client, "testCQEvents_1", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ 0, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ 0, /* queryDeletes: */ 0, /* totalEvents: */ size);

    // do updates
    createValues(client2, regions[0], size);
    createValues(client2, regions[1], size);

    for (int i = 1; i <= size; i++) {
      cqHelper.waitForUpdated(client, "testCQEvents_0", KEY + i);
      cqHelper.waitForUpdated(client, "testCQEvents_1", KEY + i);
    }

    // validate cqs again.
    cqHelper.validateCQ(client, "testCQEvents_0", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ size, /* queryDeletes: */ 0, /* totalEvents: */ (size + size));

    cqHelper.validateCQ(client, "testCQEvents_1", /* resultSize: */ CqQueryDUnitTest.noTest,
        /* creates: */ size, /* updates: */ size, /* deletes; */ 0, /* queryInserts: */ size,
        /* queryUpdates: */ size, /* queryDeletes: */ 0, /* totalEvents: */ (size + size));

    // invalidate all the values.
    int numInvalidates = size;
