/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.cache.management;

import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.OFF_HEAP_MEMORY_SIZE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.CacheLoaderException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.LoaderHelper;
import org.apache.geode.cache.LowMemoryException;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.ServerOperationException;
import org.apache.geode.cache.control.ResourceManager;
import org.apache.geode.cache.management.MemoryThresholdsDUnitTest.Range;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.cache30.ClientServerTestCase;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.PartitionedRegionHelper;
import org.apache.geode.internal.cache.ProxyBucketRegion;
import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType;
import org.apache.geode.internal.cache.control.MemoryEvent;
import org.apache.geode.internal.cache.control.MemoryThresholds.MemoryState;
import org.apache.geode.internal.cache.control.OffHeapMemoryMonitor;
import org.apache.geode.internal.cache.control.OffHeapMemoryMonitor.OffHeapMemoryMonitorObserver;
import org.apache.geode.internal.cache.control.ResourceAdvisor;
import org.apache.geode.internal.cache.control.ResourceListener;
import org.apache.geode.internal.cache.control.TestMemoryThresholdListener;
import org.apache.geode.internal.cache.partitioned.RegionAdvisor;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.DistributedTestUtils;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.SerializableCallable;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.dunit.WaitCriterion;
import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
import org.apache.geode.test.junit.categories.OffHeapTest;

/**
 * Tests the Off-Heap Memory thresholds of {@link ResourceManager}
 *
 * @since Geode 1.0
 */
@Category({OffHeapTest.class})
public class MemoryThresholdsOffHeapDUnitTest extends ClientServerTestCase {

  final String expectedEx = "Member: .*? above .*? critical threshold";
  final String addExpectedExString =
      "<ExpectedException action=add>" + expectedEx + "</ExpectedException>";
  final String removeExpectedExString =
      "<ExpectedException action=remove>" + expectedEx + "</ExpectedException>";
  final String expectedBelow = "Member: .*? below .*? critical threshold";
  final String addExpectedBelow =
      "<ExpectedException action=add>" + expectedBelow + "</ExpectedException>";
  final String removeExpectedBelow =
      "<ExpectedException action=remove>" + expectedBelow + "</ExpectedException>";

  @Override
  public final void postSetUpClientServerTestCase() throws Exception {
    IgnoredException.addIgnoredException(expectedEx);
    IgnoredException.addIgnoredException(expectedBelow);
  }

  @Override
  protected void preTearDownClientServerTestCase() throws Exception {
    Invoke.invokeInEveryVM(resetResourceManager);
  }

  private final SerializableCallable resetResourceManager = new SerializableCallable() {
    @Override
    public Object call() throws Exception {
      InternalResourceManager irm = getCache().getInternalResourceManager();
      Set<ResourceListener<?>> listeners = irm.getResourceListeners(ResourceType.OFFHEAP_MEMORY);
      for (final ResourceListener<?> l : listeners) {
        if (l instanceof TestMemoryThresholdListener) {
          ((TestMemoryThresholdListener) l).resetThresholdCalls();
        }
      }
      return null;
    }
  };

