@Test
  public void testCloseCqAndDrainEvents() {
    String greaterThan5Query = "select * from " + SEPARATOR + regionName + " p where p.ID > 5";
    String allQuery = "select * from " + SEPARATOR + regionName + " p where p.ID > -1";
    String lessThan5Query = "select * from " + SEPARATOR + regionName + " p where p.ID < 5";

    // Start a server
    server1Port = server1VM.invoke(() -> createCacheServer(regionName, true));

    // Start a durable client that is kept alive on the server when it stops
    // normally
    durableClientId = getName() + "_client";
    startDurableClient(durableClientVM, durableClientId, server1Port, regionName);
    // register durable cqs
    createCq(durableClientVM, "GreaterThan5", greaterThan5Query, true);
    createCq(durableClientVM, "All", allQuery, true);
    createCq(durableClientVM, "LessThan5", lessThan5Query, true);
    // send client ready
    sendClientReady(durableClientVM);

    // Verify durable client on server
    verifyDurableClientPresent(DistributionConfig.DEFAULT_DURABLE_CLIENT_TIMEOUT, durableClientId,
        server1VM);

    // Stop the durable client
    disconnectDurableClient(true);

    // Start normal publisher client
    startClient(publisherClientVM, server1Port, regionName);

    // Publish some entries
    publishEntries(publisherClientVM, regionName, 10);

    server1VM.invoke("Close cq for durable client", new CacheSerializableRunnable() {
      @Override
      public void run2() throws CacheException {

        final CacheClientNotifier ccnInstance = CacheClientNotifier.getInstance();

        try {
          ccnInstance.closeClientCq(durableClientId, "All");
        } catch (CqException e) {
          fail("failed", e);
        }
      }
    });

    // Restart the durable client
    startDurableClient(durableClientVM, durableClientId, server1Port, regionName);

    // Re-register durable cqs
    createCq(durableClientVM, "GreaterThan5",
        "select * from " + SEPARATOR + regionName + " p where p.ID > 5",
        true);
    createCq(durableClientVM, "All",
        "select * from " + SEPARATOR + regionName + " p where p.ID > -1", true);
    createCq(durableClientVM, "LessThan5",
        "select * from " + SEPARATOR + regionName + " p where p.ID < 5",
        true);
    // send client ready
    sendClientReady(durableClientVM);

    // verify cq events for all 3 cqs
    checkCqListenerEvents(durableClientVM, "GreaterThan5", 4 /* numEventsExpected */,
        /* numEventsToWaitFor */ 15/* secondsToWait */);
    checkCqListenerEvents(durableClientVM, "LessThan5", 5 /* numEventsExpected */,
        /* numEventsToWaitFor */ 15/* secondsToWait */);
    checkCqListenerEvents(durableClientVM, "All", 0 /* numEventsExpected */,
        /* numEventsToWaitFor */ 5/* secondsToWait */);

    // Stop the durable client
    durableClientVM.invoke((SerializableRunnableIF) CacheServerTestUtil::closeCache);

    // Stop the publisher client
    publisherClientVM.invoke((SerializableRunnableIF) CacheServerTestUtil::closeCache);

    // Stop the server
    server1VM.invoke((SerializableRunnableIF) CacheServerTestUtil::closeCache);
  }
Test functionality to close the cq and drain all events from the ha queue from the server
geode-cq/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/DurableClientCQDUnitTest.java