/*
 * 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 java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.geode.cache.Scope.LOCAL;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.internal.cache.PartitionedRegionHelper.getHashKey;
import static org.apache.geode.internal.cache.control.HeapMemoryMonitor.getTenuredMemoryPoolMXBean;
import static org.apache.geode.internal.cache.control.HeapMemoryMonitor.getTenuredPoolMaxMemory;
import static org.apache.geode.internal.cache.control.HeapMemoryMonitor.getTenuredPoolStatistics;
import static org.apache.geode.test.dunit.Assert.assertEquals;
import static org.apache.geode.test.dunit.Assert.assertFalse;
import static org.apache.geode.test.dunit.Assert.assertNotNull;
import static org.apache.geode.test.dunit.Assert.assertNull;
import static org.apache.geode.test.dunit.Assert.assertTrue;
import static org.apache.geode.test.dunit.Assert.fail;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.io.Serializable;
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.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Ignore;
import org.junit.Test;

import org.apache.geode.DataSerializable;
import org.apache.geode.Statistics;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.Cache;
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.Pool;
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.execute.Execution;
import org.apache.geode.cache.execute.FunctionAdapter;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.FunctionException;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.execute.RegionFunctionContext;
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.AvailablePortHelper;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.PartitionedRegionHelper;
import org.apache.geode.internal.cache.control.HeapMemoryMonitor;
import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType;
import org.apache.geode.internal.cache.control.MemoryThresholds.MemoryState;
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.statistics.GemFireStatSampler;
import org.apache.geode.internal.statistics.LocalStatListener;
import org.apache.geode.test.awaitility.GeodeAwaitility;
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.WaitCriterion;
import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;

/**
 * Tests the Heap Memory thresholds of {@link ResourceManager}
 *
 * @since GemFire 6.0
 */

public class MemoryThresholdsDUnitTest extends ClientServerTestCase {

  public static class Range implements DataSerializable {
    public static final Range DEFAULT = new Range(0, 20);
    public int start;
    public int end;

    public Range() {}

    public Range(int s, int e) {
      start = s;
      end = e;
    }

    public Range(Range r, int shift) {
      start = r.start + shift;
      end = r.end + shift;
    }

    public int width() {
      return end - start;
    }

    @Override
    public void toData(DataOutput out) throws IOException {
      out.writeInt(start);
      out.writeInt(end);
    }

    @Override
    public void fromData(DataInput in) throws IOException, ClassNotFoundException {
      start = in.readInt();
      end = in.readInt();
    }
  }

  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>";

  final String expectedFunctionEx = " cannot be executed";
  final String addExpectedFunctionExString =
      "<ExpectedException action=add>" + expectedFunctionEx + "</ExpectedException>";
  final String removeExpectedFunctionExString =
      "<ExpectedException action=remove>" + expectedFunctionEx + "</ExpectedException>";

  @Override
  protected final void postSetUpClientServerTestCase() throws Exception {
    Invoke.invokeInEveryVM(setHeapMemoryMonitorTestMode);
    IgnoredException.addIgnoredException(expectedEx);
    IgnoredException.addIgnoredException(expectedBelow);
  }

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

  @Test
  public void testPRClientPutRejection() throws Exception {
    doClientServerTest("parRegReject", true/* createPR */);
  }

  @Test
  public void testDistributedRegionClientPutRejection() throws Exception {
    doClientServerTest("distrReject", false/* createPR */);
  }

