@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();
    final float criticalHeapThresh = 0.90f;
    final int fakeHeapMaxSize = 1000;

    // Make sure the desired VMs will have a fresh DS.
    AsyncInvocation<Void> d0 = accessor.invokeAsync(JUnit4DistributedTestCase::disconnectFromDS);
    AsyncInvocation<Void> d1 = ds1.invokeAsync(JUnit4DistributedTestCase::disconnectFromDS);
    d0.join();
    assertFalse(d0.exceptionOccurred());
    d1.join();
    assertFalse(d1.exceptionOccurred());
    CacheSerializableRunnable establishConnectivity =
        new CacheSerializableRunnable("establishcConnectivity") {
          @Override
          public void run2() throws CacheException {
            getSystem();
          }
        };
    ds1.invoke(establishConnectivity);
    accessor.invoke(establishConnectivity);

    ds1.invoke(createPR(rName, false, fakeHeapMaxSize, criticalHeapThresh));
    accessor.invoke(createPR(rName, true, fakeHeapMaxSize, criticalHeapThresh));

    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 {
            long newfakeHeapUsage = Math.round(fakeHeapMaxSize * (criticalHeapThresh + 0.1f)); // usage
                                                                                               // above
                                                                                               // critical
                                                                                               // by
                                                                                               // 10%
            assertTrue(newfakeHeapUsage > 0);
            assertTrue(newfakeHeapUsage <= fakeHeapMaxSize);
            HeapMemoryMonitor hmm =
                ((InternalResourceManager) getCache().getResourceManager()).getHeapMonitor();
            hmm.updateStateAndSendEvent(newfakeHeapUsage, "test");
            assertTrue(hmm.getState().isCritical());
            final Integer k = 2; // reload with same key again and again
            final Integer expectedInvocations3 = expectedInvocations.getAndIncrement();
            Region<Integer, String> r = getCache().getRegion(rName);
            assertEquals(k.toString(), r.get(k, expectedInvocations3)); // load
            assertFalse(r.containsKey(k));
            Integer expectedInvocations4 = expectedInvocations.getAndIncrement();
            assertEquals(k.toString(), r.get(k, expectedInvocations4)); // load
            assertFalse(r.containsKey(k));
            Integer expectedInvocations5 = expectedInvocations.get();
            assertEquals(k.toString(), r.get(k, expectedInvocations5)); // load
            assertFalse(r.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 {
            HeapMemoryMonitor hmm =
                ((InternalResourceManager) getCache().getResourceManager()).getHeapMonitor();
            int newfakeHeapUsage = Math.round(fakeHeapMaxSize * (criticalHeapThresh - 0.3f)); // below
                                                                                              // critical
                                                                                              // by
                                                                                              // 30%
            assertTrue(fakeHeapMaxSize > 0);
            getCache().getLogger().fine(addExpectedBelow);
            hmm.updateStateAndSendEvent(newfakeHeapUsage, "test");
            getCache().getLogger().fine(removeExpectedBelow);
            assertFalse(hmm.getState().isCritical());
            Integer k = 3; // same key as previously used, this time is should stick
            Integer expectedInvocations8 = expectedInvocations.incrementAndGet();
            Region<Integer, String> r = getCache().getRegion(rName);
            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 {
        HeapMemoryMonitor hmm =
            ((InternalResourceManager) getCache().getResourceManager()).getHeapMonitor();
        assertFalse(hmm.getState().isCritical());
        Integer k = 4;
        Integer expectedInvocations9 = expectedInvocations.incrementAndGet();
        Region<Integer, String> r = 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
        getCache().getLogger().fine(addExpectedExString);
        long newfakeHeapUsage = Math.round(fakeHeapMaxSize * (criticalHeapThresh + 0.1f)); // usage
                                                                                           // above
                                                                                           // critical
                                                                                           // by 10%
        assertTrue(newfakeHeapUsage > 0);
        assertTrue(newfakeHeapUsage <= fakeHeapMaxSize);
        hmm.updateStateAndSendEvent(newfakeHeapUsage, "test");
        getCache().getLogger().fine(removeExpectedExString);
        assertTrue(hmm.getState().isCritical());
        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
        newfakeHeapUsage = Math.round(fakeHeapMaxSize * (criticalHeapThresh - 0.3f)); // below
                                                                                      // critical by
                                                                                      // 30%
        assertTrue(fakeHeapMaxSize > 0);
        getCache().getLogger().fine(addExpectedBelow);
        hmm.updateStateAndSendEvent(newfakeHeapUsage, "test");
        getCache().getLogger().fine(removeExpectedBelow);
        assertFalse(hmm.getState().isCritical());
        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/MemoryThresholdsDUnitTest.java