/*
 * 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.util.concurrent.TimeUnit.MINUTES;
import static org.apache.geode.cache.EvictionAction.OVERFLOW_TO_DISK;
import static org.apache.geode.cache.EvictionAttributes.DEFAULT_ENTRIES_MAXIMUM;
import static org.apache.geode.cache.EvictionAttributes.createLRUEntryAttributes;
import static org.apache.geode.cache.RegionShortcut.LOCAL;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;

import java.io.File;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.lang.reflect.Array;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.contrib.java.lang.system.RestoreSystemProperties;
import org.junit.rules.ErrorCollector;
import org.junit.rules.TemporaryFolder;
import org.junit.rules.TestName;

import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache.util.ObjectSizer;
import org.apache.geode.internal.cache.entries.DiskEntry;
import org.apache.geode.internal.cache.eviction.EvictionCounters;
import org.apache.geode.internal.cache.persistence.DiskRecoveryStore;
import org.apache.geode.internal.cache.persistence.UninterruptibleFileChannel;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.junit.rules.ExecutorServiceRule;

/**
 * Integration tests covering some miscellaneous functionality of Disk Region.
 */
public class DiskRegionJUnitTest {

  private static final long MAX_OPLOG_SIZE_IN_BYTES = 1024 * 1024 * 1024 * 10L;

  private Properties config;
  private InternalCache cache;
  private EvictionAttributes heapEvictionAttributes;

  private File[] diskDirs;
  private int[] diskDirSizes;

  private String uniqueName;
  private String regionName;
  private String diskStoreName;

  @Rule
  public ErrorCollector errorCollector = new ErrorCollector();

  @Rule
  public ExecutorServiceRule executorServiceRule = new ExecutorServiceRule();

  @Rule
  public RestoreSystemProperties restoreSystemProperties = new RestoreSystemProperties();

  @Rule
  public TemporaryFolder temporaryFolder = new TemporaryFolder();

  @Rule
  public TestName testName = new TestName();

  @Before
  public void setUp() throws Exception {
    uniqueName = getClass().getSimpleName() + "_" + testName.getMethodName();
    regionName = uniqueName + "_region";
    diskStoreName = uniqueName + "_diskStore";

    config = new Properties();
    config.setProperty(MCAST_PORT, "0");
    config.setProperty(LOCATORS, "");

    cache = (InternalCache) new CacheFactory(config).create();

    heapEvictionAttributes = EvictionAttributes.createLRUHeapAttributes(ObjectSizer.DEFAULT,
        OVERFLOW_TO_DISK);

    diskDirs = new File[4];
    diskDirs[0] = createDirectory(temporaryFolder.getRoot(), testName.getMethodName() + "1");
    diskDirs[1] = createDirectory(temporaryFolder.getRoot(), testName.getMethodName() + "2");
    diskDirs[2] = createDirectory(temporaryFolder.getRoot(), testName.getMethodName() + "3");
    diskDirs[3] = createDirectory(temporaryFolder.getRoot(), testName.getMethodName() + "4");

    // set default values of disk dir sizes here
    diskDirSizes = new int[4];
    diskDirSizes[0] = Integer.MAX_VALUE;
    diskDirSizes[1] = Integer.MAX_VALUE;
    diskDirSizes[2] = Integer.MAX_VALUE;
    diskDirSizes[3] = Integer.MAX_VALUE;

    DiskStoreImpl.SET_IGNORE_PREALLOCATE = true;
  }

  @After
  public void tearDown() throws Exception {
    try {
      if (cache != null) {
        cache.close();
      }
    } finally {
      CacheObserverHolder.setInstance(null);
      DiskStoreImpl.SET_IGNORE_PREALLOCATE = false;
      LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false;
    }
  }

  @Test
  public void testRemoveCorrectlyRecorded() {
    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setOverflow(true);
    diskRegionProperties.setOverFlowCapacity(1);
    diskRegionProperties.setDiskDirs(diskDirs);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory);

    Region<String, String> region =
        createRegion(regionName, diskStoreName, true, true, false, true, 1);

    region.put("1", "1");
    region.put("2", "2");
    region.put("3", "3");