  /**
   * Make sure appropriate events are delivered when moving between states.
   */
  @Test
  public void testEventDelivery() throws Exception {
    final Host host = Host.getHost(0);
    final VM server1 = host.getVM(0);
    final VM server2 = host.getVM(1);

    final String regionName = "offHeapEventDelivery";

    startCacheServer(server1, 0f, 0f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);
    startCacheServer(server2, 70f, 90f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);

    registerTestMemoryThresholdListener(server1);
    registerTestMemoryThresholdListener(server2);

    // NORMAL -> EVICTION
    setUsageAboveEvictionThreshold(server2, regionName);
    verifyListenerValue(server1, MemoryState.EVICTION, 1, true);
    verifyListenerValue(server2, MemoryState.EVICTION, 1, true);

    // EVICTION -> CRITICAL
    setUsageAboveCriticalThreshold(server2, regionName);
    verifyListenerValue(server1, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server2, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server1, MemoryState.EVICTION, 2, true);
    verifyListenerValue(server2, MemoryState.EVICTION, 2, true);

    // CRITICAL -> CRITICAL
    server2.invoke(new SerializableCallable() {
      private static final long serialVersionUID = 1L;

      @Override
      public Object call() throws Exception {
        getCache().getLogger().fine(addExpectedExString);
        getRootRegion().getSubregion(regionName).destroy("oh3");
        getCache().getLogger().fine(removeExpectedExString);
        return null;
      }
    });
    verifyListenerValue(server1, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server2, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server1, MemoryState.EVICTION, 2, true);
    verifyListenerValue(server2, MemoryState.EVICTION, 2, true);

    // CRITICAL -> EVICTION
    server2.invoke(new SerializableCallable() {
      private static final long serialVersionUID = 1L;

      @Override
      public Object call() throws Exception {
        getCache().getLogger().fine(addExpectedBelow);
        getRootRegion().getSubregion(regionName).destroy("oh2");
        getCache().getLogger().fine(removeExpectedBelow);
        return null;
      }
    });
    verifyListenerValue(server1, MemoryState.EVICTION, 3, true);
    verifyListenerValue(server2, MemoryState.EVICTION, 3, true);

    // EVICTION -> EVICTION
    server2.invoke(new SerializableCallable() {
      private static final long serialVersionUID = 1L;

      @Override
      public Object call() throws Exception {
        getRootRegion().getSubregion(regionName).put("oh6", new byte[20480]);
        return null;
      }
    });
    verifyListenerValue(server1, MemoryState.EVICTION, 3, true);
    verifyListenerValue(server2, MemoryState.EVICTION, 3, true);

    // EVICTION -> NORMAL
    server2.invoke(new SerializableCallable() {
      private static final long serialVersionUID = 1L;

      @Override
      public Object call() throws Exception {
        getRootRegion().getSubregion(regionName).destroy("oh4");
        return null;
      }
    });

    verifyListenerValue(server1, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server1, MemoryState.EVICTION, 3, true);
    verifyListenerValue(server1, MemoryState.NORMAL, 1, true);

    verifyListenerValue(server2, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server2, MemoryState.EVICTION, 3, true);
    verifyListenerValue(server2, MemoryState.NORMAL, 1, true);
  }

