/*
 * 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.internal.cache;

import static java.lang.System.getProperties;
import static java.util.Map.Entry;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.File;
import java.util.Properties;

import org.junit.Test;

import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.InterestPolicy;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.SubscriptionAttributes;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache.util.ObjectSizer;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.internal.cache.control.HeapMemoryMonitor;
import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType;
import org.apache.geode.internal.cache.eviction.HeapEvictor;
import org.apache.geode.internal.cache.eviction.HeapLRUController;
import org.apache.geode.logging.internal.OSProcess;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
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.cache.internal.JUnit4CacheTestCase;


public class PartitionedRegionEvictionDUnitTest extends JUnit4CacheTestCase {

  @Test
  public void testHeapLRUWithOverflowToDisk() {
    // Ignore this excetion as this can happen if pool is shutting down
    IgnoredException
        .addIgnoredException(java.util.concurrent.RejectedExecutionException.class.getName());
    final Host host = Host.getHost(0);
    final VM vm2 = host.getVM(2);
    final VM vm3 = host.getVM(3);
    final String uniqName = getUniqueName();
    final float heapPercentage = 50.9f;
    final int redundantCopies = 1;
    final int evictorInterval = 100;
    final String name = uniqName + "-PR";

    final CacheSerializableRunnable create =
        new CacheSerializableRunnable("Create Heap LRU with Overflow to disk partitioned Region") {
          @Override
          public void run2() {
            System.setProperty(HeapLRUController.TOP_UP_HEAP_EVICTION_PERCENTAGE_PROPERTY,
                Float.toString(0));
            setEvictionPercentage(heapPercentage);
            final Properties sp = System.getProperties();
            final AttributesFactory factory = new AttributesFactory();
            factory.setOffHeap(isOffHeap());
            factory.setPartitionAttributes(
                new PartitionAttributesFactory().setRedundantCopies(redundantCopies).create());
            factory.setEvictionAttributes(
                EvictionAttributes.createLRUHeapAttributes(null, EvictionAction.OVERFLOW_TO_DISK));
            factory.setDiskSynchronous(true);
            DiskStoreFactory dsf = getCache().createDiskStoreFactory();
            final File[] diskDirs = new File[1];
            diskDirs[0] = new File("overflowDir/" + uniqName + "_" + OSProcess.getId());
            diskDirs[0].mkdirs();
            dsf.setDiskDirs(diskDirs);
            DiskStore ds = dsf.create(name);
            factory.setDiskStoreName(ds.getName());
            final PartitionedRegion pr =
                (PartitionedRegion) createRootRegion(name, factory.create());
            assertNotNull(pr);
          }
        };
    vm3.invoke(create);
    vm2.invoke(create);

    final int bucketsToCreate = 3;
    final SerializableRunnable createBuckets = new SerializableRunnable("Create Buckets") {
      @Override
      public void run() {
        setEvictionPercentage(heapPercentage);
        final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
        assertNotNull(pr);
        // Create three buckets, add enough stuff in each to force two
        // overflow
        // ops in each
        for (int i = 0; i < bucketsToCreate; i++) {
          // assume mod-based hashing for bucket creation
          pr.put(i, "value0");
          pr.put(i + pr.getPartitionAttributes().getTotalNumBuckets(), "value1");
          pr.put(i + (pr.getPartitionAttributes().getTotalNumBuckets()) * 2, "value2");
        }
      }
    };
    vm3.invoke(createBuckets);

    final SerializableCallable assertBucketAttributesAndEviction =
        new SerializableCallable("Assert bucket attributes and eviction") {
          @Override
          public Object call() throws Exception {

            try {
              setEvictionPercentage(heapPercentage);
              final Properties sp = getProperties();
              final int expectedInvocations = 10;
              final int maximumWaitSeconds = 60; // seconds
              final int pollWaitMillis = evictorInterval * 2;
              assertTrue(pollWaitMillis < (SECONDS.toMillis(maximumWaitSeconds) * 4));
              final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
              assertNotNull(pr);
              for (final Entry<Integer, BucketRegion> integerBucketRegionEntry : pr.getDataStore()
                  .getAllLocalBuckets()) {
                final Entry entry = (Entry) integerBucketRegionEntry;
                final BucketRegion bucketRegion = (BucketRegion) entry.getValue();
                if (bucketRegion == null) {
                  continue;
                }
                assertTrue(bucketRegion.getAttributes().getEvictionAttributes().getAlgorithm()
                    .isLRUHeap());
                assertTrue(bucketRegion.getAttributes().getEvictionAttributes().getAction()
                    .isOverflowToDisk());
              }
              raiseFakeNotification();
              WaitCriterion wc = new WaitCriterion() {
                String excuse;

                @Override
                public boolean done() {
                  // we have a primary
                  return pr.getDiskRegionStats().getNumOverflowOnDisk() == 9;
                }

                @Override
                public String description() {
                  return excuse;
                }
              };
              GeodeAwaitility.await().untilAsserted(wc);

              int entriesEvicted = 0;

              entriesEvicted += pr.getDiskRegionStats().getNumOverflowOnDisk();
              return entriesEvicted;
            } finally {
              cleanUpAfterFakeNotification();
              // Clean up the observer for future tests
              // final Properties sp = System.getProperties();
              /*
               * assertNotNull(sp .remove(HeapLRUCapacityController.FREE_MEMORY_OBSERVER));
               */
            }
          }
        };
    final Integer v2i = (Integer) vm2.invoke(assertBucketAttributesAndEviction);
    final Integer v3i = (Integer) vm3.invoke(assertBucketAttributesAndEviction);
    final int totalEvicted = v2i + v3i;
    // assume all three entries in each bucket were evicted to disk
    assertEquals((3 * bucketsToCreate * (redundantCopies + 1)), totalEvicted);
  }

  protected void raiseFakeNotification() {
    getCache().getHeapEvictor().setTestAbortAfterLoopCount(1);
    HeapMemoryMonitor.setTestDisableMemoryUpdates(true);

    setEvictionPercentage(85);
    HeapMemoryMonitor hmm =
        getCache().getInternalResourceManager().getHeapMonitor();
    hmm.setTestMaxMemoryBytes(100);

    hmm.updateStateAndSendEvent(90, "test");
  }

  protected void cleanUpAfterFakeNotification() {
    getCache().getHeapEvictor().setTestAbortAfterLoopCount(Integer.MAX_VALUE);
    HeapMemoryMonitor.setTestDisableMemoryUpdates(false);
  }

  @Test
  public void testHeapLRUWithLocalDestroy() {
    // Ignore this excetion as this can happen if pool is shutting down
    IgnoredException
        .addIgnoredException(java.util.concurrent.RejectedExecutionException.class.getName());
    final Host host = Host.getHost(0);
    final VM vm2 = host.getVM(2);
    final VM vm3 = host.getVM(3);
    final String uniqName = getUniqueName();
    final float heapPercentage = 50.9f;
    final int redundantCopies = 1;
    final int evictorInterval = 100;
    final String name = uniqName + "-PR";

    final SerializableRunnable create = new CacheSerializableRunnable(
        "Create Heap LRU with local destroy on a partitioned Region") {
      @Override
      public void run2() {
        System.setProperty(HeapLRUController.TOP_UP_HEAP_EVICTION_PERCENTAGE_PROPERTY,
            Float.toString(0));
        setEvictionPercentage(heapPercentage);
        final Properties sp = System.getProperties();
        final AttributesFactory factory = new AttributesFactory();
        factory.setOffHeap(isOffHeap());
        factory.setPartitionAttributes(
            new PartitionAttributesFactory().setRedundantCopies(redundantCopies).create());
        factory.setEvictionAttributes(EvictionAttributes.createLRUHeapAttributes());
        final PartitionedRegion pr = (PartitionedRegion) createRootRegion(name, factory.create());
        assertNotNull(pr);
      }
    };
    vm3.invoke(create);
    vm2.invoke(create);

    final int bucketsToCreate = 3;
    final SerializableRunnable createBuckets = new SerializableRunnable("Create Buckets") {
      @Override
      public void run() {
        final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
        assertNotNull(pr);
        // Create three buckets, add enough stuff in each to force two
        // overflow
        // ops in each
        for (int i = 0; i < bucketsToCreate; i++) {
          // assume mod-based hashing for bucket creation
          pr.put(i, "value0");
          pr.put(i + pr.getPartitionAttributes().getTotalNumBuckets(), "value1");
          pr.put(i + (pr.getPartitionAttributes().getTotalNumBuckets()) * 2, "value2");
        }
      }
    };
    vm3.invoke(createBuckets);

    final SerializableCallable assertBucketAttributesAndEviction =
        new SerializableCallable("Assert bucket attributes and eviction") {
          @Override
          public Object call() throws Exception {
            try {
              final Properties sp = getProperties();
              final int expectedInvocations = 10;
              final int maximumWaitSeconds = 60; // seconds
              final int pollWaitMillis = evictorInterval * 2;
              assertTrue(pollWaitMillis < (SECONDS.toMillis(maximumWaitSeconds) * 4));
              final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
              assertNotNull(pr);

              long entriesEvicted = 0;
              for (final Entry<Integer, BucketRegion> integerBucketRegionEntry : pr.getDataStore()
                  .getAllLocalBuckets()) {
                final Entry entry = (Entry) integerBucketRegionEntry;

                final BucketRegion bucketRegion = (BucketRegion) entry.getValue();
                if (bucketRegion == null) {
                  continue;
                }
                assertTrue(bucketRegion.getAttributes().getEvictionAttributes().getAlgorithm()
                    .isLRUHeap());
                assertTrue(bucketRegion.getAttributes().getEvictionAttributes().getAction()
                    .isLocalDestroy());
              }
              raiseFakeNotification();
              WaitCriterion wc = new WaitCriterion() {
                String excuse;

                @Override
                public boolean done() {
                  // we have a primary
                  return pr.getTotalEvictions() == 9;
                }

                @Override
                public String description() {
                  return excuse;
                }
              };
              GeodeAwaitility.await().untilAsserted(wc);

              entriesEvicted = pr.getTotalEvictions();
              return entriesEvicted;
            } finally {
              cleanUpAfterFakeNotification();
              // Clean up the observer for future tests
              // final Properties sp = System.getProperties();
              /*
               * assertNotNull(sp .remove(HeapLRUCapacityController.FREE_MEMORY_OBSERVER));
               */
            }
          }
        };
    final Long v2i = (Long) vm2.invoke(assertBucketAttributesAndEviction);
    final Long v3i = (Long) vm3.invoke(assertBucketAttributesAndEviction);
    final int totalEvicted = v2i.intValue() + v3i.intValue();
    // assume all three entries in each bucket were evicted to disk
    assertEquals((3 * bucketsToCreate * (redundantCopies + 1)), totalEvicted);
  }

  @Test
  public void testMemoryLRUWithOverflowToDisk() {
    final Host host = Host.getHost(0);
    final VM vm2 = host.getVM(2);
    final VM vm3 = host.getVM(3);
    final VM vm0 = host.getVM(0);
    final String uniqName = getUniqueName();
    final int redundantCopies = 1;
    final int maxBuckets = 8;
    final int localMaxMem = 16;
    final int halfKb = 512;
    final int justShyOfTwoMb = (2 * 1024 * 1024) - halfKb;

    final String name = uniqName + "-PR";
    final SerializableRunnable create =
        new SerializableRunnable("Create Memory LRU with Overflow to disk partitioned Region") {
          @Override
          public void run() {
            try {
              final AttributesFactory factory = new AttributesFactory();
              factory.setOffHeap(isOffHeap());
              factory.setPartitionAttributes(
                  new PartitionAttributesFactory().setRedundantCopies(redundantCopies)
                      .setLocalMaxMemory(localMaxMem).setTotalNumBuckets(maxBuckets).create());

              factory.setEvictionAttributes(EvictionAttributes
                  .createLRUMemoryAttributes(ObjectSizer.DEFAULT, EvictionAction.OVERFLOW_TO_DISK));
              factory.setDiskSynchronous(true);
              DiskStoreFactory dsf = getCache().createDiskStoreFactory();
              final File[] diskDirs = new File[1];
              diskDirs[0] = new File("overflowDir/" + uniqName + "_" + OSProcess.getId());
              diskDirs[0].mkdirs();
              dsf.setDiskDirs(diskDirs);
              DiskStore ds = dsf.create(name);
              factory.setDiskStoreName(ds.getName());
              final PartitionedRegion pr =
                  (PartitionedRegion) createRootRegion(name, factory.create());
              assertNotNull(pr);
            } catch (final CacheException ex) {
              Assert.fail("While creating Partitioned region", ex);
            }
          }
        };
    vm3.invoke(create);
    vm2.invoke(create);
    final int extraEntries = 4;
    final SerializableRunnable createBuckets = new SerializableRunnable("Create Buckets") {
      @Override
      public void run() {
        final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
        assertNotNull(pr);
        for (int counter = 1; counter <= localMaxMem + extraEntries; counter++) {
          pr.put(counter, new byte[1 * 1024 * 1024]);
        }
      }
    };
    vm3.invoke(createBuckets);

    final SerializableCallable assertBucketAttributesAndEviction =
        new SerializableCallable("Assert bucket attributes and eviction") {
          @Override
          public Object call() throws Exception {
            final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
            assertNotNull(pr);
            assertNull(pr.getDiskRegion());

            // assert over-flow behavior in local buckets and number of
            // entries
            // overflowed
            long entriesEvicted = 0;
            for (final Entry<Integer, BucketRegion> integerBucketRegionEntry : pr.getDataStore()
                .getAllLocalBuckets()) {
              final Entry entry = (Entry) integerBucketRegionEntry;

              final BucketRegion bucketRegion = (BucketRegion) entry.getValue();
              if (bucketRegion == null) {
                continue;
              }
              assertTrue(bucketRegion.getAttributes().getEvictionAttributes().getAlgorithm()
                  .isLRUMemory());
              assertTrue(bucketRegion.getAttributes().getEvictionAttributes().getAction()
                  .isOverflowToDisk());

            }
            entriesEvicted += pr.getDiskRegionStats().getNumOverflowOnDisk();
            return entriesEvicted;
          }
        };
    final Long vm2i = (Long) vm2.invoke(assertBucketAttributesAndEviction);
    final Long vm3i = (Long) vm3.invoke(assertBucketAttributesAndEviction);
    final int totalEvicted = vm2i.intValue() + vm3i.intValue();
    assertTrue(2 * extraEntries <= totalEvicted);

    // Test for bug 42056 - make sure we get the faulted out entries.
    vm0.invoke(create);

    vm0.invoke(new SerializableRunnable("Test to see that we can get keys") {
      @Override
      public void run() {
        final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
        // Make sure we can get all of the entries
        for (int counter = 1; counter <= localMaxMem + extraEntries; counter++) {
          assertNotNull(pr.get(counter));
        }
      }
    });
  }

  @Test
  public void testMemoryLRUWithLocalDestroy() {
    final Host host = Host.getHost(0);
    final VM vm2 = host.getVM(2);
    final VM vm3 = host.getVM(3);
    final String uniqName = getUniqueName();
    final int redundantCopies = 1;
    final int maxBuckets = 8;
    final int localMaxMem = 16;
    final int halfKb = 512;
    final int justShyOfTwoMb = (2 * 1024 * 1024) - halfKb;
    final String name = uniqName + "-PR";

    final SerializableRunnable create = new CacheSerializableRunnable(
        "Create Memory LRU with local destroy on a partitioned Region") {
      @Override
      public void run2() {
        final AttributesFactory factory = new AttributesFactory();
        factory.setOffHeap(isOffHeap());
        factory.setPartitionAttributes(
            new PartitionAttributesFactory().setRedundantCopies(redundantCopies)
                .setLocalMaxMemory(16).setTotalNumBuckets(maxBuckets).create());
        factory.setEvictionAttributes(EvictionAttributes
            .createLRUMemoryAttributes(ObjectSizer.DEFAULT, EvictionAction.LOCAL_DESTROY));
        final PartitionedRegion pr = (PartitionedRegion) createRootRegion(name, factory.create());
        assertNotNull(pr);
      }
    };
    vm3.invoke(create);
    vm2.invoke(create);

    final int extraEntries = 4;
    final SerializableRunnable createBuckets = new SerializableRunnable("Create Buckets") {
      @Override
      public void run() {
        final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
        assertNotNull(pr);
        for (int counter = 1; counter <= localMaxMem + extraEntries; counter++) {
          pr.put(counter, new byte[1 * 1024 * 1024]);
        }
      }
    };
    vm3.invoke(createBuckets);

    final SerializableCallable assertBucketAttributesAndEviction =
        new SerializableCallable("Assert bucket attributes and eviction") {
          @Override
          public Object call() throws Exception {
            try {
              final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
              assertNotNull(pr);
              long entriesEvicted = 0;
              for (final Entry<Integer, BucketRegion> integerBucketRegionEntry : pr.getDataStore()
                  .getAllLocalBuckets()) {
                final Entry entry = (Entry) integerBucketRegionEntry;

                final BucketRegion bucketRegion = (BucketRegion) entry.getValue();
                if (bucketRegion == null) {
                  continue;
                }
                assertTrue(bucketRegion.getAttributes().getEvictionAttributes().getAlgorithm()
                    .isLRUMemory());
                assertTrue(bucketRegion.getAttributes().getEvictionAttributes().getAction()
                    .isLocalDestroy());
              }
              entriesEvicted = pr.getTotalEvictions();
              return entriesEvicted;
            } finally {
            }
          }
        };
    final Long v2i = (Long) vm2.invoke(assertBucketAttributesAndEviction);
    final Long v3i = (Long) vm3.invoke(assertBucketAttributesAndEviction);
    final int totalEvicted = v2i.intValue() + v3i.intValue();
    assertTrue(2 * extraEntries <= totalEvicted);
  }

  @Test
  public void testEntryLRUWithOverflowToDisk() {
    final Host host = Host.getHost(0);
    final VM vm2 = host.getVM(2);
    final VM vm3 = host.getVM(3);
    final String uniqName = getUniqueName();
    final int redundantCopies = 1;
    final int maxBuckets = 8;
    final int maxEntries = 16;
    final String name = uniqName + "-PR";
    final SerializableRunnable create =
        new SerializableRunnable("Create Entry LRU with Overflow to disk partitioned Region") {
          @Override
          public void run() {
            try {
              final AttributesFactory factory = new AttributesFactory();
              factory.setOffHeap(isOffHeap());
              factory.setPartitionAttributes(new PartitionAttributesFactory()
                  .setRedundantCopies(redundantCopies).setTotalNumBuckets(maxBuckets).create());
              factory.setEvictionAttributes(EvictionAttributes.createLRUEntryAttributes(maxEntries,
                  EvictionAction.OVERFLOW_TO_DISK));
              factory.setDiskSynchronous(true);
              DiskStoreFactory dsf = getCache().createDiskStoreFactory();
              final File[] diskDirs = new File[1];
              diskDirs[0] = new File("overflowDir/" + uniqName + "_" + OSProcess.getId());
              diskDirs[0].mkdirs();
              dsf.setDiskDirs(diskDirs);
              DiskStore ds = dsf.create(name);
              factory.setDiskStoreName(ds.getName());
              final PartitionedRegion pr =
                  (PartitionedRegion) createRootRegion(name, factory.create());
              assertNotNull(pr);
            } catch (final CacheException ex) {
              Assert.fail("While creating Partitioned region", ex);
            }
          }
        };
    vm3.invoke(create);
    vm2.invoke(create);
    final int extraEntries = 4;
    final SerializableRunnable createBuckets = new SerializableRunnable("Create Buckets") {
      @Override
      public void run() {
        final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
        assertNotNull(pr);
        for (int counter = 1; counter <= maxEntries + extraEntries; counter++) {
          pr.put(counter, new byte[1 * 1024 * 1024]);
        }
      }
    };
    vm3.invoke(createBuckets);

    final SerializableCallable assertBucketAttributesAndEviction =
        new SerializableCallable("Assert bucket attributes and eviction") {
          @Override
          public Object call() throws Exception {
            final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
            assertNotNull(pr);

            // assert over-flow behavior in local buckets and number of
            // entries
            // overflowed
            int entriesEvicted = 0;
            for (final Entry<Integer, BucketRegion> integerBucketRegionEntry : pr.getDataStore()
                .getAllLocalBuckets()) {
              final Entry entry = (Entry) integerBucketRegionEntry;

              final BucketRegion bucketRegion = (BucketRegion) entry.getValue();
              if (bucketRegion == null) {
                continue;
              }
              assertTrue(
                  bucketRegion.getAttributes().getEvictionAttributes().getAlgorithm().isLRUEntry());
              assertTrue(bucketRegion.getAttributes().getEvictionAttributes().getAction()
                  .isOverflowToDisk());
            }
            entriesEvicted += pr.getDiskRegionStats().getNumOverflowOnDisk();
            return entriesEvicted;

          }
        };
    final Integer vm2i = (Integer) vm2.invoke(assertBucketAttributesAndEviction);
    final Integer vm3i = (Integer) vm3.invoke(assertBucketAttributesAndEviction);
    final int totalEvicted = vm2i + vm3i;
    assertEquals(extraEntries * 2, totalEvicted);
  }

  private static class VerifiableCacheListener extends CacheListenerAdapter
      implements java.io.Serializable {
    public boolean verify(long expectedEvictions) {
      return false;
    }
  }

  @Test
  public void testEntryLRUWithLocalDestroy() {
    final Host host = Host.getHost(0);
    final VM vm2 = host.getVM(2);
    final VM vm3 = host.getVM(3);
    final String uniqName = getUniqueName();
    final int redundantCopies = 1;
    final int maxBuckets = 8;
    final int maxEntries = 16;
    final String name = uniqName + "-PR";
    final int extraEntries = 4;

    // final int heapPercentage = 66;
    // final int evictorInterval = 100;

    final SerializableRunnable create = new CacheSerializableRunnable(
        "Create Entry LRU with local destroy on a partitioned Region") {
      @Override
      public void run2() {
        final AttributesFactory factory = new AttributesFactory();
        factory.setOffHeap(isOffHeap());
        factory.setPartitionAttributes(new PartitionAttributesFactory()
            .setRedundantCopies(redundantCopies).setTotalNumBuckets(maxBuckets).create());
        factory.setEvictionAttributes(
            EvictionAttributes.createLRUEntryAttributes(maxEntries, EvictionAction.LOCAL_DESTROY));
        factory.addCacheListener(new VerifiableCacheListener() {
          private long evictionDestroyEvents = 0;

          @Override
          public void afterDestroy(EntryEvent e) {
            System.out.println("EEEEEEEEEEEEEE key:" + e.getKey());
            EntryEventImpl eei = (EntryEventImpl) e;
            if (Operation.EVICT_DESTROY.equals(eei.getOperation())) {
              evictionDestroyEvents++;
            }
          }

          @Override
          public boolean verify(long expectedEntries) {
            return expectedEntries == evictionDestroyEvents;
          }
        });
        final PartitionedRegion pr = (PartitionedRegion) createRootRegion(name, factory.create());
        assertNotNull(pr);
      }
    };
    vm3.invoke(create);

    final SerializableRunnable create2 =
        new SerializableRunnable("Create Entry LRU with local destroy on a partitioned Region") {
          @Override
          public void run() {
            try {
              final AttributesFactory factory = new AttributesFactory();
              factory.setOffHeap(isOffHeap());
              factory.setPartitionAttributes(new PartitionAttributesFactory()
                  .setRedundantCopies(redundantCopies).setTotalNumBuckets(8).create());
              factory
                  .setEvictionAttributes(EvictionAttributes.createLRUEntryAttributes(maxEntries));
              final PartitionedRegion pr =
                  (PartitionedRegion) createRootRegion(name, factory.create());
              assertNotNull(pr);
            } catch (final CacheException ex) {
              Assert.fail("While creating Partitioned region", ex);
            }
          }
        };
    vm2.invoke(create2);

    final SerializableRunnable createBuckets = new SerializableRunnable("Create Buckets") {
      @Override
      public void run() {
        final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
        assertNotNull(pr);
        for (int counter = 1; counter <= maxEntries + extraEntries; counter++) {
          pr.put(counter, new byte[1 * 1024 * 1024]);
        }
      }
    };
    vm3.invoke(createBuckets);

    final SerializableCallable assertBucketAttributesAndEviction =
        new SerializableCallable("Assert bucket attributes and eviction") {
          @Override
          public Object call() throws Exception {
            try {
              final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
              assertNotNull(pr);
              for (final Entry<Integer, BucketRegion> integerBucketRegionEntry : pr.getDataStore()
                  .getAllLocalBuckets()) {
                final Entry entry = (Entry) integerBucketRegionEntry;

                final BucketRegion bucketRegion = (BucketRegion) entry.getValue();
                if (bucketRegion == null) {
                  continue;
                }
                assertTrue(bucketRegion.getAttributes().getEvictionAttributes().getAlgorithm()
                    .isLRUEntry());
                assertTrue(bucketRegion.getAttributes().getEvictionAttributes().getAction()
                    .isLocalDestroy());
              }
              return pr.getTotalEvictions();
            } finally {
            }
          }
        };

    final Long v2i = (Long) vm2.invoke(assertBucketAttributesAndEviction);
    final Long v3i = (Long) vm3.invoke(assertBucketAttributesAndEviction);
    final int totalEvicted = v2i.intValue() + v3i.intValue();
    assertEquals(2 * extraEntries, totalEvicted);

    final SerializableCallable assertListenerCount = new SerializableCallable(
        "Assert that the number of listener invocations matches the expected total") {
      @Override
      public Object call() throws Exception {
        final PartitionedRegion pr = (PartitionedRegion) getRootRegion(name);
        assertNotNull(pr);
        RegionAttributes attrs = pr.getAttributes();
        assertNotNull(attrs);
        long entriesEvicted = pr.getTotalEvictions();
        VerifiableCacheListener verifyMe = null;
        for (CacheListener listener : attrs.getCacheListeners()) {
          if (listener instanceof VerifiableCacheListener) {
            verifyMe = ((VerifiableCacheListener) listener);
          }
        }
        assertNotNull(verifyMe); // assert if unable to find the expected listener
        return verifyMe.verify(entriesEvicted);
      }
    };
    assertTrue((Boolean) vm3.invoke(assertListenerCount));
  }

  // Test to validate the Eviction Attribute : LRU Check
  @Test
  public void testEvictionValidationForLRUEntry() {
    final Host host = Host.getHost(0);
    final VM testAccessor = host.getVM(1);
    final VM testDatastore = host.getVM(2);
    final VM firstDatastore = host.getVM(3);
    final String uniqName = getUniqueName();
    final int redundantCopies = 1;
    final int maxEntries = 226;
    final String name = uniqName + "-PR";

    // final int evictionSizeInMB = 200;
    final EvictionAttributes firstEvictionAttrs =
        EvictionAttributes.createLRUEntryAttributes(maxEntries, EvictionAction.LOCAL_DESTROY);

    // Creating LRU Entry Count Eviction Attribute
    final SerializableRunnable create = new CacheSerializableRunnable(
        "Create Entry LRU with local destroy on a partitioned Region") {
      @Override
      public void run2() {

        // Assert that LRUEntry maximum can be less than 1 entry per
        // bucket
        // DUnit not required for this test, but its a convenient place
        // to put it.
        {
          final int buks = 11;
          final AttributesFactory factory = new AttributesFactory();
          factory.setOffHeap(isOffHeap());
          factory.setPartitionAttributes(new PartitionAttributesFactory().setTotalNumBuckets(buks)
              .setRedundantCopies(0).create());
