private void mutateCQAttributes(VM vm, final String cqName, final int mutator_function)
      throws Exception {
    vm.invoke(new CacheSerializableRunnable("Stop CQ :" + cqName) {
      @Override
      public void run2() throws CacheException {
        CqQuery cq1 = null;
        LogWriterUtils.getLogWriter().info("### CQ attributes mutator for ###" + cqName);
        // Get CQ Service.
        QueryService cqService = null;
        try {
          cqService = getCache().getQueryService();
        } catch (Exception cqe) {
          Assert.fail("Failed to getCQService.", cqe);
        }

        // Get CQ.
        try {
          cq1 = cqService.getCq(cqName);
        } catch (Exception ex) {
          Assert.fail("Failed to get CQ " + cqName + " . ", ex);
        }
        CqAttributesMutator cqAttrMutator = cq1.getCqAttributesMutator();
        CqAttributes cqAttr = cq1.getCqAttributes();
        CqListener[] cqListeners;
        switch (mutator_function) {
          case CREATE:
            // Reinitialize with 2 CQ Listeners
            CqListener[] cqListenersArray = {new CqQueryTestListener(getCache().getLogger()),
                new CqQueryTestListener(getCache().getLogger())};
            cqAttrMutator.initCqListeners(cqListenersArray);
            cqListeners = cqAttr.getCqListeners();
            assertEquals("CqListener count mismatch", cqListeners.length, 2);
            break;

          case UPDATE:
            // Add 2 new CQ Listeners
            CqListener newListener1 = new CqQueryTestListener(getCache().getLogger());
            CqListener newListener2 = new CqQueryTestListener(getCache().getLogger());
            cqAttrMutator.addCqListener(newListener1);
            cqAttrMutator.addCqListener(newListener2);

            cqListeners = cqAttr.getCqListeners();
            assertEquals("CqListener count mismatch", cqListeners.length, 3);
            break;

          case DESTROY:
            cqListeners = cqAttr.getCqListeners();
            cqAttrMutator.removeCqListener(cqListeners[0]);
            cqListeners = cqAttr.getCqListeners();
            assertEquals("CqListener count mismatch", cqListeners.length, 2);

            // Remove a listener and validate
            cqAttrMutator.removeCqListener(cqListeners[0]);
            cqListeners = cqAttr.getCqListeners();
            assertEquals("CqListener count mismatch", cqListeners.length, 1);
            break;
        }
      }
    });
  }
Waits till the CQ state is same as the expected. Waits for max time, if the CQ state is not
same as expected throws exception.
geode-cq/src/distributedTest/java/org/apache/geode/cache/query/cq/dunit/CqQueryUsingPoolDUnitTest.java