@Test
  public void testDRLoadRejection() throws Exception {
    final Host host = Host.getHost(0);
    final VM replicate1 = host.getVM(2);
    final VM replicate2 = host.getVM(3);
    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> d1 = replicate1.invokeAsync(JUnit4DistributedTestCase::disconnectFromDS);
    AsyncInvocation<Void> d2 = replicate2.invokeAsync(JUnit4DistributedTestCase::disconnectFromDS);
    d1.join();
    assertFalse(d1.exceptionOccurred());
    d2.join();
    assertFalse(d2.exceptionOccurred());
    CacheSerializableRunnable establishConnectivity =
        new CacheSerializableRunnable("establishcConnectivity") {
          @Override
          public void run2() throws CacheException {
            getSystem();
          }
        };
    replicate1.invoke(establishConnectivity);
    replicate2.invoke(establishConnectivity);

    CacheSerializableRunnable createRegion =
        new CacheSerializableRunnable("create DistributedRegion") {
          @Override
          public void run2() throws CacheException {
            // Assert some level of connectivity
            InternalDistributedSystem ds = getSystem();
            assertTrue(ds.getDistributionManager().getNormalDistributionManagerIds().size() >= 1);

            final long fakeHeapUsage = Math.round(fakeHeapMaxSize * (criticalHeapThresh - 0.5f)); // below
                                                                                                  // critical
                                                                                                  // by
                                                                                                  // 50%
            InternalResourceManager irm = (InternalResourceManager) getCache().getResourceManager();
            HeapMemoryMonitor hmm = irm.getHeapMonitor();
            assertTrue(fakeHeapMaxSize > 0);
            hmm.setTestMaxMemoryBytes(fakeHeapMaxSize);
            HeapMemoryMonitor.setTestBytesUsedForThresholdSet(fakeHeapUsage);
            irm.setCriticalHeapPercentage((criticalHeapThresh * 100.0f));
            AttributesFactory<Integer, String> af = new AttributesFactory<>();
            af.setScope(Scope.DISTRIBUTED_ACK);
            af.setDataPolicy(DataPolicy.REPLICATE);
            getCache().createRegion(rName, af.create());
          }
        };
    replicate1.invoke(createRegion);
    replicate2.invoke(createRegion);

    replicate1.invoke(addExpectedException);
    replicate2.invoke(addExpectedException);

    final Integer expected =
        (Integer) replicate1.invoke(new SerializableCallable("test Local DistributedRegion Load") {
          @Override
          public Object call() throws Exception {
            Region<Integer, String> r = getCache().getRegion(rName);
            AttributesMutator<Integer, String> am = r.getAttributesMutator();
            am.setCacheLoader(new CacheLoader<Integer, String>() {
              final AtomicInteger numLoaderInvocations = new AtomicInteger();

              @Override
              public String load(LoaderHelper<Integer, String> helper) throws CacheLoaderException {
                Integer expectedInvocations = (Integer) helper.getArgument();
                final int actualInvocations = numLoaderInvocations.getAndIncrement();
                if (expectedInvocations != actualInvocations) {
                  throw new CacheLoaderException("Expected " + expectedInvocations
                      + " invocations, actual is " + actualInvocations);
                }
                return helper.getKey().toString();
              }

              @Override
              public void close() {}
            });

            int expectedInvocations = 0;
            HeapMemoryMonitor hmm =
                ((InternalResourceManager) getCache().getResourceManager()).getHeapMonitor();
            assertFalse(hmm.getState().isCritical());
            {
              Integer k = 1;
              assertEquals(k.toString(), r.get(k, expectedInvocations++));
            }

            long newfakeHeapUsage = Math.round(fakeHeapMaxSize * (criticalHeapThresh + 0.1f)); // usage
                                                                                               // above
                                                                                               // critical
                                                                                               // by
                                                                                               // 10%
            assertTrue(newfakeHeapUsage > 0);
            assertTrue(newfakeHeapUsage <= fakeHeapMaxSize);
            hmm.updateStateAndSendEvent(newfakeHeapUsage, "test");
            assertTrue(hmm.getState().isCritical());
            {
              Integer k = 2;
              assertEquals(k.toString(), r.get(k, expectedInvocations++));
            }

            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;
              assertEquals(k.toString(), r.get(k, expectedInvocations++));
            }
            return expectedInvocations;
          }
        });

    final CacheSerializableRunnable validateData1 =
        new CacheSerializableRunnable("Validate data 1") {
          @Override
          public void run2() throws CacheException {
            Region<Integer, String> r = getCache().getRegion(rName);
            Integer i1 = 1;
            assertTrue(r.containsKey(i1));
            assertNotNull(r.getEntry(i1));
            Integer i2 = 2;
            assertFalse(r.containsKey(i2));
            assertNull(r.getEntry(i2));
            Integer i3 = 3;
            assertTrue(r.containsKey(i3));
            assertNotNull(r.getEntry(i3));
          }
        };
    replicate1.invoke(validateData1);
    replicate2.invoke(validateData1);

    replicate2.invoke(new SerializableCallable("test DistributedRegion netLoad") {
      @Override
      public Object call() throws Exception {
        Region<Integer, String> r = getCache().getRegion(rName);
        HeapMemoryMonitor hmm =
            ((InternalResourceManager) getCache().getResourceManager()).getHeapMonitor();
        assertFalse(hmm.getState().isCritical());

        int expectedInvocations = expected;
        {
          Integer k = 4;
          assertEquals(k.toString(), r.get(k, expectedInvocations++));
          assertFalse(hmm.getState().isCritical());
          assertTrue(r.containsKey(k));
        }

        // Place in a critical state for the next test
        long newfakeHeapUsage = Math.round(fakeHeapMaxSize * (criticalHeapThresh + 0.1f)); // usage
                                                                                           // above
                                                                                           // critical
                                                                                           // by 10%
        assertTrue(newfakeHeapUsage > 0);
        assertTrue(newfakeHeapUsage <= fakeHeapMaxSize);
        hmm.updateStateAndSendEvent(newfakeHeapUsage, "test");
        assertTrue(hmm.getState().isCritical());
        {
          Integer k = 5;
          assertEquals(k.toString(), r.get(k, expectedInvocations++));
          assertTrue(hmm.getState().isCritical());
          assertFalse(r.containsKey(k));
        }

        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 = 6;
          assertEquals(k.toString(), r.get(k, expectedInvocations++));
          assertFalse(hmm.getState().isCritical());
          assertTrue(r.containsKey(k));
        }
        return expectedInvocations;
      }
    });

    replicate1.invoke(removeExpectedException);
    replicate2.invoke(removeExpectedException);

    final CacheSerializableRunnable validateData2 =
        new CacheSerializableRunnable("Validate data 2") {
          @Override
          public void run2() throws CacheException {
            Region<Integer, String> r = getCache().getRegion(rName);
            Integer i4 = 4;
            assertTrue(r.containsKey(i4));
            assertNotNull(r.getEntry(i4));
            Integer i5 = 5;
            assertFalse(r.containsKey(i5));
            assertNull(r.getEntry(i5));
            Integer i6 = 6;
            assertTrue(r.containsKey(i6));
            assertNotNull(r.getEntry(i6));
          }
        };
    replicate1.invoke(validateData2);
    replicate2.invoke(validateData2);
  }
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/MemoryThresholdsDUnitTest.java