@Test
  public void testCqCloseExceptionDueToActivatingClient() throws Exception {
    try {
      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 server 1
      server1Port = server1VM.invoke(
          () -> createCacheServer(regionName, true));

      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);

      AsyncInvocation<Void> async =
          server1VM.invokeAsync("Close cq for durable client", new CacheSerializableRunnable() {
            @Override
            public void run2() throws CacheException {

              // Set the Test Hook!
              // This test hook will pause during the drain process
              CacheClientProxy.testHook = new CqExceptionDueToActivatingClientTestHook();

              final CacheClientNotifier ccnInstance = CacheClientNotifier.getInstance();
              final CacheClientProxy clientProxy = ccnInstance.getClientProxy(durableClientId);
              ClientProxyMembershipID proxyId = clientProxy.getProxyID();

              try {
                ccnInstance.closeClientCq(durableClientId, "All");
                fail("Should have thrown an exception due to activating client");
              } catch (CqException e) {
                String expected =
                    String.format(
                        "CacheClientProxy: Could not drain cq %s due to client proxy id %s reconnecting.",
                        "All", proxyId.getDurableId());
                if (!e.getMessage().equals(expected)) {
                  fail("Not the expected exception, was expecting "
                      + expected
                      + " instead of exception: " + e.getMessage());
                }
              }
            }
          });

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

      // Reregister 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);

      async.await();

      // verify cq listener events
      checkCqListenerEvents(durableClientVM, "GreaterThan5", 4 /* numEventsExpected */,
          /* numEventsToWaitFor */ 15/* secondsToWait */);
      checkCqListenerEvents(durableClientVM, "LessThan5", 5 /* numEventsExpected */,
          /* numEventsToWaitFor */ 15/* secondsToWait */);
      checkCqListenerEvents(durableClientVM, "All", 10 /* numEventsExpected */,
          /* numEventsToWaitFor */ 15/* 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);
    } finally {
      server1VM.invoke("unset test hook", new CacheSerializableRunnable() {
        @Override
        public void run2() throws CacheException {
          CacheClientProxy.testHook = null;
        }
      });
    }
  }
Tests scenario where close cq will throw an exception due to a client being reactivated
geode-cq/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/DurableClientCQDUnitTest.java