@Test
  public void testCQsWithPutallsWithTxValidateStats() 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, false, 1);

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

    // create values
    final int size = 100;

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

    String poolName = "testCQsWithPutallsTx";
    createPool(client, poolName, new String[] {host0}, new int[] {port});
    createClient(client, new int[] {port}, host0, null);

    // register cq.
    String cqQueryString =
        "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0] + " p where p.ID < 101";
    createCQ(client, poolName, "testCQEvents_0", cqQueryString);
    cqHelper.executeCQ(client, "testCQEvents_0", false, null);

    cqStatsHelper.validateCQStats(client, "testCQEvents_0", 0, 0, 0, 0, 0);

    cqHelper.registerInterestListCQ(client, regions[0], size, true);

    server1.invoke(new CacheSerializableRunnable("begin transaction") {
      @Override
      public void run2() throws CacheException {
        getCache().getCacheTransactionManager().begin();
      }
    });

    createValuesPutall(server1, regions[0], size);

    server1.invoke(new CacheSerializableRunnable("commit transaction") {
      @Override
      public void run2() throws CacheException {
        getCache().getCacheTransactionManager().commit();
      }
    });

    // wait for last creates...

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

    cqStatsHelper.validateCQStats(client, "testCQEvents_0", size, 0, 0, size, size);

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

    // size = 2;

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

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

    cqStatsHelper.validateCQStats(client, "testCQEvents_0", size, size, 0, size + size,
        size + size);

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


    // destroy all the values.
    int numInvalidates = size;
    server1.invoke(new CacheSerializableRunnable("begin transaction") {
      @Override
      public void run2() throws CacheException {
        getCache().getCacheTransactionManager().begin();
      }
    });

    cqHelper.deleteValues(server1, regions[0], numInvalidates);

    server1.invoke(new CacheSerializableRunnable("commit transaction") {
      @Override
      public void run2() throws CacheException {
        getCache().getCacheTransactionManager().commit();
      }
    });


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

    cqStatsHelper.validateCQStats(client, "testCQEvents_0", size, size, numInvalidates,
        size + size + numInvalidates, size + size + numInvalidates);

    // validate cqs after destroy.

    cqHelper.validateCQ(client, "testCQEvents_0",
        /* resultSize: */ CqQueryUsingPoolDUnitTest.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 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