@Test
  public void testCqAndInterestRegistrations() 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)

    final String[] queries =
        new String[] {"SELECT * FROM " + regName + " p WHERE p.ticker = 'vmware'",
            "SELECT * FROM " + regName + " WHERE id > 5",};

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

    // Start server2
    vm1.invoke(new CacheSerializableRunnable("Create cache server") {
      @Override
      public void run2() throws CacheException {
        configAndStartBridgeServer(false, true);
        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, host0}, new int[] {port0, port1}, true);
    createPool(vm3, poolName, new String[] {host0, host0}, new int[] {port1, port0}, true);

    final String cqName = "testCq";

    vm3.invoke(new CacheSerializableRunnable("init region") {
      @Override
      public void run2() throws CacheException {
        QueryService localQueryService = null;

        AttributesFactory factory = new AttributesFactory();
        factory.setScope(Scope.LOCAL);
        ClientServerTestCase.configureConnectionPool(factory, host0, port1, -1, true, -1, -1, null);
        Region region = createRegion(regionName, rootRegionName, factory.create());

        for (int i = 0; i < numberOfEntries; i++) {
          region.put("key-" + i, new TestObject(i, "vmware"));
        }
      }
    });

    vm2.invoke(new CacheSerializableRunnable("init region") {
      @Override
      public void run2() throws CacheException {
        QueryService localQueryService = null;

        AttributesFactory factory = new AttributesFactory();
        factory.setScope(Scope.LOCAL);
        ClientServerTestCase.configureConnectionPool(factory, host0, port0, -1, true, -1, -1, null);
        Region region = createRegion(regionName, rootRegionName, factory.create());
      }
    });

    SerializableRunnable subscribe = new CacheSerializableRunnable("subscribe") {
      @Override
      public void run2() throws CacheException {

        // Register interest
        Region region = getRootRegion().getSubregion(regionName);
        List list = new ArrayList();
        for (int i = 1; i <= numberOfEntries * 3; i++) {
          if (i % 4 == 0) {
            list.add("key-" + i);
          }
        }
        region.registerInterest(list);

        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.
        for (int i = 0; i < queries.length; i++) {
          CqAttributesFactory cqf = new CqAttributesFactory();
          CqListener[] cqListeners = {new CqQueryTestListener(LogWriterUtils.getLogWriter())};
          ((CqQueryTestListener) cqListeners[0]).cqName = (cqName + i);

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

          // Create CQ.
          try {
            CqQuery cq = qService.newCq(cqName + i, queries[i], 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(subscribe);
    vm3.invoke(subscribe);

    // 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);
      }
    });

    vm0.invoke(new CacheSerializableRunnable("Create cache server") {
      @Override
      public void run2() throws CacheException {
        Region region = getRootRegion().getSubregion(regionName);
        // Check for TestObject instances.
        assertEquals(0, TestObject.numInstance);
      }
    });

    vm3.invoke(new CacheSerializableRunnable("Update") {
      @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"));
        }
      }
    });

    // Validate CQs.
    for (int i = 0; i < queries.length; i++) {
      int expectedEvent = 0;
      int updateEvents = 0;

      if (i != 0) {
        expectedEvent = numberOfEntries * 2 - queryLimit;
        updateEvents = numberOfEntries - queryLimit;
      } else {
        expectedEvent = numberOfEntries * 2;
        updateEvents = numberOfEntries;
      }

      validateCq(vm2, cqName + i, expectedEvent, numberOfEntries, updateEvents);
      validateCq(vm3, cqName + i, expectedEvent, numberOfEntries, updateEvents);
    }


    vm0.invoke(new CacheSerializableRunnable("Create cache server") {
      @Override
      public void run2() throws CacheException {
        Region region = getRootRegion().getSubregion(regionName);
        // Check for TestObject instances.
        assertEquals(0, 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);
      }
    });


    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