@Test
  public void testDRPutWithInterrupt() throws Throwable {
    Host host = Host.getHost(0);
    final VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);

    createRegion(vm0);

    // put some data in vm0
    createData(vm0, 0, 10, "a");

    final SerializableCallable interruptTask = new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        puttingThread.interrupt();
        return null;
      }
    };

    vm1.invoke(new SerializableCallable() {

      @Override
      public Object call() throws Exception {
        disconnectFromDS();
        DistributionMessageObserver.setInstance(new DistributionMessageObserver() {

          @Override
          public void beforeProcessMessage(ClusterDistributionManager dm,
              DistributionMessage message) {
            if (message instanceof UpdateMessage
                && ((UpdateMessage) message).regionPath.contains("region")
                && doInterrupt.compareAndSet(true, false)) {
              vm0.invoke(interruptTask);
              DistributionMessageObserver.setInstance(null);
            }
          }

        });
        return null;
      }
    });

    createRegion(vm1);

    SerializableCallable doPuts = new SerializableCallable() {

      @Override
      public Object call() throws Exception {
        puttingThread = Thread.currentThread();
        Region<Object, Object> region = getCache().getRegion("region");
        long value = 0;
        long end = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(MAX_WAIT);
        while (!Thread.currentThread().isInterrupted()) {
          region.put(0, value);
          if (System.nanoTime() > end) {
            fail("Did not get interrupted in 60 seconds");
          }
        }
        return null;
      }
    };

    AsyncInvocation<Void> async0 = vm0.invokeAsync(doPuts);

    vm1.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        doInterrupt.set(true);
        return null;
      }
    });

    // vm0.invoke(new SerializableCallable() {
    //
    // @Override
    // public Object call() throws Exception {
    // long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(MAX_WAIT);
    // while(puttingThread == null) {
    // Thread.sleep(50);
    // if(System.nanoTime() > end) {
    // fail("Putting thread not set in 60 seconds");
    // }
    // }
    //
    // puttingThread.interrupt();
    // return null;
    // }
    // });

    async0.getResult();

    Object value0 = checkCacheAndGetValue(vm0);
    Object value1 = checkCacheAndGetValue(vm1);

    assertEquals(value0, value1);

  }
A simple test case that we are actually persisting with a PR.
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/InterruptsDUnitTest.java