@Test
  public void testRejectClientWhenDrainingCq() {
    try {
      IgnoredException.addIgnoredException(
          "CacheClientNotifier: Connection refused due to cq queue being drained from admin command, please wait...");
      IgnoredException.addIgnoredException(
          "Could not initialize a primary queue on startup. No queue servers available.");

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

      final String durableClientId = getName() + "_client";
      durableClientVM.invoke(CacheServerTestUtil::disableShufflingOfEndpoints);

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

      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("Set test hook", new CacheSerializableRunnable() {
        @Override
        public void run2() throws CacheException {
          // Set the Test Hook!
          // This test hook will pause during the drain process
          CacheClientProxy.testHook = new RejectClientReconnectTestHook();
        }
      });

      server1VM.invokeAsync("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);

      server1VM.invoke("verify was rejected at least once", new CacheSerializableRunnable() {
        @Override
        public void run2() throws CacheException {
          await()
              .until(() -> CacheClientProxy.testHook != null
                  && (((RejectClientReconnectTestHook) CacheClientProxy.testHook)
                      .wasClientRejected()));
          assertThat(
              ((RejectClientReconnectTestHook) CacheClientProxy.testHook).wasClientRejected())
                  .isTrue();
        }
      });

      checkPrimaryUpdater(durableClientVM);

      // After rejection, the client will retry and eventually connect
      // Verify durable client on server2
      verifyDurableClientPresent(DistributionConfig.DEFAULT_DURABLE_CLIENT_TIMEOUT, durableClientId,
          server1VM);

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

      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);
    } finally {
      server1VM.invoke("unset test hook", new CacheSerializableRunnable() {
        @Override
        public void run2() throws CacheException {
          CacheClientProxy.testHook = null;
        }
      });
    }
  }
Tests situation where a client is trying to reconnect while a cq is being drained. The client
should be rejected until no cqs are currently being drained
geode-cq/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/DurableClientCQDUnitTest.java