private void validateCQStats(VM vm, final String cqName, final int creates, final int updates,
      final int deletes, final int totalEvents, final int cqListenerInvocations) {
    vm.invoke(new CacheSerializableRunnable("Validate CQs") {
      @Override
      public void run2() throws CacheException {
        LogWriterUtils.getLogWriter().info("### Validating CQ Stats. ### " + cqName);
        // Get CQ Service.
        QueryService qService = null;
        try {
          qService = getCache().getQueryService();
        } catch (Exception cqe) {
          cqe.printStackTrace();
          fail("Failed to get query service.");
        }

        CqService cqService = null;
        try {
          cqService = ((DefaultQueryService) qService).getCqService();
        } catch (CqException e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
          fail("Failed to get CqService, CQ : " + cqName);
        }
        Collection<? extends InternalCqQuery> cqs = cqService.getAllCqs();
        if (cqs.size() == 0) {
          fail("Failed to get CqQuery for CQ : " + cqName);
        }
        CqQueryImpl cQuery = (CqQueryImpl) cqs.iterator().next();

        CqStatistics cqStats = cQuery.getStatistics();
        CqQueryVsdStats cqVsdStats = cQuery.getVsdStats();
        if (cqStats == null || cqVsdStats == null) {
          fail("Failed to get CqQuery Stats for CQ : " + cqName);
        }

        getCache().getLogger()
            .info("#### CQ stats for " + cQuery.getName() + ": " + " Events Total: "
                + cqStats.numEvents() + " Events Created: " + cqStats.numInserts()
                + " Events Updated: " + cqStats.numUpdates() + " Events Deleted: "
                + cqStats.numDeletes() + " CQ Listener invocations: "
                + cqVsdStats.getNumCqListenerInvocations() + " Initial results time (nano sec): "
                + cqVsdStats.getCqInitialResultsTime());


        // Check for totalEvents count.
        if (totalEvents != CqQueryUsingPoolDUnitTest.noTest) {
          // Result size validation.
          assertEquals("Total Event Count mismatch", totalEvents, cqStats.numEvents());
        }

        // Check for create count.
        if (creates != CqQueryUsingPoolDUnitTest.noTest) {
          assertEquals("Create Event mismatch", creates, cqStats.numInserts());
        }

        // Check for update count.
        if (updates != CqQueryUsingPoolDUnitTest.noTest) {
          assertEquals("Update Event mismatch", updates, cqStats.numUpdates());
        }

        // Check for delete count.
        if (deletes != CqQueryUsingPoolDUnitTest.noTest) {
          assertEquals("Delete Event mismatch", deletes, cqStats.numDeletes());
        }

        // Check for CQ listener invocations.
        if (cqListenerInvocations != CqQueryUsingPoolDUnitTest.noTest) {
          assertEquals("CQ Listener invocations mismatch", cqListenerInvocations,
              cqVsdStats.getNumCqListenerInvocations());
        }
        //// Check for initial results time.
        // if (initialResultsTime != CqQueryUsingPoolDUnitTest.noTest &&
        //// cqVsdStats.getCqInitialResultsTime() <= 0) {
        // assertIndexDetailsEquals("Initial results time mismatch", initialResultsTime,
        //// cqVsdStats.getCqInitialResultsTime());
        // }
      }
    });
  }
This class tests the ContinuousQuery mechanism in GemFire. This includes the test with different
data activities.
geode-cq/src/distributedTest/java/org/apache/geode/cache/query/cq/dunit/CqStatsUsingPoolDUnitTest.java