@Test
  public void testCqStatInitializationTimingIssue() {
    disconnectAllFromDS();

    // The async close can cause this exception on the server
    IgnoredException.addIgnoredException("java.net.SocketException: Broken pipe");
    final String regionName = "testCqStatInitializationTimingIssue";
    final String cq1Name = "testCq1";
    final Host host = Host.getHost(0);
    VM server = host.getVM(0);
    VM client = host.getVM(1);
    VM client2 = host.getVM(2);

    // Start server 1
    final int server1Port = server
        .invoke(() -> CacheServerTestUtil.createCacheServer(regionName, Boolean.TRUE));

    // Start a client
    client.invoke(() -> CacheServerTestUtil.createCacheClient(
        getClientPool(NetworkUtils.getServerHostName(client.getHost()), server1Port), regionName));

    // Start a pub client
    client2.invoke(() -> CacheServerTestUtil.createCacheClient(
        getClientPool(NetworkUtils.getServerHostName(client2.getHost()), server1Port), regionName));

    // client has thread that invokes new and remove cq over and over
    client.invokeAsync(new CacheSerializableRunnable("Register cq") {
      @Override
      public void run2() throws CacheException {
        for (int i = 0; i < 10000; i++) {
          CqQuery query = createCq(regionName, cq1Name);
          if (query != null) {
            try {
              query.close();
            } catch (Exception e) {
              fail("exception while closing cq:", e);
            }
          }
        }
      }
    });

    client2.invokeAsync(new CacheSerializableRunnable("pub updates") {
      @Override
      public void run2() throws CacheException {
        Region region = CacheServerTestUtil.getCache().getRegion(regionName);
        while (true) {
          for (int i = 0; i < 50000; i++) {
            region.put("" + i, "" + Math.random());
          }
        }
      }
    });

    server.invokeAsync(new CacheSerializableRunnable("pub updates") {
      @Override
      public void run2() throws CacheException {
        Region region = CacheServerTestUtil.getCache().getRegion(regionName);
        while (true) {
          for (int i = 0; i < 50000; i++) {
            region.put("" + i, "" + Math.random());
          }
        }
      }
    });

    // client has another thread that retrieves cq map and checks stat over and over
    client.invoke(new CacheSerializableRunnable("Check Stats") {
      @Override
      public void run2() throws CacheException {
        for (int i = 0; i < 10000; i++) {
          checkCqStats(cq1Name);
        }
      }
    });

    client.invoke(() -> CacheServerTestUtil.closeCache());
    client2.invoke(() -> CacheServerTestUtil.closeCache());
    server.invoke(() -> CacheServerTestUtil.closeCache());
  }
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/CqDataUsingPoolDUnitTest.java