@Test
  public void testCq() throws CacheException {

    final Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);
    VM vm2 = host.getVM(2);
    VM vm3 = host.getVM(3);
    final int numberOfEntries = 10;
    final int queryLimit = 6; // where id > 5 (0-5)

    // Start server1
    vm0.invoke(new CacheSerializableRunnable("Create cache server") {
      @Override
      public void run2() throws CacheException {
        configAndStartBridgeServer();
        Region region = getRootRegion().getSubregion(regionName);
        for (int i = 0; i < numberOfEntries; i++) {
          region.put("key-" + i, new TestObject(i, "vmware"));
        }
      }
    });

    // Start server2
    vm1.invoke(new CacheSerializableRunnable("Create cache server") {
      @Override
      public void run2() throws CacheException {
        configAndStartBridgeServer();
        Region region = getRootRegion().getSubregion(regionName);
      }
    });

    // Client pool.
    final int port0 = vm0.invoke(PdxQueryCQTestBase::getCacheServerPort);
    final int port1 = vm1.invoke(PdxQueryCQTestBase::getCacheServerPort);

    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());

    // Create client pool.
    final String poolName = "testCqPool";
    createPool(vm2, poolName, new String[] {host0}, new int[] {port0}, true);
    createPool(vm3, poolName, new String[] {host0}, new int[] {port1}, true);
    final String cqName = "testCq";

    // Execute CQ
    SerializableRunnable executeCq = new CacheSerializableRunnable("Execute queries") {
      @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 cq = qService.newCq(cqName, queryString[3], cqa);
          SelectResults sr = cq.executeWithInitialResults();
          for (Object o : sr.asSet()) {
            Struct s = (Struct) o;
            Object value = s.get("value");
            if (!(value instanceof TestObject)) {
              fail(
                  "Expected type TestObject, not found in result set. Found type :" + o.getClass());
            }
          }
        } catch (Exception ex) {
          AssertionError err = new AssertionError("Failed to create CQ " + cqName + " . ", ex);
          LogWriterUtils.getLogWriter().info("QueryService is :" + qService, err);
          throw err;
        }
      }
    };

    vm2.invoke(executeCq);
    vm3.invoke(executeCq);

    // Check for TestObject instances on Server2.
    // It should be 0
    vm1.invoke(new CacheSerializableRunnable("Create cache server") {
      @Override
      public void run2() throws CacheException {
        assertEquals(0, TestObject.numInstance);
      }
    });

    // update
    vm0.invoke(new CacheSerializableRunnable("Create cache server") {
      @Override
      public void run2() throws CacheException {
        Region region = getRootRegion().getSubregion(regionName);
        for (int i = 0; i < numberOfEntries * 2; i++) {
          region.put("key-" + i, new TestObject(i, "vmware"));
        }
        // Check for TestObject instances.
        assertEquals(numberOfEntries * 3, TestObject.numInstance);
      }
    });

    // Check for TestObject instances on Server2.
    // It should be 0
    vm1.invoke(new CacheSerializableRunnable("Create cache server") {
      @Override
      public void run2() throws CacheException {
        assertEquals(0, TestObject.numInstance);
      }
    });

    SerializableRunnable validateCq = new CacheSerializableRunnable("Validate CQs") {
      @Override
      public void run2() throws CacheException {
        LogWriterUtils.getLogWriter().info("### Validating CQ. ### " + cqName);
        // Get CQ Service.
        QueryService cqService = null;
        try {
          cqService = getCache().getQueryService();
        } catch (Exception cqe) {
          Assert.fail("Failed to getCQService.", cqe);
        }

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

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

        // Wait for the events to show up on the client.
        await().until(() -> listener.getTotalEventCount() >= (numberOfEntries * 2 - queryLimit));

        listener.printInfo(false);

        // Check for event type.
        Object[] cqEvents = listener.getEvents();
        for (Object o : cqEvents) {
          CqEvent cqEvent = (CqEvent) o;
          Object value = cqEvent.getNewValue();
          if (!(value instanceof TestObject)) {
            fail("Expected type TestObject, not found in result set. Found type :" + o.getClass());
          }
        }

        // Check for totalEvents count.
        assertEquals("Total Event Count mismatch", (numberOfEntries * 2 - queryLimit),
            listener.getTotalEventCount());

        // Check for create count.
        assertEquals("Create Event mismatch", numberOfEntries, listener.getCreateEventCount());

        // Check for update count.
        assertEquals("Update Event mismatch", numberOfEntries - queryLimit,
            listener.getUpdateEventCount());
      }
    };

    vm2.invoke(validateCq);
    vm3.invoke(validateCq);

    closeClient(vm2);
    closeClient(vm3);
    closeClient(vm1);
    closeClient(vm0);
  }
Tests client-server query on PdxInstance.
geode-cq/src/distributedTest/java/org/apache/geode/cache/query/dunit/PdxQueryCQDUnitTest.java