@Test
  public void testPRLoadRejection() throws Exception {
    final Host host = Host.getHost(0);
    final VM accessor = host.getVM(1);
    final VM ds1 = host.getVM(2);
    final String rName = getUniqueName();

    // Make sure the desired VMs will have a fresh DS. TODO: convert these from AsyncInvocation to
    // invoke
    accessor.invoke(JUnit4DistributedTestCase::disconnectFromDS);
    ds1.invoke(JUnit4DistributedTestCase::disconnectFromDS);

    ds1.invoke("establishcConnectivity", () -> {
      getSystem();
      createPR(rName, false);
    });

    accessor.invoke("establishcConnectivity", () -> {
      getSystem();
      createPR(rName, true);
    });

    final AtomicInteger expectedInvocations = new AtomicInteger(0);

    Integer ex = (Integer) accessor
        .invoke(new SerializableCallable("Invoke loader from accessor, non-critical") {
          @Override
          public Object call() throws Exception {
            Region<Integer, String> r = getCache().getRegion(rName);
            Integer k = 1;
            Integer expectedInvocations0 = expectedInvocations.getAndIncrement();
            assertEquals(k.toString(), r.get(k, expectedInvocations0)); // should load for new key
            assertTrue(r.containsKey(k));
            Integer expectedInvocations1 = expectedInvocations.get();
            assertEquals(k.toString(), r.get(k, expectedInvocations1)); // no load
            assertEquals(k.toString(), r.get(k, expectedInvocations1)); // no load
            return expectedInvocations1;
          }
        });
    expectedInvocations.set(ex);

    ex = (Integer) ds1
        .invoke(new SerializableCallable("Invoke loader from datastore, non-critical") {
          @Override
          public Object call() throws Exception {
            Region<Integer, String> r = getCache().getRegion(rName);
            Integer k = 2;
            Integer expectedInvocations1 = expectedInvocations.getAndIncrement();
            assertEquals(k.toString(), r.get(k, expectedInvocations1)); // should load for new key
            assertTrue(r.containsKey(k));
            Integer expectedInvocations2 = expectedInvocations.get();
            assertEquals(k.toString(), r.get(k, expectedInvocations2)); // no load
            assertEquals(k.toString(), r.get(k, expectedInvocations2)); // no load
            String oldVal = r.remove(k);
            assertFalse(r.containsKey(k));
            assertEquals(k.toString(), oldVal);
            return expectedInvocations2;
          }
        });
    expectedInvocations.set(ex);

    accessor.invoke(addExpectedException);
    ds1.invoke(addExpectedException);

    ex = (Integer) ds1
        .invoke(new SerializableCallable("Set critical state, assert local load behavior") {
          @Override
          public Object call() throws Exception {
            final OffHeapMemoryMonitor ohmm =
                ((InternalResourceManager) getCache().getResourceManager()).getOffHeapMonitor();
            final PartitionedRegion pr = (PartitionedRegion) getCache().getRegion(rName);
            final RegionAdvisor advisor = pr.getRegionAdvisor();

            pr.put("oh1", new byte[838860]);
            pr.put("oh3", new byte[157287]);

            WaitCriterion wc = new WaitCriterion() {
              @Override
              public String description() {
                return "verify critical state";
              }

              @Override
              public boolean done() {
                for (final ProxyBucketRegion bucket : advisor.getProxyBucketArray()) {
                  if (bucket.isBucketSick()) {
                    return true;
                  }
                }
                return false;
              }
            };
            Wait.waitForCriterion(wc, 30 * 1000, 10, true);

            final Integer k = 2; // reload with same key again and again
            final Integer expectedInvocations3 = expectedInvocations.getAndIncrement();
            assertEquals(k.toString(), pr.get(k, expectedInvocations3)); // load
            assertFalse(pr.containsKey(k));
            Integer expectedInvocations4 = expectedInvocations.getAndIncrement();
            assertEquals(k.toString(), pr.get(k, expectedInvocations4)); // load
            assertFalse(pr.containsKey(k));
            Integer expectedInvocations5 = expectedInvocations.get();
            assertEquals(k.toString(), pr.get(k, expectedInvocations5)); // load
            assertFalse(pr.containsKey(k));
            return expectedInvocations5;
          }
        });
    expectedInvocations.set(ex);

    ex = (Integer) accessor.invoke(new SerializableCallable(
        "During critical state on datastore, assert accesor load behavior") {
      @Override
      public Object call() throws Exception {
        final Integer k = 2; // reload with same key again and again
        Integer expectedInvocations6 = expectedInvocations.incrementAndGet();
        Region<Integer, String> r = getCache().getRegion(rName);
        assertEquals(k.toString(), r.get(k, expectedInvocations6)); // load
        assertFalse(r.containsKey(k));
        Integer expectedInvocations7 = expectedInvocations.incrementAndGet();
        assertEquals(k.toString(), r.get(k, expectedInvocations7)); // load
        assertFalse(r.containsKey(k));
        return expectedInvocations7;
      }
    });
    expectedInvocations.set(ex);

    ex = (Integer) ds1.invoke(
        new SerializableCallable("Set safe state on datastore, assert local load behavior") {
          @Override
          public Object call() throws Exception {
            final PartitionedRegion r = (PartitionedRegion) getCache().getRegion(rName);
            final OffHeapMemoryMonitor ohmm =
                ((InternalResourceManager) getCache().getResourceManager()).getOffHeapMonitor();

            r.destroy("oh3");
            WaitCriterion wc = new WaitCriterion() {
              @Override
              public String description() {
                return "verify critical state";
              }

              @Override
              public boolean done() {
                return !ohmm.getState().isCritical();
              }
            };
            Wait.waitForCriterion(wc, 30 * 1000, 10, true);

            Integer k = 3; // same key as previously used, this time is should stick
            Integer expectedInvocations8 = expectedInvocations.incrementAndGet();
            assertEquals(k.toString(), r.get(k, expectedInvocations8)); // last load for 3
            assertTrue(r.containsKey(k));
            return expectedInvocations8;
          }
        });
    expectedInvocations.set(ex);

    accessor.invoke(new SerializableCallable(
        "Data store in safe state, assert load behavior, accessor sets critical state, assert load behavior") {
      @Override
      public Object call() throws Exception {
        final OffHeapMemoryMonitor ohmm =
            ((InternalResourceManager) getCache().getResourceManager()).getOffHeapMonitor();
        assertFalse(ohmm.getState().isCritical());
        Integer k = 4;
        Integer expectedInvocations9 = expectedInvocations.incrementAndGet();
        final PartitionedRegion r = (PartitionedRegion) getCache().getRegion(rName);
        assertEquals(k.toString(), r.get(k, expectedInvocations9)); // load for 4
        assertTrue(r.containsKey(k));
        assertEquals(k.toString(), r.get(k, expectedInvocations9)); // no load

        // Go critical in accessor by creating entries in local node
        String localRegionName = "localRegionName";
        AttributesFactory<Integer, String> af = getLocalRegionAttributesFactory();
        final LocalRegion localRegion =
            (LocalRegion) getCache().createRegion(localRegionName, af.create());
        localRegion.put("oh1", new byte[838860]);
        localRegion.put("oh3", new byte[157287]);

        WaitCriterion wc = new WaitCriterion() {
          @Override
          public String description() {
            return "verify critical state";
          }

          @Override
          public boolean done() {
            return ohmm.getState().isCritical();
          }
        };
        Wait.waitForCriterion(wc, 30 * 1000, 10, true);

        k = 5;
        Integer expectedInvocations10 = expectedInvocations.incrementAndGet();
        assertEquals(k.toString(), r.get(k, expectedInvocations10)); // load for key 5
        assertTrue(r.containsKey(k));
        assertEquals(k.toString(), r.get(k, expectedInvocations10)); // no load

        // Clean up critical state
        localRegion.destroy("oh3");
        wc = new WaitCriterion() {
          @Override
          public String description() {
            return "verify critical state";
          }

          @Override
          public boolean done() {
            return !ohmm.getState().isCritical();
          }
        };
        Wait.waitForCriterion(wc, 30 * 1000, 10, true);

        return expectedInvocations10;
      }
    });

    accessor.invoke(removeExpectedException);
    ds1.invoke(removeExpectedException);
  }
Test that a Partitioned Region loader invocation is rejected if the VM with the bucket is in a
critical state.
geode-core/src/distributedTest/java/org/apache/geode/cache/management/MemoryThresholdsOffHeapDUnitTest.java