  private void doClientServerTest(final String regionName, boolean createPR) throws Exception {
    // create region on the server
    final Host host = Host.getHost(0);
    final VM server = host.getVM(0);
    final VM client = host.getVM(1);

    ServerPorts ports = startCacheServer(server, 0f, 90f, regionName, createPR, false, 0);
    startClient(client, server, ports.getPort(), regionName);
    doPuts(client, regionName, false/* catchServerException */, false/* catchLowMemoryException */);
    doPutAlls(client, regionName, false/* catchServerException */,
        false/* catchLowMemoryException */, Range.DEFAULT);

    // make the region sick in the server
    server.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        InternalResourceManager irm = (InternalResourceManager) getCache().getResourceManager();
        irm.setCriticalHeapPercentage(90f);

        getCache().getLogger().fine(addExpectedExString);
        HeapMemoryMonitor.setTestBytesUsedForThresholdSet(950);
        irm.getHeapMonitor().updateStateAndSendEvent();
        getCache().getLogger().fine(removeExpectedExString);
        return null;
      }
    });

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

  @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 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 = "rejectRemoteOp";

    ServerPorts ports1 = startCacheServer(server1, 0f, 0f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);
    ServerPorts ports2 = 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);

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

    // 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);
    }

    // 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;
          }
        };
        GeodeAwaitility.await().untilAsserted(wc);
        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
  public void testBug45513() {
    ResourceManager rm = getCache().getResourceManager();
    assertEquals(0.0f, rm.getCriticalHeapPercentage(), 0);
    assertEquals(0.0f, rm.getEvictionHeapPercentage(), 0);

    rm.setEvictionHeapPercentage(50);
    rm.setCriticalHeapPercentage(90);

    // verify
    assertEquals(50.0f, rm.getEvictionHeapPercentage(), 0);
    assertEquals(90.0f, rm.getCriticalHeapPercentage(), 0);

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

    assertEquals(50.0f, rm.getEvictionHeapPercentage(), 0);
    assertEquals(90.0f, rm.getCriticalHeapPercentage(), 0);
  }

  /**
   * 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 = "rejectRemoteClientOp";

    ServerPorts ports1 = startCacheServer(server1, 0f, 0f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);
    ServerPorts ports2 = startCacheServer(server2, 0f, 90f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);

    startClient(client, server1, ports1.getPort(), 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);

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

    // 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));
  }


  /**
   * 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 = "disableThresholdPr";

    ServerPorts ports1 = startCacheServer(server1, 0f, 0f, regionName, true/* createPR */,
        false/* notifyBySubscription */, 0);
    ServerPorts ports2 = startCacheServer(server2, 0f, 0f, regionName, true/* createPR */,
        false/* notifyBySubscription */, 0);

    registerTestMemoryThresholdListener(server1);
    registerTestMemoryThresholdListener(server2);

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

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

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

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

    setThresholds(server1, 0f, 90f);
    verifyListenerValue(server1, MemoryState.CRITICAL, 1, false);
    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().getEvictionStartEvents());
        assertEquals(0, irm.getStats().getHeapCriticalEvents());
        assertEquals(0, irm.getStats().getCriticalThreshold());
        assertEquals(0, irm.getStats().getEvictionThreshold());
        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 = "testEventDelivery";

    ServerPorts ports1 = startCacheServer(server1, 0f, 0f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);
    ServerPorts ports2 = startCacheServer(server2, 80f, 90f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);

    registerLoggingTestMemoryThresholdListener(server1);
    registerTestMemoryThresholdListener(server2);

    // NORMAL -> CRITICAL
    server2.invoke(new SerializableCallable("NORMAL->CRITICAL") {
      @Override
      public Object call() throws Exception {
        InternalCache gfCache = getCache();
        getCache().getLogger().fine(addExpectedExString);
        gfCache.getInternalResourceManager().getHeapMonitor().updateStateAndSendEvent(950, "test");
        getCache().getLogger().fine(removeExpectedExString);
        return null;
      }
    });
    verifyListenerValue(server2, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server2, MemoryState.EVICTION, 1, true);
    verifyListenerValue(server2, MemoryState.NORMAL, 0, true);

    // make sure we get two events on remote server
    verifyListenerValue(server1, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server1, MemoryState.EVICTION, 1, true);
    verifyListenerValue(server1, MemoryState.NORMAL, 0, true);

    // CRITICAL -> EVICTION
    server2.invoke(new SerializableCallable("CRITICAL->EVICTION") {
      @Override
      public Object call() throws Exception {
        InternalCache gfCache = getCache();
        getCache().getLogger().fine(addExpectedBelow);
        gfCache.getInternalResourceManager().getHeapMonitor().updateStateAndSendEvent(850, "test");
        getCache().getLogger().fine(removeExpectedBelow);
        return null;
      }
    });
    verifyListenerValue(server2, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server2, MemoryState.EVICTION, 2, true);
    verifyListenerValue(server2, MemoryState.NORMAL, 0, true);
    verifyListenerValue(server1, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server1, MemoryState.EVICTION, 2, true);
    verifyListenerValue(server1, MemoryState.NORMAL, 0, true);

    // EVICTION -> EVICTION
    server2.invoke(new SerializableCallable("EVICTION->EVICTION") {
      @Override
      public Object call() throws Exception {
        InternalCache gfCache = getCache();
        gfCache.getInternalResourceManager().getHeapMonitor().updateStateAndSendEvent(840, "test");
        return null;
      }
    });
    verifyListenerValue(server2, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server2, MemoryState.EVICTION, 2, true);
    verifyListenerValue(server2, MemoryState.NORMAL, 0, true);
    verifyListenerValue(server1, MemoryState.CRITICAL, 1, true);
    verifyListenerValue(server1, MemoryState.EVICTION, 2, true);
    verifyListenerValue(server1, MemoryState.NORMAL, 0, true);

    // EVICTION -> NORMAL
    server2.invoke(new SerializableCallable("EVICTION->NORMAL") {
      @Override
      public Object call() throws Exception {
        InternalCache gfCache = getCache();
        gfCache.getInternalResourceManager().getHeapMonitor().updateStateAndSendEvent(750, "test");
        return null;
      }
    });

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

    // NORMAL -> CRITICAL
    server2.invoke(new SerializableCallable("NORMAL->CRITICAL") {
      @Override
      public Object call() throws Exception {
        InternalCache gfCache = getCache();
        gfCache.getInternalResourceManager().getHeapMonitor().updateStateAndSendEvent(950, "test");
        return null;
      }
    });

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

    server2.invoke(new SerializableCallable("CRITICAL->NORMAL") {
      @Override
      public Object call() throws Exception {
        InternalCache gfCache = getCache();
        gfCache.getInternalResourceManager().getHeapMonitor().updateStateAndSendEvent(750, "test");
        return null;
      }
    });

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

    // NORMAL -> EVICTION
    server2.invoke(new SerializableCallable("NORMAL->EVICTION") {
      @Override
      public Object call() throws Exception {
        InternalCache gfCache = getCache();
        gfCache.getInternalResourceManager().getHeapMonitor().updateStateAndSendEvent(850, "test");
        return null;
      }
    });

    verifyListenerValue(server2, MemoryState.CRITICAL, 2, true);
    verifyListenerValue(server2, MemoryState.EVICTION, 4, true);
    verifyListenerValue(server2, MemoryState.NORMAL, 2, true);
    verifyListenerValue(server1, MemoryState.CRITICAL, 2, true);
    verifyListenerValue(server1, MemoryState.EVICTION, 4, true);
    verifyListenerValue(server1, MemoryState.NORMAL, 2, true);
  }

  @Test
  public void testCleanAdvisorClose() throws Exception {
    final Host host = Host.getHost(0);
    final VM server1 = host.getVM(0);
    final VM server2 = host.getVM(1);
    final VM server3 = host.getVM(2);

    final String regionName = "testEventOrger";

    ServerPorts ports1 = startCacheServer(server1, 0f, 0f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);
    ServerPorts ports2 = startCacheServer(server2, 0f, 0f, regionName, false/* createPR */,
        false/* notifyBySubscription */, 0);

    verifyProfiles(server1, 2);
    verifyProfiles(server2, 2);

    server2.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        closeCache();
        return null;
      }
    });

    verifyProfiles(server1, 1);

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

    verifyProfiles(server1, 2);
    verifyProfiles(server3, 2);
  }

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

  @Test
  public void testPR_RemotePutRejectionCacheClose() throws Exception {
    // Ignore this excetion as this can happen if pool is shutting down
    IgnoredException
        .addIgnoredException(java.util.concurrent.RejectedExecutionException.class.getName());
    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 {
    // Ignore this excetion as this can happen if pool is shutting down
    IgnoredException
        .addIgnoredException(java.util.concurrent.RejectedExecutionException.class.getName());
    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 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);
  }

  @Ignore("this test is DISABLED due to test issues.  It sometimes fails with a TransactionDataNotColocatedException.  See bug #52222")
  @Test
  public void testTxCommitInCritical() 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);

    registerTestMemoryThresholdListener(server1);
    registerTestMemoryThresholdListener(server2);
    registerTestMemoryThresholdListener(server3);

    accessor.invoke(new SerializableCallable() {
