private void doRetriedOperation(final Operation op, boolean usePR) {
    Host host = Host.getHost(0);
    final VM server1 = host.getVM(0);
    final VM server2 = host.getVM(1);
    final VM client = host.getVM(2);
    final int port1 = createRegionsAndStartServer(server1, true);
    final int port2 = createRegionsAndStartServer(server2, true);
    final String regionName = usePR ? PR_REG_NAME : REP_REG_NAME;

    IgnoredException.addIgnoredException("java.net.SocketException");

    createClientRegion(client, port1, false, port2);

    SerializableCallable getID = new SerializableCallable("get DM ID") {
      @Override
      public Object call() {
        return getSystem().getDistributedMember();
      }
    };

    final InternalDistributedMember server1ID = (InternalDistributedMember) server1.invoke(getID);
    final InternalDistributedMember server2ID = (InternalDistributedMember) server2.invoke(getID);

    Set<IgnoredException> exceptions = new HashSet<>();
    exceptions.add(IgnoredException.addIgnoredException("Membership: requesting removal", server1));
    exceptions.add(IgnoredException.addIgnoredException("Membership: requesting removal", server2));
    exceptions.add(IgnoredException
        .addIgnoredException(ForcedDisconnectException.class.getSimpleName(), server1));
    exceptions.add(IgnoredException
        .addIgnoredException(ForcedDisconnectException.class.getSimpleName(), server2));
    exceptions.add(IgnoredException
        .addIgnoredException(MemberDisconnectedException.class.getSimpleName(), server1));
    exceptions.add(IgnoredException
        .addIgnoredException(MemberDisconnectedException.class.getSimpleName(), server2));

    try {

      server1.invoke(new SerializableCallable("install crasher in server1") {
        @Override
        public Object call() throws Exception {
          Region r = getCache().getRegion(regionName);
          r.put("key0", "value");
          if (op == Operation.PUT_IF_ABSENT) {
            r.destroy("key0");
          }
          // force client to use server1 for now
          // getCache().getCacheServers().get(0).stop();
          r.getAttributesMutator().addCacheListener(new CacheListenerAdapter() {
            private void killSender(EntryEvent event) {
              if (event.isOriginRemote()) {
                Distribution mgr =
                    MembershipManagerHelper.getDistribution(getSystem());
                mgr.requestMemberRemoval(server2ID, "removing for test");
                try {
                  mgr.waitForDeparture(server2ID);
                } catch (Exception e) {
                  fail("failed to stop the other server for this test:" + e.getMessage());
                }
              }
            }

            @Override
            public void afterCreate(EntryEvent event) {
              getCache().getLogger().info("afterCreate invoked with " + event);
              killSender(event);
            }

            @Override
            public void afterUpdate(EntryEvent event) {
              getCache().getLogger().info("afterUpdate invoked with " + event);
              killSender(event);
            }

            @Override
            public void afterDestroy(EntryEvent event) {
              getCache().getLogger().info("afterDestroy invoked with " + event);
              killSender(event);
            }
          });
          return null;
        }
      });

      server2.invoke(new SerializableCallable("install crasher in server2") {
        @Override
        public Object call() throws Exception {
          Region r = getCache().getRegion(regionName);
          // force client to use server1 for now
          // getCache().getCacheServers().get(0).stop();
          r.getAttributesMutator().addCacheListener(new CacheListenerAdapter() {
            private void killSender(EntryEvent event) {
              if (event.isOriginRemote()) {
                Distribution mgr =
                    MembershipManagerHelper.getDistribution(getSystem());
                mgr.requestMemberRemoval(server1ID, "removing for test");
                try {
                  mgr.waitForDeparture(server1ID);
                } catch (Exception e) {
                  fail("failed to stop the other server for this test:" + e.getMessage());
                }
              }
            }

            @Override
            public void afterCreate(EntryEvent event) {
              getCache().getLogger().info("afterCreate invoked with " + event);
              killSender(event);
            }

            @Override
            public void afterUpdate(EntryEvent event) {
              getCache().getLogger().info("afterUpdate invoked with " + event);
              killSender(event);
            }

            @Override
            public void afterDestroy(EntryEvent event) {
              getCache().getLogger().info("afterDestroy invoked with " + event);
              killSender(event);
            }
          });
          return null;
        }
      });


      client.invoke(new SerializableRunnable() {
        @Override
        public void run() {
          GemFireCacheImpl cache = (GemFireCacheImpl) getCache();
          Region r = cache.getRegion(regionName);
          if (op == Operation.PUT_IF_ABSENT) {
            assertTrue("expected putIfAbsent to succeed and return null",
                r.putIfAbsent("key0", "newvalue") == null);
          } else if (op == Operation.REMOVE) {
            assertTrue("expected remove operation to succeed and return true",
                r.remove("key0", "value"));
          } else if (op == Operation.REPLACE) {
            assertTrue("expected replace operation to succeed and return true",
                r.replace("key0", "value", "newvalue"));
          }
        }
      });
    } finally {
      disconnectAllFromDS();
      for (IgnoredException ex : exceptions) {
        ex.remove();
      }
    }
  }
test putIfAbsent with failover & retry. This is bugs 42559 and 43640
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/ConcurrentMapOpsDUnitTest.java