  /**
   * test that disabling threshold does not cause remote event and remote DISABLED events are
   * delivered
   */
  @Test
  public void testDisabledThresholds() throws Exception {
    final Host host = Host.getHost(0);
    final VM server1 = host.getVM(0);
    final VM server2 = host.getVM(1);

    final String regionName = "offHeapDisabledThresholds";

    startCacheServer(server1, 0f, 0f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);
    startCacheServer(server2, 0f, 0f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);

    registerTestMemoryThresholdListener(server1);
    registerTestMemoryThresholdListener(server2);

    setUsageAboveEvictionThreshold(server1, regionName);
    verifyListenerValue(server1, MemoryState.EVICTION, 0, true);
    verifyListenerValue(server2, MemoryState.EVICTION, 0, true);

    setThresholds(server1, 70f, 0f);
    verifyListenerValue(server1, MemoryState.EVICTION, 1, true);
    verifyListenerValue(server2, MemoryState.EVICTION, 1, true);

    setUsageAboveCriticalThreshold(server1, regionName);
    verifyListenerValue(server1, MemoryState.CRITICAL, 0, true);
    verifyListenerValue(server2, MemoryState.CRITICAL, 0, true);

    setThresholds(server1, 0f, 0f);
    verifyListenerValue(server1, MemoryState.EVICTION_DISABLED, 1, true);
    verifyListenerValue(server2, MemoryState.EVICTION_DISABLED, 1, true);

    setThresholds(server1, 0f, 90f);
    verifyListenerValue(server1, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server2, MemoryState.CRITICAL, 1, true);

    // verify that stats on server2 are not changed by events on server1
    server2.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        InternalResourceManager irm = getCache().getInternalResourceManager();
        assertEquals(0, irm.getStats().getOffHeapEvictionStartEvents());
        assertEquals(0, irm.getStats().getOffHeapCriticalEvents());
        assertEquals(0, irm.getStats().getOffHeapCriticalThreshold());
        assertEquals(0, irm.getStats().getOffHeapEvictionThreshold());
        return null;
      }
    });
  }

  private void setUsageAboveCriticalThreshold(final VM vm, final String regionName) {
    vm.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        getCache().getLogger().fine(addExpectedExString);
        Region region = getRootRegion().getSubregion(regionName);
        if (!region.containsKey("oh1")) {
          region.put("oh5", new byte[954204]);
        } else {
          region.put("oh5", new byte[122880]);
        }
        getCache().getLogger().fine(removeExpectedExString);
        return null;
      }
    });
  }

  private void setUsageAboveEvictionThreshold(final VM vm, final String regionName) {
    vm.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        getCache().getLogger().fine(addExpectedBelow);
        Region region = getRootRegion().getSubregion(regionName);
        region.put("oh1", new byte[245760]);
        region.put("oh2", new byte[184320]);
        region.put("oh3", new byte[33488]);
        region.put("oh4", new byte[378160]);
        getCache().getLogger().fine(removeExpectedBelow);
        return null;
      }
    });
  }

  private void setUsageBelowEviction(final VM vm, final String regionName) {
    vm.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        getCache().getLogger().fine(addExpectedBelow);
        Region region = getRootRegion().getSubregion(regionName);
        region.remove("oh1");
        region.remove("oh2");
        region.remove("oh3");
        region.remove("oh4");
        region.remove("oh5");
        getCache().getLogger().fine(removeExpectedBelow);
        return null;
      }
    });
  }

  private void setThresholds(VM server, final float evictionThreshold,
      final float criticalThreshold) {

    server.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        ResourceManager irm = getCache().getResourceManager();
        irm.setCriticalOffHeapPercentage(criticalThreshold);
        irm.setEvictionOffHeapPercentage(evictionThreshold);
        return null;
      }
    });
  }

  /**
   * test that puts in a client are rejected when a remote VM crosses critical threshold
   */
  @Test
  public void testDistributedRegionRemoteClientPutRejection() throws Exception {
    final Host host = Host.getHost(0);
    final VM server1 = host.getVM(0);
    final VM server2 = host.getVM(1);
    final VM client = host.getVM(2);

    final String regionName = "offHeapDRRemoteClientPutReject";

    final int port1 = startCacheServer(server1, 0f, 0f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);
    startCacheServer(server2, 0f, 90f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);

    startClient(client, server1, port1, regionName);

    registerTestMemoryThresholdListener(server1);
    registerTestMemoryThresholdListener(server2);

    doPuts(client, regionName, false/* catchRejectedException */,
        false/* catchLowMemoryException */);
    doPutAlls(client, regionName, false/* catchRejectedException */,
        false/* catchLowMemoryException */, Range.DEFAULT);

    // make server2 critical
    setUsageAboveCriticalThreshold(server2, regionName);

    verifyListenerValue(server1, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server2, MemoryState.CRITICAL, 1, true);

    // make sure that client puts are rejected
    doPuts(client, regionName, true/* catchRejectedException */,
        false/* catchLowMemoryException */);
    doPutAlls(client, regionName, true/* catchRejectedException */,
        false/* catchLowMemoryException */, new Range(Range.DEFAULT, Range.DEFAULT.width() + 1));

    setUsageBelowEviction(server2, regionName);
  }

  @Test
  public void testDistributedRegionRemotePutRejectionLocalDestroy() throws Exception {
    doDistributedRegionRemotePutRejection(true, false);
  }

  @Test
  public void testDistributedRegionRemotePutRejectionCacheClose() throws Exception {
    doDistributedRegionRemotePutRejection(false, true);
  }

  @Test
  public void testDistributedRegionRemotePutRejectionBelowThreshold() throws Exception {
    doDistributedRegionRemotePutRejection(false, false);
  }

  @Test
  public void testGettersAndSetters() {
    getSystem(getOffHeapProperties());
    ResourceManager rm = getCache().getResourceManager();
    assertEquals(0.0f, rm.getCriticalOffHeapPercentage(), 0);
    assertEquals(0.0f, rm.getEvictionOffHeapPercentage(), 0);

    rm.setEvictionOffHeapPercentage(50);
    rm.setCriticalOffHeapPercentage(90);

    // verify
    assertEquals(50.0f, rm.getEvictionOffHeapPercentage(), 0);
    assertEquals(90.0f, rm.getCriticalOffHeapPercentage(), 0);

    getCache().createRegionFactory(RegionShortcut.REPLICATE_HEAP_LRU).create(getName());

    assertEquals(50.0f, rm.getEvictionOffHeapPercentage(), 0);
    assertEquals(90.0f, rm.getCriticalOffHeapPercentage(), 0);
  }

  /**
   * test that puts in a server are rejected when a remote VM crosses critical threshold
   *
   */
  private void doDistributedRegionRemotePutRejection(boolean localDestroy, boolean cacheClose)
      throws Exception {
    final Host host = Host.getHost(0);
    final VM server1 = host.getVM(0);
    final VM server2 = host.getVM(1);

    final String regionName = "offHeapDRRemotePutRejection";

    // set port to 0 in-order for system to pickup a random port.
    startCacheServer(server1, 0f, 0f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);
    startCacheServer(server2, 0f, 90f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);

    registerTestMemoryThresholdListener(server1);
    registerTestMemoryThresholdListener(server2);

    doPuts(server1, regionName, false/* catchRejectedException */,
        false/* catchLowMemoryException */);
    doPutAlls(server1, regionName, false/* catchRejectedException */,
        false/* catchLowMemoryException */, Range.DEFAULT);

    // make server2 critical
    setUsageAboveCriticalThreshold(server2, regionName);

    verifyListenerValue(server1, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server2, MemoryState.CRITICAL, 1, true);

    // make sure that local server1 puts are rejected
    doPuts(server1, regionName, false/* catchRejectedException */,
        true/* catchLowMemoryException */);
    Range r1 = new Range(Range.DEFAULT, Range.DEFAULT.width() + 1);
    doPutAlls(server1, regionName, false/* catchRejectedException */,
        true/* catchLowMemoryException */, r1);

    if (localDestroy) {
      // local destroy the region on sick member
      server2.invoke(new SerializableCallable("local destroy") {
        @Override
        public Object call() throws Exception {
          Region r = getRootRegion().getSubregion(regionName);
          r.localDestroyRegion();
          return null;
        }
      });
    } else if (cacheClose) {
      server2.invoke(new SerializableCallable() {
        @Override
        public Object call() throws Exception {
          getCache().close();
          return null;
        }
      });
    } else {
      setUsageBelowEviction(server2, regionName);
    }

    // wait for remote region destroyed message to be processed
    server1.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        WaitCriterion wc = new WaitCriterion() {
          @Override
          public String description() {
            return "remote localRegionDestroyed message not received";
          }

          @Override
          public boolean done() {
            DistributedRegion dr = (DistributedRegion) getRootRegion().getSubregion(regionName);
            return dr.getAtomicThresholdInfo().getMembersThatReachedThreshold().size() == 0;
          }
        };
        Wait.waitForCriterion(wc, 10000, 10, true);
        return null;
      }
    });

    // make sure puts succeed
    doPuts(server1, regionName, false/* catchRejectedException */,
        false/* catchLowMemoryException */);
    Range r2 = new Range(r1, r1.width() + 1);
    doPutAlls(server1, regionName, false/* catchRejectedException */,
        false/* catchLowMemoryException */, r2);
  }

  /**
   * 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.
   */
  @Test
  public void testDRLoadRejection() throws Exception {
    final Host host = Host.getHost(0);
    final VM replicate1 = host.getVM(1);
    final VM replicate2 = host.getVM(2);
    final String rName = getUniqueName();

    // 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") {
          @SuppressWarnings("synthetic-access")
          @Override
          public void run2() throws CacheException {
            getSystem(getOffHeapProperties());
          }
        };
    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(getOffHeapProperties());
            assertTrue(ds.getDistributionManager().getNormalDistributionManagerIds().size() >= 1);

            InternalResourceManager irm = (InternalResourceManager) getCache().getResourceManager();
            irm.setCriticalOffHeapPercentage(90f);
            AttributesFactory af = new AttributesFactory();
            af.setScope(Scope.DISTRIBUTED_ACK);
            af.setDataPolicy(DataPolicy.REPLICATE);
            af.setOffHeap(true);
            Region region = 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 {
            final DistributedRegion r = (DistributedRegion) getCache().getRegion(rName);
            AttributesMutator<Integer, String> am = r.getAttributesMutator();
            am.setCacheLoader(new CacheLoader<Integer, String>() {
              final AtomicInteger numLoaderInvocations = new AtomicInteger(0);

              @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;
            final OffHeapMemoryMonitor ohmm =
                ((InternalResourceManager) getCache().getResourceManager()).getOffHeapMonitor();
            assertFalse(ohmm.getState().isCritical());
            {
              Integer k = 1;
              assertEquals(k.toString(), r.get(k, expectedInvocations++));
            }

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

            WaitCriterion wc = new WaitCriterion() {
              @Override
              public String description() {
                return "expected region " + r + " to set memoryThreshold";
              }

              @Override
              public boolean done() {
                return r.isMemoryThresholdReached();
              }
            };
            Wait.waitForCriterion(wc, 30 * 1000, 10, true);
            {
              Integer k = 2;
              assertEquals(k.toString(), r.get(k, expectedInvocations++));
            }

            r.destroy("oh3");
            wc = new WaitCriterion() {
              @Override
              public String description() {
                return "expected region " + r + " to unset memoryThreshold";
              }

              @Override
              public boolean done() {
                return !r.isMemoryThresholdReached();
              }
            };
            Wait.waitForCriterion(wc, 30 * 1000, 10, true);
            {
              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 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 {
        final DistributedRegion r = (DistributedRegion) getCache().getRegion(rName);
        final OffHeapMemoryMonitor ohmm =
            ((InternalResourceManager) getCache().getResourceManager()).getOffHeapMonitor();
        assertFalse(ohmm.getState().isCritical());

        int expectedInvocations = expected;
        {
          Integer k = 4;
          assertEquals(k.toString(), r.get(k, expectedInvocations++));
        }

        // Place in a critical state for the next test
        r.put("oh3", new byte[157287]);
        WaitCriterion wc = new WaitCriterion() {
          @Override
          public String description() {
            return "expected region " + r + " to set memoryThreshold";
          }

          @Override
          public boolean done() {
            return r.isMemoryThresholdReached();
          }
        };
        Wait.waitForCriterion(wc, 30 * 1000, 10, true);
        {
          Integer k = 5;
          assertEquals(k.toString(), r.get(k, expectedInvocations++));
        }

        r.destroy("oh3");
        wc = new WaitCriterion() {
          @Override
          public String description() {
            return "expected region " + r + " to unset memoryThreshold";
          }

          @Override
          public boolean done() {
            return !r.isMemoryThresholdReached();
          }
        };
        Wait.waitForCriterion(wc, 30 * 1000, 10, true);
        {
          Integer k = 6;
          assertEquals(k.toString(), r.get(k, expectedInvocations++));
        }
        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);
  }


  private final SerializableRunnable addExpectedException =
      new SerializableRunnable("addExpectedEx") {
        @Override
        public void run() {
          getCache().getLogger().fine(addExpectedExString);
          getCache().getLogger().fine(addExpectedBelow);
        }
      };

  private final SerializableRunnable removeExpectedException =
      new SerializableRunnable("removeExpectedException") {
        @Override
        public void run() {
          getCache().getLogger().fine(removeExpectedExString);
          getCache().getLogger().fine(removeExpectedBelow);
        }
      };

  @Test
  public void testPR_RemotePutRejectionLocalDestroy() throws Exception {
    prRemotePutRejection(false, true, false);
  }

  @Test
  public void testPR_RemotePutRejectionCacheClose() throws Exception {
    prRemotePutRejection(true, false, false);
  }

  @Test
  public void testPR_RemotePutRejection() throws Exception {
    prRemotePutRejection(false, false, false);
  }

  @Test
  public void testPR_RemotePutRejectionLocalDestroyWithTx() throws Exception {
    prRemotePutRejection(false, true, true);
  }

  @Test
  public void testPR_RemotePutRejectionCacheCloseWithTx() throws Exception {
    prRemotePutRejection(true, false, true);
  }

  @Test
  public void testPR_RemotePutRejectionWithTx() throws Exception {
    prRemotePutRejection(false, false, true);
  }

  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