    AfterDestroyListener<String, String> cacheListener = new AfterDestroyListener<>();
    region.getAttributesMutator().addCacheListener(cacheListener);

    region.destroy("1");

    // Make sure we don't get an old value when doing a destroy of an entry that overflowed to disk.
    // If we do then we have hit bug 40795.
    assertThat(cacheListener.getLastEvent()).isNotNull();
    assertThat(cacheListener.getLastEvent().getOldValue()).isNull();

    assertThat(region.get("1")).isNull();

    // does the following ever throw EntryNotFoundException? that would be ok
    Object valueOnDisk = ((InternalRegion) region).getValueOnDisk("1");
    assertThat(valueOnDisk == null || valueOnDisk.equals(Token.TOMBSTONE)).isTrue();

    region.close();

    Region<String, String> region2 =
        createRegion(regionName, diskStoreName, true, true, false, true, 1);
    assertThat(region2.get("1")).isNull();
  }

  /**
   * Tests if region overflows correctly and stats are create and updated correctly.
   */
  @Test
  public void testDiskRegionOverflow() {
    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setOverflow(true);
    diskRegionProperties.setOverFlowCapacity(5);
    diskRegionProperties.setDiskDirs(diskDirs);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory);

    Region<Integer, Object> region =
        createRegion(regionName, diskStoreName, false, false, false, true, 5);

    DiskRegion diskRegion = getDiskRegion(region);

    DiskRegionStats diskStats = diskRegion.getStats();
    EvictionCounters evictionCounters =
        ((InternalRegion) region).getEvictionController().getCounters();

    assertThat(diskStats).isNotNull();
    assertThat(evictionCounters).isNotNull();

    diskRegion.flushForTesting();

    assertThat(diskStats.getWrites()).isEqualTo(0);
    assertThat(diskStats.getReads()).isEqualTo(0);
    assertThat(evictionCounters.getEvictions()).isEqualTo(0);

    // Put in larger stuff until we start evicting
    int total;
    for (total = 0; evictionCounters.getEvictions() <= 0; total++) {
      int[] array = new int[250];
      array[0] = total;
      region.put(total, array);
    }

    diskRegion.flushForTesting();

    assertThat(diskStats.getWrites()).isEqualTo(1);
    assertThat(diskStats.getReads()).isEqualTo(0);
    assertThat(diskStats.getNumEntriesInVM()).isEqualTo(total - 1);
    assertThat(diskStats.getNumOverflowOnDisk()).isEqualTo(1);
    assertThat(evictionCounters.getEvictions()).isEqualTo(1);

    Object value = region.get(0);
    assertThat(value).isNotNull().isInstanceOf(int[].class);
    assertThat(((int[]) value)[0]).isEqualTo(0);

    diskRegion.flushForTesting();

    assertThat(diskStats.getWrites()).isEqualTo(2);
    assertThat(diskStats.getReads()).isEqualTo(1);
    assertThat(evictionCounters.getEvictions()).isEqualTo(2);

    for (int i = 0; i < total; i++) {
      int[] array = (int[]) region.get(i);
      assertThat(array).isNotNull().isInstanceOf(int[].class);
      assertThat(array[0]).isEqualTo(i);
    }
  }

  /**
   * test method for putting different objects and validating that they have been correctly put
   */
  @Test
  public void testDifferentObjectTypePuts() {
    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setOverflow(true);
    diskRegionProperties.setOverFlowCapacity(100);
    diskRegionProperties.setDiskDirs(diskDirs);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory);

    Region<String, Object> region =
        createRegion(regionName, diskStoreName, true, false, false, true, 100);

    DiskRegion diskRegion = getDiskRegion(region);

    int total = 10;
    for (int i = 0; i < total; i++) {
      region.put(String.valueOf(i), String.valueOf(i));
    }
    region.put("foobar", "junk");

    region.localDestroy("foobar");

    region.put("foobar2", "junk");
    diskRegion.flushForTesting();
    region.localDestroy("foobar2");

    // test invalidate
    region.put("invalid", "invalid");
    diskRegion.flushForTesting();
    region.invalidate("invalid");
    diskRegion.flushForTesting();
    assertThat(region.containsKey("invalid")).isTrue();
    assertThat(region.containsValueForKey("invalid")).isFalse();
    total++;

    // test local-invalidate
    region.put("localinvalid", "localinvalid");
    diskRegion.flushForTesting();
    region.localInvalidate("localinvalid");
    diskRegion.flushForTesting();
    assertThat(region.containsKey("localinvalid")).isTrue();
    assertThat(region.containsValueForKey("localinvalid")).isFalse();
    total++;

    // test byte[] values
    region.put("byteArray", new byte[0]);
    diskRegion.flushForTesting();
    assertThatArrayEquals(new byte[0], region.get("byteArray"));
    total++;

    // test modification
    region.put("modified", "originalValue");
    diskRegion.flushForTesting();
    region.put("modified", "modified");
    diskRegion.flushForTesting();
    assertThat(region.get("modified")).isEqualTo("modified");
    total++;

    assertThat(region.size()).isEqualTo(total);

    cache.close();
    cache = (InternalCache) new CacheFactory(config).create();

    diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory);

    region = createRegion(regionName, diskStoreName, true, false, false, true, 100);

    assertThat(region.size()).isEqualTo(total);
    assertThat(region.containsKey("invalid")).isTrue();
    assertThat(region.get("invalid")).isNull();
    assertThat(region.containsValueForKey("invalid")).isFalse();

    region.localDestroy("invalid");
    total--;

    assertThat(region.containsKey("localinvalid")).isTrue();
    assertThat(region.containsValueForKey("localinvalid")).isFalse();
    region.localDestroy("localinvalid");
    total--;

    assertThatArrayEquals(new byte[0], region.get("byteArray"));
    region.localDestroy("byteArray");
    total--;

    assertThat(region.get("modified")).isEqualTo("modified");
    region.localDestroy("modified");
    total--;

    assertThat(region.size()).isEqualTo(total);
  }

  @Test
  public void testFaultingInRemovalFromAsyncBuffer() throws Exception {
    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setOverflow(true);
    diskRegionProperties.setRolling(true);
    diskRegionProperties.setOverFlowCapacity(100);
    diskRegionProperties.setDiskDirs(diskDirs);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory);

    Region<Integer, Integer> region =
        createRegion(regionName, diskStoreName, false, true, false, true, 100);

    CountDownLatch doLatch = new CountDownLatch(1);
    DelayedGet get = new DelayedGet(doLatch, region);

    CompletableFuture<Void> getter1 = executorServiceRule.runAsync(get);
    CompletableFuture<Void> getter2 = executorServiceRule.runAsync(get);
    CompletableFuture<Void> getter3 = executorServiceRule.runAsync(get);
    CompletableFuture<Void> getter4 = executorServiceRule.runAsync(get);
    CompletableFuture<Void> getter5 = executorServiceRule.runAsync(get);

    for (int i = 0; i < 110; i++) {
      region.put(i, i);
    }

    doLatch.countDown();

    CompletableFuture.allOf(getter1, getter2, getter3, getter4, getter5).get(2, MINUTES);
  }

  /**
   * DiskDirectoriesJUnitTest:
   *
   * This tests the potential deadlock situation if the region is created such that rolling is
   * turned on but the Max directory space is less than or equal to the Max Oplog Size. In such
   * situations , if during switch over , if the Oplog to be rolled is added after function call of
   * obtaining nextDir , a dead lock occurs
   */
  @Test
  public void testSingleDirectoryNotHanging() {
    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setDiskDirsAndSizes(new File[] {diskDirs[0]}, new int[] {2048});
    diskRegionProperties.setMaxOplogSize(2097152);
    diskRegionProperties.setRolling(true);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory);

    Region<String, byte[]> region =
        createRegion(regionName, diskStoreName, true, true, false, false, 0);

    Puts puts = new Puts(region);
    puts.performPuts();

    assertThat(puts.putSuccessful(0)).as(" first put did not succeed").isTrue();
    assertThat(puts.putSuccessful(1)).as(" second put did not succeed").isTrue();
    assertThat(puts.putSuccessful(2)).as(" third put did not succeed").isTrue();

    assertThat(puts.diskAccessExceptionOccurred())
        .as(" Exception was not supposed to occur but did occur").isFalse();
  }

  @Test
  public void testOperationGreaterThanMaxOplogSize() {
    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setDiskDirs(diskDirs);
    diskRegionProperties.setMaxOplogSize(512);
    diskRegionProperties.setRolling(true);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory);

    Region<String, byte[]> region =
        createRegion(regionName, diskStoreName, true, true, false, false, 0);

    Puts puts = new Puts(region);
    puts.performPuts();

    assertThat(puts.putSuccessful(0)).as(" first put did not succeed").isTrue();
    assertThat(puts.putSuccessful(1)).as(" second put did not succeed").isTrue();
    assertThat(puts.putSuccessful(2)).as(" third put did not succeed").isTrue();

    assertThat(puts.diskAccessExceptionOccurred())
        .as(" Exception was not supposed to occur but did occur").isFalse();
  }

  /**
   * As we have relaxed the constraint of max dir size
   */
  @Test
  public void testOperationGreaterThanMaxDirSize() {
    int[] diskDirSizes = {1025, 1025, 1025, 1025};

    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setRegionName(regionName);
    diskRegionProperties.setDiskDirsAndSizes(diskDirs, diskDirSizes);
    diskRegionProperties.setMaxOplogSize(600);
    diskRegionProperties.setRolling(false);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    DiskStore diskStore = createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory);

    Region<String, byte[]> region =
        createRegion(regionName, diskStoreName, true, true, false, false, 0);

    assertThat(diskStore.getDiskDirSizes())
        .as("expected=" + Arrays.toString(diskDirSizes) + " actual="
            + Arrays.toString(diskStore.getDiskDirSizes()))
        .isEqualTo(diskDirSizes);

    Puts puts = new Puts(region, 1026);
    puts.performPuts();

    assertThat(puts.diskAccessExceptionOccurred())
        .as(" Exception was not supposed to occur but did occur").isTrue();

    assertThat(puts.putSuccessful(0)).as(" first put did not succeed").isFalse();
    assertThat(puts.putSuccessful(1)).as(" second put did not succeed").isFalse();
    assertThat(puts.putSuccessful(2)).as(" third put did not succeed").isFalse();

    // if the exception occurred then the region should be closed already
    verifyClosedDueToDiskAccessException(region);
  }

  /**
   * Regression test for TRAC #42464: overfilling disk directory causes rampant oplog creation
   *
   * <p>
   * When max-dir-size is exceeded and compaction is enabled we allow oplogs to keep getting
   * created. Make sure that when they do they do not keep putting one op per oplog (which is caused
   * by bug 42464).
   */
  @Test
  public void testBug42464() {
    int[] dirSizes = new int[] {900};

    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setDiskDirsAndSizes(new File[] {diskDirs[0]}, dirSizes);
    diskRegionProperties.setMaxOplogSize(500);
    diskRegionProperties.setRolling(true);
    diskRegionProperties.setOverFlowCapacity(1);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    DiskStore diskStore = createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory, 500);

    Region<Integer, Object> region =
        createRegion(regionName, diskStoreName, false, true, false, true, 1);

    assertThat(diskStore.getDiskDirSizes()).as("expected=" + Arrays.toString(dirSizes) + " actual="
        + Arrays.toString(diskStore.getDiskDirSizes())).isEqualTo(dirSizes);

    // One entry is kept in memory
    // since the crf max is 500 we should only be able to have 4 entries. The
    // 5th should not fit because of record overhead.
    for (int i = 0; i <= 9; i++) {
      region.getCache().getLogger().info("putting " + i);
      region.put(i, new byte[101]);
    }

    // At this point we should have two oplogs that are basically full
    // (they should each contain 4 entries) and a third oplog that
    // contains a single entry. But the 3rd one will end up also containing 4
    // entries.
    // TODO what is the max size of this 3rd oplog's crf? The first two crfs
    // will be close to 400 bytes each. So the max size of the 3rd oplog should
    // be close to 100.
    ArrayList<OverflowOplog> oplogs = toDiskStoreImpl(diskStore).testHookGetAllOverflowOplogs();
    assertThat(oplogs).hasSize(3);

    // TODO verify oplogs
    // Now make sure that further oplogs can hold 4 entries
    for (int j = 10; j <= 13; j++) {
      region.put(j, new byte[101]);
    }
    oplogs = toDiskStoreImpl(diskStore).testHookGetAllOverflowOplogs();
    assertThat(oplogs).hasSize(4);

    // now remove all entries and make sure old oplogs go away
    for (int i = 0; i <= 13; i++) {
      region.remove(i);
    }

    // give background compactor chance to remove oplogs
    oplogs = toDiskStoreImpl(diskStore).testHookGetAllOverflowOplogs();
    int retryCount = 20;
    while (oplogs.size() > 1 && retryCount > 0) {
      Wait.pause(100);
      oplogs = toDiskStoreImpl(diskStore).testHookGetAllOverflowOplogs();
      retryCount--;
    }
    assertThat(oplogs).hasSize(1);
  }

  @Test
  public void testSingleDirectorySizeViolation() {
    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setRegionName(regionName);
    diskRegionProperties.setDiskDirsAndSizes(new File[] {diskDirs[0]}, new int[] {2048});
    diskRegionProperties.setMaxOplogSize(2097152);
    diskRegionProperties.setRolling(false);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory);

    Region<String, byte[]> region =
        createRegion(regionName, diskStoreName, true, true, false, false, 0);

    Puts puts = new Puts(region);
    puts.performPuts();

    assertThat(puts.putSuccessful(0)).as(" first put did not succeed").isTrue();
    assertThat(puts.putSuccessful(1)).as(" second put should have failed").isFalse();
    assertThat(puts.putSuccessful(2)).as(" third put should have failed").isFalse();

    assertThat(puts.diskAccessExceptionOccurred())
        .as(" Exception was supposed to occur but did not occur").isTrue();

    verifyClosedDueToDiskAccessException(region);
  }

  /**
   * DiskRegDiskAccessExceptionTest : Disk region test for DiskAccessException.
   */
  @Test
  public void testDiskFullExcep() {
    int[] diskDirSizes = new int[4];
    diskDirSizes[0] = 2048 + 500;
    diskDirSizes[1] = 2048 + 500;
    diskDirSizes[2] = 2048 + 500;
    diskDirSizes[3] = 2048 + 500;

    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setDiskDirsAndSizes(diskDirs, diskDirSizes);
    diskRegionProperties.setPersistBackup(true);
    diskRegionProperties.setRolling(false);
    diskRegionProperties.setMaxOplogSize(1000000000);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory, 1_000_000_000);

    Region<Object, Object> region =
        createRegion(regionName, diskStoreName, true, true, false, false, 0);

    int[] actualDiskDirSizes = ((InternalRegion) region).getDiskDirSizes();

    assertThat(actualDiskDirSizes[0]).isEqualTo(2048 + 500);
    assertThat(actualDiskDirSizes[1]).isEqualTo(2048 + 500);
    assertThat(actualDiskDirSizes[2]).isEqualTo(2048 + 500);
    assertThat(actualDiskDirSizes[3]).isEqualTo(2048 + 500);

    // we have room for 2 values per dir

    byte[] value = new byte[1024];
    Arrays.fill(value, (byte) 77);

    for (int i = 0; i < 8; i++) {
      region.put(String.valueOf(i), value);
    }

    // we should have put 2 values in each dir so the next one should not fit
    try (IgnoredException ie = addIgnoredException(DiskAccessException.class)) {
      Throwable thrown = catchThrowable(() -> region.put("FULL", value));
      assertThat(thrown).isInstanceOf(DiskAccessException.class);
    }

    verifyClosedDueToDiskAccessException(region);
  }

  /**
   * Make sure if compaction is enabled that we can exceed the disk dir limit
   */
  @Test
  public void testNoDiskFullExcep() {
    int[] diskDirSizes = new int[4];
    diskDirSizes[0] = 2048 + 500;
    diskDirSizes[1] = 2048 + 500;
    diskDirSizes[2] = 2048 + 500;
    diskDirSizes[3] = 2048 + 500;

    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setDiskDirsAndSizes(diskDirs, diskDirSizes);
    diskRegionProperties.setPersistBackup(true);
    diskRegionProperties.setRolling(true);
    diskRegionProperties.setMaxOplogSize(1000000000);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory, 1_000_000_000);

    Region<Object, Object> region =
        createRegion(regionName, diskStoreName, true, true, false, false, 0);

    int[] actualDiskDirSizes = ((InternalRegion) region).getDiskDirSizes();

    assertThat(actualDiskDirSizes[0]).isEqualTo(2048 + 500);
    assertThat(actualDiskDirSizes[1]).isEqualTo(2048 + 500);
    assertThat(actualDiskDirSizes[2]).isEqualTo(2048 + 500);
    assertThat(actualDiskDirSizes[3]).isEqualTo(2048 + 500);

    // we have room for 2 values per dir

    byte[] value = new byte[1024];
    Arrays.fill(value, (byte) 77);

    for (int i = 0; i < 8; i++) {
      region.put(String.valueOf(i), value);
    }

    // we should have put 2 values in each dir so the next one should not fit but will be allowed
    // because compaction is enabled. It should log a warning.
    region.put("OK", value);

    assertThat(cache.isClosed()).isFalse();
  }

  /**
   * DiskRegDiskAccessExceptionTest : Disk region test for DiskAccessException.
   */
  @Test
  public void testDiskFullExcepOverflowOnly() {
    int[] diskDirSizes = new int[4];
    diskDirSizes[0] = 2048 + 500;
    diskDirSizes[1] = 2048 + 500;
    diskDirSizes[2] = 2048 + 500;
    diskDirSizes[3] = 2048 + 500;

    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setDiskDirsAndSizes(diskDirs, diskDirSizes);
    diskRegionProperties.setPersistBackup(false);
    diskRegionProperties.setRolling(false);
    diskRegionProperties.setMaxOplogSize(1000000000);
    diskRegionProperties.setOverFlowCapacity(1);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory, 1_000_000_000);

    Region<Object, Object> region =
        createRegion(regionName, diskStoreName, false, true, false, true, 1);

    int[] actualDiskDirSizes = ((InternalRegion) region).getDiskDirSizes();

    assertThat(actualDiskDirSizes[0]).isEqualTo(2048 + 500);
    assertThat(actualDiskDirSizes[1]).isEqualTo(2048 + 500);
    assertThat(actualDiskDirSizes[2]).isEqualTo(2048 + 500);
    assertThat(actualDiskDirSizes[3]).isEqualTo(2048 + 500);

    // we have room for 2 values per dir

    byte[] value = new byte[1024];
    Arrays.fill(value, (byte) 77);

    // put a dummy value in since one value stays in memory
    region.put("FIRST", value);

    for (int i = 0; i < 8; i++) {
      region.put(String.valueOf(i), value);
    }

    // we should have put 2 values in each dir so the next one should not fit
    try (IgnoredException ie = addIgnoredException(DiskAccessException.class)) {
      Throwable thrown = catchThrowable(() -> region.put("FULL", value));
      assertThat(thrown).isInstanceOf(DiskAccessException.class);
    }

    verifyClosedDueToDiskAccessException(region);
  }

  /**
   * Make sure if compaction is enabled that we can exceed the disk dir limit
   */
  @Test
  public void testNoDiskFullExcepOverflowOnly() {
    int[] diskDirSizes = new int[4];
    diskDirSizes[0] = 2048 + 500;
    diskDirSizes[1] = 2048 + 500;
    diskDirSizes[2] = 2048 + 500;
    diskDirSizes[3] = 2048 + 500;

    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setDiskDirsAndSizes(diskDirs, diskDirSizes);
    diskRegionProperties.setPersistBackup(false);
    diskRegionProperties.setRolling(true);
    diskRegionProperties.setMaxOplogSize(1000000000);
    diskRegionProperties.setOverFlowCapacity(1);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory, 1_000_000_000);

    Region<Object, Object> region =
        createRegion(regionName, diskStoreName, false, true, false, true, 1);

    int[] actualDiskDirSizes = ((InternalRegion) region).getDiskDirSizes();

    assertThat(actualDiskDirSizes[0]).isEqualTo(2048 + 500);
    assertThat(actualDiskDirSizes[1]).isEqualTo(2048 + 500);
    assertThat(actualDiskDirSizes[2]).isEqualTo(2048 + 500);
    assertThat(actualDiskDirSizes[3]).isEqualTo(2048 + 500);

    // we have room for 2 values per dir

    byte[] value = new byte[1024];
    Arrays.fill(value, (byte) 77);

    // put a dummy value in since one value stays in memory
    region.put("FIRST", value);

    for (int i = 0; i < 8; i++) {
      region.put(String.valueOf(i), value);
    }

    // we should have put 2 values in each dir so the next one should not fit
    // but will be allowed because compaction is enabled. It should log a warning.
    region.put("OK", value);

    assertThat(cache.isClosed()).isFalse();
  }

  /**
   * DiskAccessException Test : Even if rolling doesn't free the space in stipulated time, the
   * operation should not get stuck or see Exception
   */
  @Test
  public void testSyncModeAllowOperationToProceedEvenIfDiskSpaceIsNotSufficient() {
    int[] diskDirSizes = {2048};

    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setDiskDirsAndSizes(new File[] {diskDirs[0]}, diskDirSizes);
    diskRegionProperties.setPersistBackup(true);
    diskRegionProperties.setRolling(true);
    diskRegionProperties.setCompactionThreshold(100);
    diskRegionProperties.setMaxOplogSize(100000000);
    diskRegionProperties.setRegionName(regionName);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory, 100_000_000);

    Region<Integer, Object> region =
        createRegion(regionName, diskStoreName, true, true, false, false, 0);

    int[] actualDiskSizes = ((InternalRegion) region).getDiskDirSizes();
    assertThat(actualDiskSizes).isEqualTo(diskDirSizes);

    // puts should not fail even after surpassing disk dir sizes
    byte[] value = new byte[990];
    Arrays.fill(value, (byte) 77);
    region.put(1, value);
    region.put(1, value);
    region.put(1, value);

    region.close();
  }// end of testSyncPersistRegionDAExp

  @Test
  public void testAsyncModeAllowOperationToProceedEvenIfDiskSpaceIsNotSufficient() {
    int[] dirSizes = {2048};

    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setDiskDirsAndSizes(new File[] {diskDirs[0]}, dirSizes);
    diskRegionProperties.setPersistBackup(true);
    diskRegionProperties.setRolling(true);
    diskRegionProperties.setCompactionThreshold(100);
    diskRegionProperties.setMaxOplogSize(100000000);
    diskRegionProperties.setBytesThreshold(1000000);
    diskRegionProperties.setTimeInterval(1500000);
    diskRegionProperties.setRegionName(regionName);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    DiskStore diskStore =
        createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory, 100_000_000);

    Region<Object, Object> region =
        createRegion(regionName, diskStoreName, true, false, false, false, 0);

    assertThat(diskStore.getDiskDirSizes()).isEqualTo(dirSizes);

    // puts should not fail even after surpassing disk dir sizes
    byte[] value = new byte[990];
    Arrays.fill(value, (byte) 77);
    region.put(0, value);
    region.put(1, value);
    region.put(2, value);

    region.close();
  }// end of testAsyncPersistRegionDAExp

  /**
   * DiskRegGetInvalidEntryTest: get invalid entry should return null.
   */
  @Test
  public void testDiskGetInvalidEntry() {
    DiskRegionProperties diskRegionProperties = new DiskRegionProperties();
    diskRegionProperties.setDiskDirsAndSizes(diskDirs, diskDirSizes);
    diskRegionProperties.setRolling(false);
    diskRegionProperties.setMaxOplogSize(1000000000);
    diskRegionProperties.setBytesThreshold(1000000);
    diskRegionProperties.setTimeInterval(1500000);

    DiskStoreFactory diskStoreFactory = toDiskStoreFactory(diskRegionProperties);

    createDiskStoreWithSizeInBytes(diskStoreName, diskStoreFactory, 1_000_000_000);

    Region<Object, Object> region =
        createRegion(regionName, diskStoreName, false, false, false, true, 1000);

    byte[] value = new byte[1024];
    Arrays.fill(value, (byte) 77);
    for (int i = 0; i < 10; i++) {
      region.put("key" + i, value);
    }

    // invalidate an entry
