private void doDistributedRegionRemotePutRejection(boolean localDestroy, boolean cacheClose)
      throws Exception {
    final Host host = Host.getHost(0);
    final VM server1 = host.getVM(0);
    final VM server2 = host.getVM(1);

    final String regionName = "offHeapDRRemotePutRejection";

    // set port to 0 in-order for system to pickup a random port.
    startCacheServer(server1, 0f, 0f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);
    startCacheServer(server2, 0f, 90f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);

    registerTestMemoryThresholdListener(server1);
    registerTestMemoryThresholdListener(server2);

    doPuts(server1, regionName, false/* catchRejectedException */,
        false/* catchLowMemoryException */);
    doPutAlls(server1, regionName, false/* catchRejectedException */,
        false/* catchLowMemoryException */, Range.DEFAULT);

    // make server2 critical
    setUsageAboveCriticalThreshold(server2, regionName);

    verifyListenerValue(server1, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server2, MemoryState.CRITICAL, 1, true);

    // make sure that local server1 puts are rejected
    doPuts(server1, regionName, false/* catchRejectedException */,
        true/* catchLowMemoryException */);
    Range r1 = new Range(Range.DEFAULT, Range.DEFAULT.width() + 1);
    doPutAlls(server1, regionName, false/* catchRejectedException */,
        true/* catchLowMemoryException */, r1);

    if (localDestroy) {
      // local destroy the region on sick member
      server2.invoke(new SerializableCallable("local destroy") {
        @Override
        public Object call() throws Exception {
          Region r = getRootRegion().getSubregion(regionName);
          r.localDestroyRegion();
          return null;
        }
      });
    } else if (cacheClose) {
      server2.invoke(new SerializableCallable() {
        @Override
        public Object call() throws Exception {
          getCache().close();
          return null;
        }
      });
    } else {
      setUsageBelowEviction(server2, regionName);
    }

    // wait for remote region destroyed message to be processed
    server1.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        WaitCriterion wc = new WaitCriterion() {
          @Override
          public String description() {
            return "remote localRegionDestroyed message not received";
          }

          @Override
          public boolean done() {
            DistributedRegion dr = (DistributedRegion) getRootRegion().getSubregion(regionName);
            return dr.getAtomicThresholdInfo().getMembersThatReachedThreshold().size() == 0;
          }
        };
        Wait.waitForCriterion(wc, 10000, 10, true);
        return null;
      }
    });

    // make sure puts succeed
    doPuts(server1, regionName, false/* catchRejectedException */,
        false/* catchLowMemoryException */);
    Range r2 = new Range(r1, r1.width() + 1);
    doPutAlls(server1, regionName, false/* catchRejectedException */,
        false/* catchLowMemoryException */, r2);
  }
test that puts in a server are rejected when a remote VM crosses critical threshold
geode-core/src/distributedTest/java/org/apache/geode/cache/management/MemoryThresholdsOffHeapDUnitTest.java