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 server1 = host.getVM(1);
    final VM server2 = host.getVM(2);
    final VM server3 = host.getVM(3);

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

    final ServerPorts ports1 = startCacheServer(server1, 80f, 90f, regionName, true/* createPR */,
        false/* notifyBySubscription */, redundancy);
    ServerPorts ports2 = startCacheServer(server2, 80f, 90f, regionName, true/* createPR */,
        false/* notifyBySubscription */, redundancy);
    ServerPorts ports3 = startCacheServer(server3, 80f, 90f, regionName, true/* createPR */,
        false/* notifyBySubscription */, redundancy);
    accessor.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        getSystem(getServerProperties());
        getCache();
        AttributesFactory factory = new AttributesFactory();
        PartitionAttributesFactory paf = new PartitionAttributesFactory();
        paf.setRedundantCopies(redundancy);
        paf.setLocalMaxMemory(0);
        paf.setTotalNumBuckets(11);
        factory.setPartitionAttributes(paf.create());
        createRegion(regionName, factory.create());
        return null;
      }
    });

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

    SerializableCallable getMyId = new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        return getCache().getMyId();
      }
    };

    final DistributedMember server1Id = (DistributedMember) server1.invoke(getMyId);

    setUsageAboveCriticalThreshold(server1);

    accessor.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final PartitionedRegion pr = (PartitionedRegion) getRootRegion().getSubregion(regionName);
        final String regionPath = getRootRegion().getSubregion(regionName).getFullPath();
        // server1 is sick, look for a key on server1, and attempt put again
        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 = getHashKey(pr, null, key, null, null);
              Set<InternalDistributedMember> owners = pr.getRegionAdvisor().getBucketOwners(hKey);
              if (owners.contains(server1Id)) {
                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;
          }
        };
        GeodeAwaitility.await().untilAsserted(wc);
        return null;
      }
    });

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

    if (localDestroy) {
      // local destroy the region on sick member
      server1.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 member
      server1.invoke(new SerializableCallable("close cache sick member") {
        @Override
        public Object call() throws Exception {
          getCache().close();
          return null;
        }
      });
    } else {
      setUsageBelowEviction(server1);
    }

    // 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;
          }
        };
        GeodeAwaitility.await().untilAsserted(wc);
        return null;
      }
    });
    doPutAlls(accessor, regionName, false, false, r1);
  }
Make sure appropriate events are delivered when moving between states.
geode-core/src/distributedTest/java/org/apache/geode/cache/management/MemoryThresholdsDUnitTest.java