@Test
  public void testBug40165ClientFailsOver() {
    VM clientVM = getController();
    VM server1VM = vm0;
    VM server2VM = vm1;

    int serverPort1 = server1VM.invoke(() -> new ServerFactory()
        .evictionPolicy(HARegionQueue.HA_EVICTION_POLICY_MEMORY)
        .regionShortcut(RegionShortcut.REPLICATE)
        .create());

    // 1. Create two cache servers with slow dispatcher
    // 2. Start a durable client with a custom cache listener
    // 3. Do some puts on the server region
    // 4. Let the dispatcher start dispatching
    // 5. Wait till the durable client receives marker from its primary.
    // 6. Kill the primary server, so that the second one becomes primary.
    // 7. Wait till the durable client processes all its events.
    // 8. Verify that expected number of deltas are received by it.

    // Step 0
    for (VM vm : toArray(getController(), server1VM, server2VM)) {
      vm.invoke(this::prepareDeltas);
    }

    // Step 1
    for (VM serverVM : toArray(server1VM, server2VM)) {
      serverVM.invoke(() -> {
        System.setProperty(INTERNAL_FACTORY_PROPERTY, CACHE_CLIENT_PROXY_FACTORY_NAME);
        LATCH.set(new CountDownLatch(1));
      });
    }

    int serverPort2 = server2VM.invoke(() -> new ServerFactory()
        .evictionPolicy(HARegionQueue.HA_EVICTION_POLICY_MEMORY)
        .regionShortcut(RegionShortcut.REPLICATE)
        .create());

    // Step 2
    clientVM.invoke(() -> {
      String durableClientId = getName() + "_client";

      Properties props = new Properties();
      props.setProperty(LOCATORS, "");
      props.setProperty(DURABLE_CLIENT_ID, durableClientId);
      props.setProperty(DURABLE_CLIENT_TIMEOUT, String.valueOf(60));

      new ClientFactory().create(new ClientCacheFactory(props));

      Pool pool = PoolManager.createFactory()
          .addServer("localhost", serverPort1)
          .addServer("localhost", serverPort2)
          .setSubscriptionEnabled(true)
          .setSubscriptionAckInterval(1)
          .setSubscriptionRedundancy(2)
          .create(getName() + "_pool");

      ClientRegionFactory<String, Object> clientRegionFactory =
          getClientCache().createClientRegionFactory(ClientRegionShortcut.LOCAL);

      clientRegionFactory
          .addCacheListener(new DurableClientListener(errorCollector));

      clientRegionFactory
          .setConcurrencyChecksEnabled(false);

      clientRegionFactory
          .setPoolName(pool.getName());

      Region<String, Object> region = clientRegionFactory.create(regionName);

      region.registerInterest("ALL_KEYS");
      getClientCache().readyForEvents();
    });

    // Step 3
    server1VM.invoke(() -> {
      Region<String, Object> region = getCache().getRegion(regionName);
      for (int i = 0; i < 100; i++) {
        DeltaTestImpl value = new DeltaTestImpl();
        value.setStr(String.valueOf(i));
        region.put(DELTA_KEY, value);
      }
      region.put(LAST_KEY, "");
    });

    for (VM serverVM : toArray(server1VM, server2VM)) {
      serverVM.invoke(() -> LATCH.get().countDown());
    }

    // Step 5
    clientVM.invoke(() -> {
      InternalPool pool = (InternalPool) PoolManager.getAll().values().stream().findFirst().get();
      VM primaryVM = pool.getPrimaryPort() == serverPort1 ? server1VM : server2VM;

      await().until(MARKER_RECEIVED::get);

      // Step 6
      primaryVM.invoke(this::closeCache);

      // Step 7
      await().until(LAST_KEY_RECEIVED::get);

      // Step 8
      long fromDeltasOnClient = DeltaTestImpl.getFromDeltaInvokations();
      assertThat(fromDeltasOnClient).isGreaterThanOrEqualTo(99);
    });
  }
<pre>
1. Create a cache server with slow dispatcher
2. Start a durable client with a custom cache listener which shuts itself down as soon as it
   receives a marker message.
3. Do some puts on the server region
4. Let the dispatcher start dispatching
5. Verify that durable client is disconnected as soon as it processes the marker. Server will
   retain its queue which has some events (containing deltas) in it.
6. Restart the durable client without the self-destructing listener.
7. Wait till the durable client processes all its events.
8. Verify that no deltas are received by it.
</pre>
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/DeltaPropagationDUnitTest.java