/*
 * 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 org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.IntStream;

import org.apache.commons.io.FileUtils;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;

import org.apache.geode.StatisticsFactory;
import org.apache.geode.cache.CacheWriterException;
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.EntryNotFoundException;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.util.CacheWriterAdapter;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.Oplog.OPLOG_TYPE;
import org.apache.geode.internal.cache.entries.AbstractDiskLRURegionEntry;
import org.apache.geode.internal.cache.entries.DiskEntry;
import org.apache.geode.test.dunit.ThreadUtils;

/**
 * Testing Oplog API's
 */
public class OplogJUnitTest extends DiskRegionTestingBase {

  private boolean proceed = false;

  private final DiskRegionProperties diskProps = new DiskRegionProperties();

  private long delta;

  private volatile boolean assertDone = false;

  private boolean failure = false;

  @Override
  protected final void postSetUp() throws Exception {
    diskProps.setDiskDirs(dirs);
    DiskStoreImpl.SET_IGNORE_PREALLOCATE = true;
  }

  @Override
  protected final void postTearDown() throws Exception {
    DiskStoreImpl.SET_IGNORE_PREALLOCATE = false;
  }

  /**
   * Test method for 'org.apache.geode.internal.cache.Oplog.isBackup()'
   */
  @Test
  public void testIsBackup() {

    InternalRegion overFlowAndPersistRegionRegion =
        (InternalRegion) DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps);
    assertTrue("Not correctly setup for overflow and persist",
        overFlowAndPersistRegionRegion.getDiskRegion().isBackup());
    closeDown(overFlowAndPersistRegionRegion);

