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

import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.junit.Assert.assertEquals;

import java.util.Properties;

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

import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.distributed.ConfigurationProperties;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.test.junit.categories.IntegrationTest;

/**
 * Test to reproduce GEODE-2097.
 */
@Category(IntegrationTest.class)
public class OffHeapLRURecoveryRegressionTest {

  static final String DS_NAME = "OffHeapLRURecoveryRegressionTestDS";

  /**
   * Test populates an offheap heaplru persistent region that contains more data than can fit in
   * offheap memory. It then recovers the region to demonstrate that recovering this data will not
   * try to put everything into offheap but instead leave some of it on disk.
   */
  @Test
  public void recoveringTooMuchDataDoesNotRunOutOfOffHeapMemory() {
    final int ENTRY_COUNT = 40;
    final int expectedObjectCount;
    GemFireCacheImpl gfc = createCache();
    try {
      Region<Object, Object> r = createRegion(gfc);
      byte[] v = new byte[1024 * 1024];
      for (int i = 0; i < ENTRY_COUNT; i++) {
        r.put(i, v);
      }
      // expect one more during recovery because of the way the LRU limit is
      // enforced during recover.
      expectedObjectCount = MemoryAllocatorImpl.getAllocator().getStats().getObjects() + 1;
    } finally {
      gfc.close();
    }
    System.setProperty("gemfire.disk.recoverValuesSync", "true");
    System.setProperty("gemfire.disk.recoverLruValues", "true");
    try {
      gfc = createCache();
      try {
        Region<Object, Object> r = createRegion(gfc);
        try {
          assertEquals(ENTRY_COUNT, r.size());
          assertEquals(expectedObjectCount,
              MemoryAllocatorImpl.getAllocator().getStats().getObjects());
        } finally {
          r.destroyRegion();
          DiskStore ds = gfc.findDiskStore(DS_NAME);
          ds.destroy();
        }
      } finally {
        gfc.close();
      }
    } finally {
      System.clearProperty("gemfire.disk.recoverValuesSync");
      System.clearProperty("gemfire.disk.recoverLruValues");
    }
  }

  private GemFireCacheImpl createCache() {
    Properties props = new Properties();
    props.setProperty(LOCATORS, "");
    props.setProperty(MCAST_PORT, "0");
    props.setProperty(ConfigurationProperties.OFF_HEAP_MEMORY_SIZE, "20m");
    GemFireCacheImpl result = (GemFireCacheImpl) new CacheFactory(props).create();
    result.getResourceManager().setEvictionOffHeapPercentage(50.0f);
    return result;
  }

  private Region<Object, Object> createRegion(GemFireCacheImpl gfc) {
    DiskStoreFactory dsf = gfc.createDiskStoreFactory();
    dsf.create(DS_NAME);
    RegionFactory<Object, Object> rf =
        gfc.createRegionFactory(RegionShortcut.LOCAL_PERSISTENT_OVERFLOW);
    rf.setOffHeap(true);
    rf.setDiskStoreName(DS_NAME);
    Region<Object, Object> r = rf.create("OffHeapLRURecoveryRegressionTestRegion");
    return r;
  }

  private void closeCache(GemFireCacheImpl gfc) {
    gfc.close();
    MemoryAllocatorImpl.freeOffHeapMemory();
  }
}

@@ -1058,16 +1058,88 @@
 erties;%0A
+import java.util.concurrent.TimeUnit;%0A%0Aimport org.awaitility.Awaitility;
 %0Aimport 
@@ -2242,43 +2242,8 @@
 40;%0A
-    final int expectedObjectCount;%0A
     
@@ -2465,142 +2465,127 @@
     
-  // expect one more during recovery because of the way the LRU limit is%0A      // enforced during recover.%0A      expectedObjectCount =
+%7D finally %7B%0A      closeCache(gfc);%0A    %7D%0A    Awaitility.await().atMost(60, TimeUnit.SECONDS).until(() -%3E %7B%0A      return
  Mem
@@ -2644,53 +2644,21 @@
 s() 
-+ 1
+== 0
 ;%0A    %7D
- finally %7B%0A      gfc.close();%0A    %7D
+);
 %0A   
@@ -2846,47 +2846,23 @@
     
-Region%3CObject, Object%3E r = createRegion
+createDiskStore
 (gfc
@@ -2905,89 +2905,11 @@
 als(
-ENTRY_COUNT, r.size());%0A          assertEquals(expectedObjectCount,%0A             
+10,
  Mem
@@ -2990,37 +2990,8 @@
 y %7B%0A
-          r.destroyRegion();%0A
     
@@ -3095,34 +3095,38 @@
 y %7B%0A        
-gfc.
 close
-(
+Cache(gfc
 );%0A      %7D%0A 
@@ -3701,43 +3701,28 @@
 ate 
-Region%3CObject, Object%3E createRegion
+void createDiskStore
 (Gem
@@ -3816,32 +3816,133 @@
 reate(DS_NAME);%0A
+  %7D%0A%0A  private Region%3CObject, Object%3E createRegion(GemFireCacheImpl gfc) %7B%0A    createDiskStore(gfc);%0A
     RegionFactor
@@ -4272,53 +4272,8 @@
 ();%0A
-    MemoryAllocatorImpl.freeOffHeapMemory();%0A
   %7D%0A

fix flaky test