private void prRemotePutRejection(boolean cacheClose, boolean localDestroy, final boolean useTx)
      throws Exception {
    final Host host = Host.getHost(0);
    final VM accessor = host.getVM(0);
    final VM[] servers = new VM[3];
    servers[0] = host.getVM(1);
    servers[1] = host.getVM(2);
    servers[2] = host.getVM(3);

    final String regionName = "offHeapPRRemotePutRejection";
    final int redundancy = 1;

    startCacheServer(servers[0], 0f, 90f, regionName, true/* createPR */,
        false/* notifyBySubscription */, redundancy);
    startCacheServer(servers[1], 0f, 90f, regionName, true/* createPR */,
        false/* notifyBySubscription */, redundancy);
    startCacheServer(servers[2], 0f, 90f, regionName, true/* createPR */,
        false/* notifyBySubscription */, redundancy);
    accessor.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        getSystem(getOffHeapProperties());
        getCache();
        AttributesFactory factory = new AttributesFactory();
        PartitionAttributesFactory paf = new PartitionAttributesFactory();
        paf.setRedundantCopies(redundancy);
        paf.setLocalMaxMemory(0);
        paf.setTotalNumBuckets(11);
        factory.setPartitionAttributes(paf.create());
        factory.setOffHeap(true);
        createRegion(regionName, factory.create());
        return null;
      }
    });

    doPuts(accessor, regionName, false, false);
    final Range r1 = Range.DEFAULT;
    doPutAlls(accessor, regionName, false, false, r1);

    servers[0].invoke(addExpectedException);
    servers[1].invoke(addExpectedException);
    servers[2].invoke(addExpectedException);
    setUsageAboveCriticalThreshold(servers[0], regionName);

    final Set<InternalDistributedMember> criticalMembers =
        (Set) servers[0].invoke(new SerializableCallable() {
          @Override
          public Object call() throws Exception {
            final PartitionedRegion pr =
                (PartitionedRegion) getRootRegion().getSubregion(regionName);
            final int hashKey = PartitionedRegionHelper.getHashKey(pr, null, "oh5", null, null);
            return pr.getRegionAdvisor().getBucketOwners(hashKey);
          }
        });

    accessor.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final PartitionedRegion pr = (PartitionedRegion) getRootRegion().getSubregion(regionName);
        WaitCriterion wc = new WaitCriterion() {
          @Override
          public String description() {
            return "remote bucket not marked sick";
          }

          @Override
          public boolean done() {
            boolean keyFoundOnSickMember = false;
            boolean caughtException = false;
            for (int i = 0; i < 20; i++) {
              Integer key = i;
              int hKey = PartitionedRegionHelper.getHashKey(pr, null, key, null, null);
              Set<InternalDistributedMember> owners = pr.getRegionAdvisor().getBucketOwners(hKey);
              final boolean hasCriticalOwners = owners.removeAll(criticalMembers);
              if (hasCriticalOwners) {
                keyFoundOnSickMember = true;
                try {
                  if (useTx) {
                    getCache().getCacheTransactionManager().begin();
                  }
                  pr.getCache().getLogger().fine("SWAP:putting in tx:" + useTx);
                  pr.put(key, "value");
                  if (useTx) {
                    getCache().getCacheTransactionManager().commit();
                  }
                } catch (LowMemoryException ex) {
                  caughtException = true;
                  if (useTx) {
                    getCache().getCacheTransactionManager().rollback();
                  }
                }
              } else {
                // puts on healthy member should continue
                pr.put(key, "value");
              }
            }
            return keyFoundOnSickMember && caughtException;
          }
        };
        Wait.waitForCriterion(wc, 10000, 10, true);
        return null;
      }
    });

    {
      Range r2 = new Range(r1, r1.width() + 1);
      doPutAlls(accessor, regionName, false, true, r2);
    }

    // Find all VMs that have a critical region
    SerializableCallable getMyId = new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        return getCache().getMyId();
      }
    };
    final Set<VM> criticalServers = new HashSet<>();
    for (final VM server : servers) {
      DistributedMember member = (DistributedMember) server.invoke(getMyId);
      if (criticalMembers.contains(member)) {
        criticalServers.add(server);
      }
    }

    if (localDestroy) {
      // local destroy the region on sick members
      for (final VM vm : criticalServers) {
        vm.invoke(new SerializableCallable("local destroy sick member") {
          @Override
          public Object call() throws Exception {
            Region r = getRootRegion().getSubregion(regionName);
            LogWriterUtils.getLogWriter().info("PRLocalDestroy");
            r.localDestroyRegion();
            return null;
          }
        });
      }
    } else if (cacheClose) {
      // close cache on sick members
      for (final VM vm : criticalServers) {
        vm.invoke(new SerializableCallable("close cache sick member") {
          @Override
          public Object call() throws Exception {
            getCache().close();
            return null;
          }
        });
      }
    } else {
      setUsageBelowEviction(servers[0], regionName);
      servers[0].invoke(removeExpectedException);
      servers[1].invoke(removeExpectedException);
      servers[2].invoke(removeExpectedException);
    }

    // do put all in a loop to allow distribution of message
    accessor.invoke(new SerializableCallable("Put in a loop") {
      @Override
      public Object call() throws Exception {
        final Region r = getRootRegion().getSubregion(regionName);
        WaitCriterion wc = new WaitCriterion() {
          @Override
          public String description() {
            return "pr should have gone un-critical";
          }

          @Override
          public boolean done() {
            boolean done = true;
            for (int i = 0; i < 20; i++) {
              try {
                r.put(i, "value");
              } catch (LowMemoryException e) {
                // expected
                done = false;
              }
            }
            return done;
          }
        };
        Wait.waitForCriterion(wc, 10000, 10, true);
        return null;
      }
    });
    doPutAlls(accessor, regionName, false, false, r1);
  }
Test that DistributedRegion cacheLoade and netLoad are passed through to the calling thread if
the local VM is in a critical state. Once the VM has moved to a safe state then test that they
are allowed.
geode-core/src/distributedTest/java/org/apache/geode/cache/management/MemoryThresholdsOffHeapDUnitTest.java