    InternalRegion overFlowOnlyRegion =
        (InternalRegion) DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps);
    assertFalse("Not correctly setup for overflow only mode",
        overFlowOnlyRegion.getDiskRegion().isBackup());
    closeDown(overFlowOnlyRegion);

    InternalRegion persistOnlyRegion = (InternalRegion) DiskRegionHelperFactory
        .getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
    assertTrue("Not correctly setup for  persist only mode",
        persistOnlyRegion.getDiskRegion().isBackup());
    closeDown(persistOnlyRegion);
  }

  /*
   * Test method for 'org.apache.geode.internal.cache.Oplog.useSyncWrites()'
   */
  @Test
  public void testUseSyncWritesWhenSet() {
    diskProps.setSynchronous(true);
    InternalRegion syncOverFlowAndPersistRegion =
        (InternalRegion) DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps);
    assertTrue(syncOverFlowAndPersistRegion.getAttributes().isDiskSynchronous());
    closeDown(syncOverFlowAndPersistRegion);

    InternalRegion syncOverFlowOnlyRegion =
        (InternalRegion) DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps);
    assertTrue(syncOverFlowOnlyRegion.getAttributes().isDiskSynchronous());
    closeDown(syncOverFlowOnlyRegion);

    InternalRegion syncPersistOnlyRegion = (InternalRegion) DiskRegionHelperFactory
        .getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
    assertTrue(syncPersistOnlyRegion.getAttributes().isDiskSynchronous());
    closeDown(syncPersistOnlyRegion);
  }

  @Test
  public void testNotUseSyncWritesWhenNotSet() {
    diskProps.setSynchronous(false);
    InternalRegion asyncOverFlowAndPersistRegion =
        (InternalRegion) DiskRegionHelperFactory.getAsyncOverFlowAndPersistRegion(cache, diskProps);
    assertFalse(asyncOverFlowAndPersistRegion.getAttributes().isDiskSynchronous());
    closeDown(asyncOverFlowAndPersistRegion);

    InternalRegion asyncOverFlowOnlyRegion =
        (InternalRegion) DiskRegionHelperFactory.getAsyncOverFlowOnlyRegion(cache, diskProps);
    assertFalse(asyncOverFlowOnlyRegion.getAttributes().isDiskSynchronous());
    closeDown(asyncOverFlowOnlyRegion);

    InternalRegion asyncPersistOnlyRegion =
        (InternalRegion) DiskRegionHelperFactory.getAsyncPersistOnlyRegion(cache, diskProps);
    assertFalse(asyncPersistOnlyRegion.getAttributes().isDiskSynchronous());
    closeDown(asyncPersistOnlyRegion);
  }

  /**
   * Test method for 'org.apache.geode.internal.cache.Oplog.clear(File)'
   */
  @Test
  public void testClear() {
    region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps);
    putTillOverFlow(region);
    region.clear();
    region.close();
    region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps);
    assertTrue(" failed in get OverflowAndPersist ", region.get(0) == null);
    closeDown();

    region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps);
    putTillOverFlow(region);
    region.clear();
    region.close();
    region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps);
    assertTrue(" failed in get OverflowOnly ", region.get(0) == null);
    closeDown();

    region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
    put100Int();
    region.clear();
    region.close();
    region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
    assertTrue(" failed in get PersistOnly ", region.get(0) == null);
    closeDown();
  }

  /**
   * Test method for 'org.apache.geode.internal.cache.Oplog.close()'
   */
  @Test
  public void testClose() {
    {
      deleteFiles();
      region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps);
      DiskRegion dr = ((LocalRegion) region).getDiskRegion();
      Oplog oplog = dr.testHook_getChild();
      long id = oplog.getOplogId();
      oplog.close();

      StatisticsFactory factory = cache.getDistributedSystem();
      Oplog newOplog =
          new Oplog(id, dr.getOplogSet(), new DirectoryHolder(factory, dirs[0], 1000, 0));
      dr.getOplogSet().setChild(newOplog);
      closeDown();
    }
    {
      deleteFiles();
      region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps);
      DiskRegion dr = ((LocalRegion) region).getDiskRegion();
      dr.testHookCloseAllOverflowOplogs();
      checkIfContainsFile("OVERFLOW");
      closeDown();
    }
    {
      deleteFiles();
      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      DiskRegion dr = ((LocalRegion) region).getDiskRegion();
      Oplog oplog = dr.testHook_getChild();
      long id = oplog.getOplogId();
      oplog.close();
      StatisticsFactory factory = cache.getDistributedSystem();
      Oplog newOplog =
          new Oplog(id, dr.getOplogSet(), new DirectoryHolder(factory, dirs[0], 1000, 2));
      dr.setChild(newOplog);
      closeDown();
    }

  }

  private void closeDown(InternalRegion region) {
    super.closeDown(region);
    DiskRegion diskRegion = region != null ? region.getDiskRegion() : null;
    if (diskRegion != null) {
      diskRegion.getDiskStore().close();
      ((InternalCache) cache).removeDiskStore(diskRegion.getDiskStore());
    }
  }

  @Override
  protected void closeDown() {
    DiskRegion dr = null;
    if (region != null) {
      dr = ((LocalRegion) region).getDiskRegion();
    }
    super.closeDown();
    if (dr != null) {
      dr.getDiskStore().close();
      ((LocalRegion) region).getGemFireCache().removeDiskStore(dr.getDiskStore());
    }
  }

  private void checkIfContainsFile(String fileExtension) {
    for (File dir : dirs) {
      File[] files = dir.listFiles();
      for (File file : files) {
        if (file.getAbsolutePath().endsWith(fileExtension)) {
          fail("file " + file + " still exists after oplog.close()");
        }
      }
    }
  }

  /**
   * Test method for 'org.apache.geode.internal.cache.Oplog.destroy()'
   */
  @Test
  public void testDestroy() {
    region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps);
    put100Int();
    putTillOverFlow(region);
    try {
      region.destroy(0);
    } catch (EntryNotFoundException e1) {
      logWriter.error("Exception occurred", e1);
      fail(" Entry not found when it was expected to be there");
    }
    region.close();
    region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps);
    assertTrue(" failed in get OverflowAndPersist ", region.get(0) == null);
    closeDown();

    region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps);
    put100Int();
    putTillOverFlow(region);
    try {
      region.destroy(0);
    } catch (EntryNotFoundException e1) {
      logWriter.error("Exception occurred", e1);
      fail(" Entry not found when it was expected to be there");
    }
    region.close();
    region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps);
    assertTrue(" failed in get OverflowOnly ", region.get(0) == null);

    closeDown();

    region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
    put100Int();
    try {
      region.destroy(0);
    } catch (EntryNotFoundException e1) {
      logWriter.error("Exception occurred", e1);
      fail(" Entry not found when it was expected to be there");
    }
    region.close();
    region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
    assertTrue(" failed in get PersistOnly ", region.get(0) == null);
    closeDown();

  }

  /**
   * Test method for 'org.apache.geode.internal.cache.Oplog.remove(long)'
   */
  @Test
  public void testRemove() {
    region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps);
    putTillOverFlow(region);
    region.remove(0);
    region.close();
    region = DiskRegionHelperFactory.getSyncOverFlowAndPersistRegion(cache, diskProps);
    assertTrue(" failed in get OverflowAndPersist ", region.get(0) == null);
    closeDown();

    region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps);
    putTillOverFlow(region);
    region.remove(0);
    assertTrue(" failed in get OverflowOnly ", region.get(0) == null);
    region.close();
    region = DiskRegionHelperFactory.getSyncOverFlowOnlyRegion(cache, diskProps);
    closeDown();

    region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
    put100Int();
    region.remove(0);
    assertTrue(" failed in get PersistOnly ", region.get(0) == null);
    region.close();
    region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
    closeDown();

  }

  /**
   * Tests whether the data is written in the right format on the disk
   *
   */
  @Test
  public void testFaultInOfValuesFromDisk() {
    try {
      // Asif First create a persist only disk region which is of aysnch
      // & switch of OplOg type
      diskProps.setMaxOplogSize(1000);

      diskProps.setPersistBackup(true);
      diskProps.setRolling(false);
      diskProps.setSynchronous(true);
      diskProps.setTimeInterval(-1);
      diskProps.setOverflow(false);

      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      byte[] val = new byte[10];
      for (int i = 0; i < val.length; ++i) {
        val[i] = (byte) i;
      }
      region.put(1, val);

      DiskEntry entry = ((DiskEntry) ((LocalRegion) region).basicGetEntry(1));
      DiskRegion dr = ((LocalRegion) region).getDiskRegion();

      val = (byte[]) dr.getNoBuffer(entry.getDiskId());
      for (int i = 0; i < val.length; ++i) {
        if (val[i] != (byte) i) {
          fail("Test for fault in from disk failed");
        }
      }
      val = (byte[]) DiskStoreImpl.convertBytesAndBitsIntoObject(
          dr.getBytesAndBitsWithoutLock(entry.getDiskId(), true, false), (InternalCache) cache);
      for (int i = 0; i < val.length; ++i) {
        if (val[i] != (byte) i) {
          fail("Test for fault in from disk failed");
        }
      }
      region.invalidate(1);
      assertTrue(dr.getNoBuffer(entry.getDiskId()) == Token.INVALID);

    } catch (Exception e) {
      logWriter.error("Exception occurred", e);
      fail(e.toString());
    }
    closeDown();
  }

  /**
   * Tests the original ByteBufferPool gets transferred to the new Oplog for synch mode
   *
   */
  @Test
  public void testByteBufferPoolTransferForSynchMode() {
    diskProps.setMaxOplogSize(1024);
    diskProps.setBytesThreshold(0);
    diskProps.setPersistBackup(true);
    diskProps.setRolling(false);
    diskProps.setSynchronous(true);
    diskProps.setTimeInterval(10000);
    diskProps.setOverflow(false);

    region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
    DiskRegion dr = ((LocalRegion) region).getDiskRegion();
    // Populate data just below the switch over threshold
    byte[] val = new byte[5];
    for (int i = 0; i < val.length; ++i) {
      val[i] = (byte) i;
    }

    region.put(1, val);

    ((LocalRegion) region).basicGetEntry(1);
    Oplog old = dr.testHook_getChild();
    ByteBuffer oldWriteBuf = old.getWriteBuf();
    dr.forceRolling();
    region.put(2, val);
    Oplog switched = dr.testHook_getChild();
    assertTrue(old != switched);
    assertEquals(dr.getDiskStore().getPersistentOplogs().getChild(2), switched);
    assertEquals(oldWriteBuf, switched.getWriteBuf());
    assertEquals(null, old.getWriteBuf());
    closeDown();

  }

  /**
   * Tests the bug which arises in case of asynch mode during oplog switching caused by conflation
   * of create/destroy operation.The bug occurs if a create operation is followed by destroy but
   * before destroy proceeds some other operation causes oplog switching
   *
   */
  @Test
  public void testBug34615() {
    final int MAX_OPLOG_SIZE = 100;
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE);
    diskProps.setPersistBackup(true);
    diskProps.setRolling(false);
    diskProps.setSynchronous(false);
    diskProps.setOverflow(false);
    diskProps.setBytesThreshold(150);

    LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true;
    final byte[] val = new byte[50];
    region = DiskRegionHelperFactory.getAsyncPersistOnlyRegion(cache, diskProps);
    final CacheObserver old = CacheObserverHolder.setInstance(new CacheObserverAdapter() {
      @Override
      public void afterConflation(ByteBuffer orig, ByteBuffer conflated) {
        Thread th = new Thread(() -> region.put("2", new byte[75]));
        assertNull(conflated);
        th.start();
        ThreadUtils.join(th, 30 * 1000);
        LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false;

      }
    });

    region.put("1", val);
    region.remove("1");
    assertFalse(failureCause, testFailed);
    CacheObserverHolder.setInstance(old);
    closeDown();

  }

  @Test
  public void testConflation() throws Exception {

    final int MAX_OPLOG_SIZE = 1000;
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE);
    diskProps.setPersistBackup(true);
    diskProps.setRolling(false);
    diskProps.setSynchronous(false);
    diskProps.setOverflow(false);
    diskProps.setBytesThreshold(1500);

    final byte[] val = new byte[50];
    final byte[][] bb = new byte[2][];
    bb[0] = new byte[5];
    region = DiskRegionHelperFactory.getAsyncPersistOnlyRegion(cache, diskProps);
    try {
      region.put("1", val);
      region.put("1", new byte[10]);
      region.put("2", val);
      region.put("2", new byte[100]);
      region.create("3", null);
      region.put("3", new byte[10]);
      region.create("4", null);
      region.put("4", new byte[0]);

      // tests for byte[][]
      region.create("5", bb);
      region.put("6", val);
      region.put("6", bb);
      region.create("7", null);
      region.put("7", bb);

      region.create("8", new byte[9]);
      region.invalidate("8");
      region.create("9", new byte[0]);
      region.invalidate("9");

      region.create("10", new byte[9]);
      region.localInvalidate("10");
      region.create("11", new byte[0]);
      region.localInvalidate("11");

      DiskRegion dr = ((LocalRegion) region).getDiskRegion();
      dr.flushForTesting();
      byte[] val_1 = ((byte[]) ((LocalRegion) region).getValueOnDisk("1"));
      assertEquals(val_1.length, 10);
      byte[] val_2 = ((byte[]) ((LocalRegion) region).getValueOnDisk("2"));
      assertEquals(val_2.length, 100);
      byte[] val_3 = ((byte[]) ((LocalRegion) region).getValueOnDisk("3"));
      assertEquals(val_3.length, 10);
      byte[] val_4 = ((byte[]) ((LocalRegion) region).getValueOnDisk("4"));
      assertEquals(val_4.length, 0);
      byte[][] val_5 = (byte[][]) ((LocalRegion) region).getValueOnDisk("5");
      assertEquals(val_5.length, 2);
      assertEquals(val_5[0].length, 5);
      assertNull(val_5[1]);
      byte[][] val_6 = (byte[][]) ((LocalRegion) region).getValueOnDisk("6");
      assertEquals(val_6.length, 2);
      assertEquals(val_6[0].length, 5);
      assertNull(val_6[1]);
      byte[][] val_7 = (byte[][]) ((LocalRegion) region).getValueOnDisk("7");
      assertEquals(val_7.length, 2);
      assertEquals(val_7[0].length, 5);
      assertNull(val_7[1]);
      Object val_8 = ((LocalRegion) region).getValueOnDisk("8");
      assertEquals(val_8, Token.INVALID);
      Object val_9 = ((LocalRegion) region).getValueOnDisk("9");
      assertEquals(val_9, Token.INVALID);
      Object val_10 = ((LocalRegion) region).getValueOnDisk("10");
      assertEquals(val_10, Token.LOCAL_INVALID);
      Object val_11 = ((LocalRegion) region).getValueOnDisk("11");
      assertEquals(val_11, Token.LOCAL_INVALID);

    } finally {
      closeDown();
    }
  }

  /**
   * This tests the retrieval of empty byte array when present in asynch buffers
   *
   */
  @Test
  public void testGetEmptyByteArrayInAsynchBuffer() {
    final int MAX_OPLOG_SIZE = 1000;
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE);
    diskProps.setPersistBackup(true);
    diskProps.setRolling(false);
    diskProps.setSynchronous(false);
    diskProps.setOverflow(false);
    diskProps.setBytesThreshold(1500);

    final byte[] val = new byte[50];
    region = DiskRegionHelperFactory.getAsyncPersistOnlyRegion(cache, diskProps);
    try {
      region.put("1", val);
      region.put("1", new byte[0]);
      byte[] val_1 = ((byte[]) ((LocalRegion) region).getValueOnDiskOrBuffer("1"));
      assertEquals(val_1.length, 0);
    } catch (Exception e) {
      logWriter.error("Exception occurred", e);
      fail("The test failed due to exception = " + e);
    }
    closeDown();
  }

  /**
   * This tests the retrieval of empty byte array in synch mode
   *
   */
  @Test
  public void testGetEmptyByteArrayInSynchMode() {
    final int MAX_OPLOG_SIZE = 1000;
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE);
    diskProps.setPersistBackup(true);
    diskProps.setRolling(false);
    diskProps.setSynchronous(true);
    diskProps.setOverflow(false);
    diskProps.setBytesThreshold(1500);

    final byte[] val = new byte[50];
    region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
    try {
      region.put("1", val);
      region.put("1", new byte[0]);
      byte[] val_1 = ((byte[]) ((LocalRegion) region).getValueOnDiskOrBuffer("1"));
      assertEquals(val_1.length, 0);
    } catch (Exception e) {
      logWriter.error("Exception occurred", e);
      fail("The test failed due to exception = " + e);
    }
    closeDown();
  }

  /**
   * This tests the bug which caused the oplogRoller to attempt to roll a removed entry whose value
   * is Token.Removed This bug can occur if a remove operation causes oplog switching & hence roller
   * thread gets notified, & the roller thread obtains the iterator of the concurrent region map
   * before the remove
   *
   */
  @Test
  public void testBug34702() {
    final int MAX_OPLOG_SIZE = 500 * 2;
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE);
    diskProps.setPersistBackup(true);
    diskProps.setRolling(true);
    diskProps.setSynchronous(true);
    diskProps.setOverflow(false);
    final byte[] val = new byte[200];
    proceed = false;

    region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
    region.put("key1", val);
    region.put("key2", val);
    LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true;
    final CacheObserver old = CacheObserverHolder.setInstance(new CacheObserverAdapter() {

      @Override
      public void afterSettingOplogOffSet(long offset) {
        ((LocalRegion) region).getDiskRegion().forceRolling();
        // Let the operation thread yield to the Roller so that
        // it is able to obtain the iterator of the concurrrent region map
        // & thus get the reference to the entry which will contain
        // value as Token.Removed as the entry though removed from
        // concurrent
        // map still will be available to the roller
        Thread.yield();
        // Sleep for some time
        try {
          Thread.sleep(5000);
        } catch (InterruptedException e) {
          testFailed = true;
          failureCause = "No guarantee that test is succesful";
          fail("No guarantee that test is succesful");
        }
      }

      @Override
      public void afterHavingCompacted() {
        proceed = true;
        synchronized (OplogJUnitTest.this) {
          OplogJUnitTest.this.notify();
        }
      }
    });
    try {
      region.destroy("key1");
      region.destroy("key2");
    } catch (Exception e1) {
      LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false;
      CacheObserverHolder.setInstance(old);
      fail("Test failed as entry deletion threw exception. Exception = " + e1);
    }
    // Wait for some time & check if the after having rolled callabck
    // is issued sucessfully or not.
    if (!proceed) {
      synchronized (this) {
        if (!proceed) {
          try {
            wait(20000);
          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            // The test will automatically fail due to proceed flag
          }
        }
      }
    }
    assertFalse(failureCause, testFailed);
    LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false;
    CacheObserverHolder.setInstance(old);

    if (!proceed) {
      fail("Test failed as afterHavingCompacted callabck not issued even after sufficient wait");
    }
    closeDown();

  }

  /**
   * tests a potential deadlock situation if the operation causing a swithcing of Oplog is waiting
   * for roller to free space. The problem can arise if the operation causing Oplog switching is
   * going on an Entry , which already has its oplog ID referring to the Oplog being switched. In
   * such case, when the roller will try to roll the entries referencing the current oplog , it will
   * not be able to acquire the lock on the entry as the switching thread has already taken a lock
   * on it.
   *
   */
  @Test
  public void testRollingDeadlockSituation() {
    final int MAX_OPLOG_SIZE = 2000;
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE);
    diskProps.setPersistBackup(true);
    diskProps.setRolling(true);
    diskProps.setSynchronous(true);
    diskProps.setOverflow(false);
    diskProps.setDiskDirsAndSizes(new File[] {dirs[0]}, new int[] {1400});
    final byte[] val = new byte[500];
    proceed = false;
    region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
    region.put("key1", val);
    region.put("key1", val);
    try {
      region.put("key1", val);
    } catch (DiskAccessException dae) {
      logWriter.error("Exception occurred", dae);
      fail(
          "Test failed as DiskAccessException was encountered where as the operation should ideally have proceeded without issue . exception = "
              + dae);
    }
  }

  /**
   * This tests whether an empty byte array is correctly writtem to the disk as a zero value length
   * operation & hence the 4 bytes field for recording the value length is absent & also since the
   * value length is zero no byte for it should also get added. Similary during recover from HTree
   * as well as Oplog , the empty byte array should be read correctly
   *
   */
  @Test
  public void testEmptyByteArrayPutAndRecovery() {
    CacheObserver old = CacheObserverHolder.setInstance(new CacheObserverAdapter() {
      @Override
      public void afterConflation(ByteBuffer origBB, ByteBuffer conflatedBB) {
        if ((2 + 4 + 1 + EntryEventImpl.serialize("key1").length) != origBB.capacity()) {
          failureCause =
              "For a backup region, addition of an empty array should result in an offset of 6 bytes where as actual offset is ="
                  + origBB.capacity();
          testFailed = true;
        }
        Assert.assertTrue(
            "For a backup region, addition of an empty array should result in an offset of 6 bytes where as actual offset is ="
                + origBB.capacity(),
            (2 + 4 + 1 + EntryEventImpl.serialize("key1").length) == origBB.capacity());

      }
    });
    try {
      final int MAX_OPLOG_SIZE = 2000;
      diskProps.setMaxOplogSize(MAX_OPLOG_SIZE);
      diskProps.setPersistBackup(true);
      diskProps.setSynchronous(true);
      diskProps.setOverflow(false);
      diskProps.setDiskDirsAndSizes(new File[] {dirs[0]}, new int[] {1400});
      final byte[] val = new byte[0];
      LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true;

      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      region.put("key1", val);
      LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false;
      region.close();
      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      byte[] _val = (byte[]) region.get("key1");
      assertTrue(
          "value of key1 after restarting the region is not an empty byte array. This may indicate problem in reading from Oplog",
          _val.length == 0);
      if (logWriter.infoEnabled()) {
        logWriter.info(
            "After first region close & opening again no problems encountered & hence Oplog has been read successfully.");
        logWriter.info(
            "Closing the region again without any operation done, would indicate that next time data will be loaded from HTree .");
      }
      region.close();
      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      _val = (byte[]) region.get("key1");
      assertTrue(
          "value of key1 after restarting the region is not an empty byte array. This may indicate problem in reading from HTRee",
          _val.length == 0);
      assertFalse(failureCause, testFailed);
      // region.close();

    } finally {
      LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false;
      CacheObserverHolder.setInstance(old);

    }
  }

  /**
   * This is used to test bug 35012 where a remove operation on a key gets unrecorded due to
   * switching of Oplog if it happens just after the remove operation has destroyed the in memory
   * entry & is about to acquire the readlock in DiskRegion to record the same. If the Oplog has
   * switched during that duration , the bug would appear
   *
   */

  @Test
  public void testBug35012() {
    final int MAX_OPLOG_SIZE = 500;
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE);
    diskProps.setPersistBackup(true);
    diskProps.setRolling(false);
    diskProps.setSynchronous(true);
    diskProps.setOverflow(false);
    final byte[] val = new byte[200];
    try {

      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);

      region.put("key1", val);
      region.put("key2", val);
      region.put("key3", val);
      final Thread th = new Thread(() -> region.remove("key1"));
      // main thread acquires the write lock
      ((LocalRegion) region).getDiskRegion().acquireWriteLock();
      try {
        th.start();
        Thread.yield();
        DiskRegion dr = ((LocalRegion) region).getDiskRegion();
        dr.testHook_getChild().forceRolling(dr);
      } finally {
        ((LocalRegion) region).getDiskRegion().releaseWriteLock();
      }
      ThreadUtils.join(th, 30 * 1000);
      region.close();
      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      assertEquals(region.size(), 2);
    } catch (Exception e) {
      logWriter.error("Exception occurred ", e);
      fail("The test could not be completed because of exception .Exception=" + e);
    }
    closeDown();

  }

  /**
   * Tests if buffer size & time are not set , the asynch writer gets awakened on time basis of
   * default 1 second
   *
   */
  @Test
  public void testAsynchWriterAttribBehaviour1() throws Exception {
    DiskStoreFactory dsf = cache.createDiskStoreFactory();
    ((DiskStoreFactoryImpl) dsf).setMaxOplogSizeInBytes(10000);
    File dir = new File("testingDirectoryDefault");
    dir.mkdir();
    dir.deleteOnExit();
    File[] dirs = {dir};
    dsf.setDiskDirs(dirs);
    RegionFactory<Object, Object> factory =
        cache.createRegionFactory(RegionShortcut.REPLICATE_PERSISTENT);
    final long t1 = System.currentTimeMillis();
    DiskStore ds = dsf.create("test");
    factory.setDiskSynchronous(false);
    factory.setDiskStoreName(ds.getName());
    factory.setScope(Scope.LOCAL);
    region = factory.create("test");

    LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true;
    CacheObserver old = CacheObserverHolder.setInstance(

        new CacheObserverAdapter() {
          private long t2;

          @Override
          public void goingToFlush() {
            t2 = System.currentTimeMillis();
            delta = t2 - t1;
            LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false;
            synchronized (OplogJUnitTest.this) {
              OplogJUnitTest.this.notify();
            }
          }
        });

    region.put("key1", "111111111111");
    synchronized (this) {
      if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) {
        wait(10000);
        assertFalse(LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER);
      }
    }
    CacheObserverHolder.setInstance(old);
    // Windows clock has an accuracy of 15 ms. Accounting for the same.
    assertTrue("delta is in miilliseconds=" + delta, delta >= 985);
    closeDown();
  }

  /**
   * Tests if buffer size is set but time is not set , the asynch writer gets awakened on buffer
   * size basis
   */
  @Ignore("TODO:DARREL_DISABLE: test is disabled")
  @Test
  public void testAsynchWriterAttribBehaviour2() throws Exception {
    DiskStoreFactory dsf = cache.createDiskStoreFactory();
    ((DiskStoreFactoryImpl) dsf).setMaxOplogSizeInBytes(10000);
    dsf.setQueueSize(2);
    File dir = new File("testingDirectoryDefault");
    dir.mkdir();
    dir.deleteOnExit();
    File[] dirs = {dir};
    dsf.setDiskDirs(dirs);
    RegionFactory<Object, Object> factory =
        cache.createRegionFactory(RegionShortcut.REPLICATE_PERSISTENT);
    DiskStore ds = dsf.create("test");
