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

    final String cqName = "testEventsDuringQueryExecution_0";

    // Server.
    createServer(server1);
    createServer(server2);

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

    String poolName1 = "testEventsDuringQueryExecution";
    createPool(client, poolName1, host0, port);

    // create CQ.
    createCQ(client, poolName1, cqName, cqs[0]);

    final int numObjects = 200;
    final int totalObjects = 500;

    // initialize Region.
    server1.invoke(new CacheSerializableRunnable("Update Region") {
      @Override
      public void run2() throws CacheException {
        Region region = getCache().getRegion(SEPARATOR + "root" + SEPARATOR + regions[0]);
        for (int i = 1; i <= numObjects; i++) {
          Portfolio p = new Portfolio(i);
          region.put("" + i, p);
        }
      }
    });

    // Keep updating region (async invocation).
    server1.invokeAsync(new CacheSerializableRunnable("Update Region") {
      @Override
      public void run2() throws CacheException {
        Region region = getCache().getRegion(SEPARATOR + "root" + SEPARATOR + regions[0]);
        for (int i = numObjects + 1; i <= totalObjects; i++) {
          Portfolio p = new Portfolio(i);
          region.put("" + i, p);
        }
      }
    });

    // Execute CQ while update is in progress.
    client.invoke(new CacheSerializableRunnable("Execute CQ") {
      @Override
      public void run2() throws CacheException {
        QueryService cqService = getCache().getQueryService();
        // Get CqQuery object.
        CqQuery cq1 = cqService.getCq(cqName);
        if (cq1 == null) {
          fail("Failed to get CQ " + cqName);
        }

        SelectResults cqResults = null;

        try {
          cqResults = cq1.executeWithInitialResults();
        } catch (Exception ex) {
          throw new AssertionError("Failed to execute  CQ " + cqName, ex);
        }

        // getLogWriter().info("initial result size = " + cqResults.size());

        CqQueryTestListener cqListener =
            (CqQueryTestListener) cq1.getCqAttributes().getCqListener();
        // Wait for the last key to arrive.
        for (int i = 0; i < 4; i++) {
          try {
            cqListener.waitForCreated("" + totalObjects);
            // Found skip from the loop.
            break;
          } catch (CacheException ex) {
            if (i == 3) {
              throw ex;
            }
          }
        }

        // Check if the events from CqListener are in order.
        int oldId = 0;
        for (Object cqEvent : cqListener.events.toArray()) {
          int newId = new Integer(cqEvent.toString());
          if (oldId > newId) {
            fail("Queued events for CQ Listener during execution with "
                + "Initial results is not in the order in which they are created.");
          }
          oldId = newId;
        }

        // Check if all the IDs are present as part of Select Results and CQ Events.
        HashSet ids = new HashSet(cqListener.events);
        for (Object o : cqResults.asList()) {
          Struct s = (Struct) o;
          ids.add(s.get("key"));
        }
        // Iterator iter = cqResults.asSet().iterator();
        // while (iter.hasNext()) {
        // Portfolio p = (Portfolio)iter.next();
        // ids.add(p.getPk());
        // //getLogWriter().info("Result set value : " + p.getPk());
        // }

        HashSet missingIds = new HashSet();
        String key = "";
        for (int i = 1; i <= totalObjects; i++) {
          key = "" + i;
          if (!(ids.contains(key))) {
            missingIds.add(key);
          }
        }

        if (!missingIds.isEmpty()) {
          fail("Missing Keys in either ResultSet or the Cq Event list. "
              + " Missing keys : [size : " + missingIds.size() + "]" + missingIds
              + " Ids in ResultSet and CQ Events :" + ids);
        }

      }
    });

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

  }
Test for events created during the CQ query execution. When CQs are executed using
executeWithInitialResults there may be possibility that the region changes during that time may
not be reflected in the query result set thus making the query data and region data
inconsistent.
geode-cq/src/distributedTest/java/org/apache/geode/cache/query/cq/dunit/PrCqUsingPoolDUnitTest.java