private void durableFailoverAfterReconnect(int redundancyLevel) {
    // Start server 1
    server1Port = server1VM.invoke(() -> createCacheServer(regionName, true));

    // start server 2
    final int server2Port = server2VM.invoke(() -> createCacheServer(regionName, true));

    // Start a durable client
    durableClientId = getName() + "_client";

    Pool clientPool;
    if (redundancyLevel == 1) {
      clientPool = getClientPool(NetworkUtils.getServerHostName(), server1Port,
          server2Port, true);
    } else {
      clientPool = getClientPool(NetworkUtils.getServerHostName(), server1Port,
          server2Port, true, 0);
    }

    durableClientVM.invoke(CacheServerTestUtil::disableShufflingOfEndpoints);
    // Create the durable client cache (type = cache)
    durableClientVM.invoke(() -> createCacheClient(clientPool, regionName,
        getClientDistributedSystemProperties(durableClientId, VERY_LONG_DURABLE_TIMEOUT_SECONDS),
        true));

    // Register interest in all entries in the region
    registerInterest(durableClientVM, regionName, true, InterestResultPolicy.NONE);

    // Send clientReady message
    durableClientVM.invoke("Send clientReady", new CacheSerializableRunnable() {
      @Override
      public void run2() throws CacheException {
        getCache().readyForEvents();
      }
    });

    // Verify the durable client is connected to the servers.
    verifyDurableClientPresent(VERY_LONG_DURABLE_TIMEOUT_SECONDS, durableClientId, server1VM);
    if (redundancyLevel == 1) {
      verifyDurableClientPresent(VERY_LONG_DURABLE_TIMEOUT_SECONDS, durableClientId, server2VM);
    }
    // Start normal publisher client
    publisherClientVM.invoke(() -> createCacheClient(
        getClientPool(NetworkUtils.getServerHostName(), server1Port,
            server2Port, false),
        regionName));

    // put key 0
    publishEntries(publisherClientVM, 0, 1);

    // Verify the durable client received key 0
    checkListenerEvents(1, 1, -1, durableClientVM);

    // Wait to until the HARegionQueue has emptied before disconnecting the Durable client
    // to avoid having left over messages that processed, in particular, the key 0 message
    waitUntilHARegionQueueSizeIsZero(server1VM);
    if (redundancyLevel == 1) {
      waitUntilHARegionQueueSizeIsZero(server2VM);
    }

    // Stop the durable client
    disconnectDurableClient(true);

    // Stop server 1 - publisher will put 1 entries during shutdown/primary identification
    server1VM.invoke((SerializableRunnableIF) CacheServerTestUtil::closeCache);

    // Put key 1 during client downtime
    publishEntries(publisherClientVM, 1, 1);

    // Re-start the durable client that is kept alive on the server
    restartDurableClient(VERY_LONG_DURABLE_TIMEOUT_SECONDS, clientPool, true);

    // Re-register interest in all entries in the region
    registerInterest(durableClientVM, regionName, true, InterestResultPolicy.NONE);

    // Send clientReady message
    sendClientReady(durableClientVM);

    // Put keys 2 and 3
    publishEntries(publisherClientVM, 2, 2);

    // Verify the durable client received the updates before failover
    if (redundancyLevel == 1) {
      checkListenerEvents(3, 1, -1, durableClientVM);
    } else {
      checkListenerEvents(1, 1, -1, durableClientVM);
    }

    // Verify that key 0 is not present
    durableClientVM.invoke("Get", () -> await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> {
      Region<Object, Object> region = getCache().getRegion(regionName);
      assertThat(region).isNotNull();

      assertThat(region.getEntry("0")).isNull();
    }));

    // put key 4
    publishEntries(publisherClientVM, 4, 1);

    // Verify the durable client received the updates after failover
    if (redundancyLevel == 1) {
      checkListenerEvents(4, 1, -1, durableClientVM);
    } else {
      checkListenerEvents(2, 1, -1, durableClientVM);
    }

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

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

    // Stop server 2
    server2VM.invoke((SerializableRunnableIF) CacheServerTestUtil::closeCache);
  }
wait unit we have the required number of events in the queue
geode-cq/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/DurableClientTestCase.java