gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/DiskRandomOperationsAndRecoveryJUnitTest.java

/*=========================================================================
 * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
 * This product is protected by U.S. and international copyright
 * and intellectual property laws. Pivotal products are covered by
 * one or more patents listed at http://www.pivotal.io/patents.
 *=========================================================================
 */
package com.gemstone.gemfire.internal.cache;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Random;

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

import static org.junit.Assert.*;

import com.gemstone.gemfire.cache.CacheFactory;
import com.gemstone.gemfire.cache.EntryDestroyedException;
import com.gemstone.gemfire.cache.EntryNotFoundException;
import com.gemstone.gemfire.cache.Region;
import com.gemstone.gemfire.cache.Scope;
import com.gemstone.gemfire.internal.cache.LocalRegion;
import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.test.junit.categories.IntegrationTest;
/**
 * 
 * @author Asif
 *
 */
@Category(IntegrationTest.class)
public class DiskRandomOperationsAndRecoveryJUnitTest extends DiskRegionTestingBase
{
  DiskRegionProperties diskProps = new DiskRegionProperties();
  private static int ENTRY_SIZE = 1024;
  final static byte[] valueBytes = new byte[ENTRY_SIZE];
  static {
    Arrays.fill(valueBytes, (byte)32);
  }
  final static private Object value = new String(valueBytes);

  // private static final boolean debug = false;

  private static int testId=0;
  
  @Before
  public void setUp() throws Exception
  {
    super.setUp();
    diskProps.setDiskDirs(dirs);
    testId++;
  }

  @After
  public void tearDown() throws Exception
  {
    super.tearDown();
  }

  private final static int ITERATIONS = 4;
  private final static long MAX_OPLOG_SIZE_IN_BYTES = 1024*16;
  /**
   * Need to limit he max open oplogs so that we don't run out of file descriptors
   */
  private final static int MAX_OPEN_OPLOGS = 400;
  private final static long RECORDS_PER_OPLOG = MAX_OPLOG_SIZE_IN_BYTES / (ENTRY_SIZE + 24/*for key and record overhead*/);
  /**
   * Maximum number of ops a test can do w/o running out of file descriptors
   * due to open oplogs
   */
  private final static long OPS_PER_TEST = RECORDS_PER_OPLOG * MAX_OPEN_OPLOGS;
  private final static long OPS_PER_ITERATION = OPS_PER_TEST / ITERATIONS;

  @Test
  public void testRollingDisabledRecoverValuesFalsePersistOnly() throws Exception { 
    String oldValue = System.getProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
    System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, "false");
    try {
    diskProps.setPersistBackup(true);
    diskProps.setRegionName("testRollingDisabledRecoverValuesFalsePersistOnly");
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE_IN_BYTES);
    diskProps.setRolling(false);
    int previousRegionSize = 0;
    HashMap<String, VersionTag> tagmapInCache = null;
    //Num time start / close cycles
    for(int i =0; i < ITERATIONS; ++i) {
      long t1 = System.currentTimeMillis();
      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      long t2 = System.currentTimeMillis();
      System.out.println("Time taken to recover = "+ (t2-t1) + " for total number of entries= "+ region.size() + " with recover values as false  and rolling disabled ");
      if ((t2-t1) > 0) {
        System.out.println("Recovery rate is= "+ region.size()/ (t2-t1) + " per milliseconds");
      }
      int startKey =this.processRegionData();
      assertEquals(previousRegionSize,region.size());
      HashMap<String, VersionTag> tagmapFromRecover = saveVersionTags((LocalRegion)region);
      if (tagmapInCache!=null) {
        compareVersionTags(tagmapInCache, tagmapFromRecover);
      }
      int thisRegionSize = startOperations(startKey, value);
      previousRegionSize = thisRegionSize;
      tagmapInCache = saveVersionTags((LocalRegion)region);
      region.close();      
    }    
    } finally {
      if (oldValue != null) {
        System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, oldValue);
      } else {
        System.clearProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
      }
    }
  }
  
  @Test
  public void testRollingDisabledRecoverValuesTruePersistOnly() throws Exception {
    String oldValue = System.getProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
    System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, "true");
    try {
    diskProps.setPersistBackup(true);
    diskProps.setRegionName("testRollingDisabledRecoverValuesTruePersistOnly");
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE_IN_BYTES);
    diskProps.setRolling(false);   
    int previousRegionSize = 0;
    HashMap<String, VersionTag> tagmapInCache = null;
    //Num time start / close cycles
    for(int i =0; i < ITERATIONS; ++i) {
      long t1 = System.currentTimeMillis();
      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      long t2 = System.currentTimeMillis();
      System.out.println("Time taken to recover = "+ (t2-t1) + " for total number of entries= "+ region.size() + " with recover values as false  and rolling disabled ");
      if ((t2-t1) > 0) {
        System.out.println("Recovery rate is= "+ region.size()/ (t2-t1) + " per milliseconds");
      }
      int startKey =this.processRegionData();
      assertEquals(previousRegionSize,region.size());
      HashMap<String, VersionTag> tagmapFromRecover = saveVersionTags((LocalRegion)region);
      if (tagmapInCache!=null) {
        compareVersionTags(tagmapInCache, tagmapFromRecover);
      }
      int thisRegionSize = startOperations(startKey, value);
      previousRegionSize = thisRegionSize;
      tagmapInCache = saveVersionTags((LocalRegion)region);
      region.close();      
    }    
    } finally {
      if (oldValue != null) {
        System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, oldValue);
      } else {
        System.clearProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
      }
    }
  }
  
  //////////////////////////////////////////////////////////////Set 1 Begin //////////////////
  @Test
  public void testRollingEnabledRecoverValuesFalsePersistOnlyWithEarlyTerminationOfRoller() throws Exception {
    System.setProperty(DiskStoreImpl.COMPLETE_COMPACTION_BEFORE_TERMINATION_PROPERTY_NAME, "false");
//     System.setProperty(DiskRegion.CHECK_ENTRY_BALANCE_PROPERTY_NAME, "true");
    try {
      this.rollingEnabledRecoverValuesFalsePersistOnly();
    }finally {
//       System.setProperty(DiskRegion.CHECK_ENTRY_BALANCE_PROPERTY_NAME, "false");
    }
  }
  
  @Test
  public void testRollingEnabledRecoverValuesFalsePersistOnlyWithRollerTerminationComplete() throws Exception {
    try {
      System.setProperty(DiskStoreImpl.COMPLETE_COMPACTION_BEFORE_TERMINATION_PROPERTY_NAME, "true");
      //System.setProperty(DiskRegion.ASSERT_ON_RECOVERY_PROPERTY_NAME, "true");
      this.rollingEnabledRecoverValuesFalsePersistOnly();
    }finally {
      System.setProperty(DiskStoreImpl.COMPLETE_COMPACTION_BEFORE_TERMINATION_PROPERTY_NAME, "false");
      //System.setProperty(DiskRegion.ASSERT_ON_RECOVERY_PROPERTY_NAME, "false");
    }
  }
  
  private void rollingEnabledRecoverValuesFalsePersistOnly() throws Exception { 
    String oldValue = System.getProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
    System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, "false");
    try {
    diskProps.setPersistBackup(true);
    diskProps.setRegionName("testRollingEnabledRecoverValuesFalsePersistOnly");
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE_IN_BYTES);
    diskProps.setRolling(true);   
    int previousRegionSize = 0;
    HashMap<String, VersionTag> tagmapInCache = null;
    //Num time start / close cycles
    for(int i =0; i < ITERATIONS; ++i) {
      long t1 = System.currentTimeMillis();
      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      long t2 = System.currentTimeMillis();
      System.out.println("Time taken to recover = "+ (t2-t1) + " for total number of entries= "+ region.size() + " with recover values as false and rolling disabled ");
      if ((t2-t1) > 0) {
        System.out.println("Recovery rate is= "+ region.size()/ (t2-t1) + " per milliseconds");
      }
      int startKey =this.processRegionData();
      assertEquals(previousRegionSize,region.size());
      HashMap<String, VersionTag> tagmapFromRecover = saveVersionTags((LocalRegion)region);
      if (tagmapInCache!=null) {
        compareVersionTags(tagmapInCache, tagmapFromRecover);
      }
      int thisRegionSize = startOperations(startKey, value);
      previousRegionSize = thisRegionSize;
      tagmapInCache = saveVersionTags((LocalRegion)region);
      region.close();      
    }   
    } finally {
      if (oldValue != null) {
        System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, oldValue);
      } else {
        System.clearProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
      }
    }
  }
  
  //////////////////////////////////////////////////////////////Set 1 END //////////////////
 

  //////////////////////////////////////////////////////////////Set 2 Begin //////////////////
  @Test
  public void testRollingEnabledRecoverValuesTruePersistOnlyWithEarlyTerminationOfRoller() throws Exception {
    System.setProperty(DiskStoreImpl.COMPLETE_COMPACTION_BEFORE_TERMINATION_PROPERTY_NAME, "false");
//     System.setProperty(DiskRegion.CHECK_ENTRY_BALANCE_PROPERTY_NAME, "true");
    try {
      this.rollingEnabledRecoverValuesTruePersistOnly();          
    }finally {
//       System.setProperty(DiskRegion.CHECK_ENTRY_BALANCE_PROPERTY_NAME, "false");
    }
  }
  
  @Test
  public void testRollingEnabledRecoverValuesTruePersistOnlyWithRollerTerminationComplete() throws Exception {
    try {
      System.setProperty(DiskStoreImpl.COMPLETE_COMPACTION_BEFORE_TERMINATION_PROPERTY_NAME, "true");
      //System.setProperty(DiskRegion.ASSERT_ON_RECOVERY_PROPERTY_NAME, "true");
      this.rollingEnabledRecoverValuesTruePersistOnly();
    }finally {
      System.setProperty(DiskStoreImpl.COMPLETE_COMPACTION_BEFORE_TERMINATION_PROPERTY_NAME, "false");
      //System.setProperty(DiskRegion.ASSERT_ON_RECOVERY_PROPERTY_NAME, "false");
    }
  }
  
  private void rollingEnabledRecoverValuesTruePersistOnly() throws Exception { 
    String oldValue = System.getProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
    System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, "true");
    try {
    diskProps.setPersistBackup(true);
    diskProps.setRegionName("testRollingEnabledRecoverValuesTruePersistOnly");
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE_IN_BYTES);
    diskProps.setRolling(true);   
    int previousRegionSize = 0;
    HashMap<String, VersionTag> tagmapInCache = null;
    //Num time start / close cycles
    for(int i =0; i < ITERATIONS; ++i) {
      long t1 = System.currentTimeMillis();
      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      long t2 = System.currentTimeMillis();
      System.out.println("Time taken to recover = "+ (t2-t1) + " for total number of entries= "+ region.size() + " with recover values as false  and rolling disabled ");
      if ((t2-t1) > 0) {
        System.out.println("Recovery rate is= "+ region.size()/ (t2-t1) + " per milliseconds");
      }
      int startKey =this.processRegionData();
      assertEquals(previousRegionSize,region.size());
      HashMap<String, VersionTag> tagmapFromRecover = saveVersionTags((LocalRegion)region);
      if (tagmapInCache!=null) {
        compareVersionTags(tagmapInCache, tagmapFromRecover);
      }
      int thisRegionSize = startOperations(startKey, value);
      previousRegionSize = thisRegionSize;
      tagmapInCache = saveVersionTags((LocalRegion)region);
      region.close();      
    }   
    } finally {
      if (oldValue != null) {
        System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, oldValue);
      } else {
        System.clearProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
      }
    }
  }
  
  //////////////////////////////////////////////////////////////Set 2 end //////////////////
  
  @Test
  public void testRollingDisabledRecoverValuesFalsePersistWithOverFlow() throws Exception { 
    String oldValue = System.getProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
    System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, "false");
    try {
    diskProps.setPersistBackup(true);
    diskProps.setRegionName("testRollingDisabledRecoverValuesFalsePersistWithOverFlow");
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE_IN_BYTES);
    diskProps.setRolling(false);
    diskProps.setOverFlowCapacity(100);
    int previousRegionSize = 0;
    HashMap<String, VersionTag> tagmapInCache = null;
    //Num time start / close cycles
    for(int i =0; i < ITERATIONS; ++i) {
      long t1 = System.currentTimeMillis();
      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      long t2 = System.currentTimeMillis();
      System.out.println("Time taken to recover = "+ (t2-t1) + " for total number of entries= "+ region.size() + " with recover values as false  and rolling disabled ");
      if ((t2-t1) > 0) {
        System.out.println("Recovery rate is= "+ region.size()/ (t2-t1) + " per milliseconds");
      }
      int startKey =this.processRegionData();
      assertEquals(previousRegionSize,region.size());
      HashMap<String, VersionTag> tagmapFromRecover = saveVersionTags((LocalRegion)region);
      if (tagmapInCache!=null) {
        compareVersionTags(tagmapInCache, tagmapFromRecover);
      }
      int thisRegionSize = startOperations(startKey, value);
      previousRegionSize = thisRegionSize;
      tagmapInCache = saveVersionTags((LocalRegion)region);
      region.close();      
    }    
    } finally {
      if (oldValue != null) {
        System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, oldValue);
      } else {
        System.clearProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
      }
    }
  }
  
  @Test
  public void testRollingDisabledRecoverValuesTruePersistWithOverFlow() throws Exception { 
    String oldValue = System.getProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
    System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, "true");
    try {
    diskProps.setPersistBackup(true);
    diskProps.setRegionName("testRollingDisabledRecoverValuesTruePersistWithOverFlow");
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE_IN_BYTES);
    diskProps.setRolling(false);
    diskProps.setOverFlowCapacity(100);
    int previousRegionSize = 0;
    HashMap<String, VersionTag> tagmapInCache = null;
    //Num time start / close cycles
    for(int i =0; i < ITERATIONS; ++i) {
      long t1 = System.currentTimeMillis();
      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      long t2 = System.currentTimeMillis();
      System.out.println("Time taken to recover = "+ (t2-t1) + " for total number of entries= "+ region.size() + " with recover values as false  and rolling disabled ");
      if ((t2-t1) > 0) {
        System.out.println("Recovery rate is= "+ region.size()/ (t2-t1) + " per milliseconds");
      }
      int startKey =this.processRegionData();
      assertEquals(previousRegionSize,region.size());
      HashMap<String, VersionTag> tagmapFromRecover = saveVersionTags((LocalRegion)region);
      if (tagmapInCache!=null) {
        compareVersionTags(tagmapInCache, tagmapFromRecover);
      }
      int thisRegionSize = startOperations(startKey, value);
      previousRegionSize = thisRegionSize;
      tagmapInCache = saveVersionTags((LocalRegion)region);
      region.close();      
    }    
    } finally {
      if (oldValue != null) {
        System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, oldValue);
      } else {
        System.clearProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
      }
    }
  }
  
  
  ///////////////////////////////////////Set 3 Begin /////////////////
  @Test
  public void testRollingEnabledRecoverValuesFalsePersistWithOverFlowWithEarlyTerminationOfRoller() throws Exception {
    System.setProperty(DiskStoreImpl.COMPLETE_COMPACTION_BEFORE_TERMINATION_PROPERTY_NAME, "false");
//     System.setProperty(DiskRegion.CHECK_ENTRY_BALANCE_PROPERTY_NAME, "true");
    try {
      this.rollingEnabledRecoverValuesFalsePersistWithOverFlow();
    }finally {
//       System.setProperty(DiskRegion.CHECK_ENTRY_BALANCE_PROPERTY_NAME, "false");
    }
  }
  @Test
  public void testRollingEnabledRecoverValuesFalsePersistWithOverFlowWithRollerTerminatingAfterCompletion() throws Exception {
    System.setProperty(DiskStoreImpl.COMPLETE_COMPACTION_BEFORE_TERMINATION_PROPERTY_NAME, "true");
    //System.setProperty(DiskRegion.ASSERT_ON_RECOVERY_PROPERTY_NAME, "true");
    try {
      this.rollingEnabledRecoverValuesFalsePersistWithOverFlow();
    }finally {
      System.setProperty(DiskStoreImpl.COMPLETE_COMPACTION_BEFORE_TERMINATION_PROPERTY_NAME, "false");
      //System.setProperty(DiskRegion.ASSERT_ON_RECOVERY_PROPERTY_NAME, "false");
    }
  }
  
  private  void rollingEnabledRecoverValuesFalsePersistWithOverFlow() throws Exception { 
    String oldValue = System.getProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
    System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, "false");
    try {
    diskProps.setPersistBackup(true);
    diskProps.setRegionName("testRollingDisabledRecoverValuesFalsePersistWithOverFlow");
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE_IN_BYTES);
    diskProps.setRolling(true);
    diskProps.setOverFlowCapacity(100);
    int previousRegionSize = 0;
    HashMap<String, VersionTag> tagmapInCache = null;
    //Num time start / close cycles
    for(int i =0; i < ITERATIONS; ++i) {
      long t1 = System.currentTimeMillis();
      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      long t2 = System.currentTimeMillis();
      System.out.println("Time taken to recover = "+ (t2-t1) + " for total number of entries= "+ region.size() + " with recover values as false and rolling disabled.");
      if ((t2-t1) > 0) {
        System.out.println("Recovery rate is= "+ region.size()/ (t2-t1) + " per milliseconds");
      }
      int startKey =this.processRegionData();
      assertEquals(previousRegionSize,region.size());
      HashMap<String, VersionTag> tagmapFromRecover = saveVersionTags((LocalRegion)region);
      if (tagmapInCache!=null) {
        compareVersionTags(tagmapInCache, tagmapFromRecover);
      }
      int thisRegionSize = startOperations(startKey, value);
      previousRegionSize = thisRegionSize;
      tagmapInCache = saveVersionTags((LocalRegion)region);
      region.close();      
    }    
    } finally {
      if (oldValue != null) {
        System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, oldValue);
      } else {
        System.clearProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
      }
    }
  }
///////////////////////////////////////Set 3 End /////////////////
  
  
  
  //////////////////////////////////////////Set 4 begin/////////////////
  @Test
  public void testRollingEnabledRecoverValuesTruePersistWithOverFlowWithEarlyTerminationOfRoller() throws Exception {
    System.setProperty(DiskStoreImpl.COMPLETE_COMPACTION_BEFORE_TERMINATION_PROPERTY_NAME, "false");
//     System.setProperty(DiskRegion.CHECK_ENTRY_BALANCE_PROPERTY_NAME, "true");
    try {
      this.rollingEnabledRecoverValuesTruePersistWithOverFlow();
    }finally {
//       System.setProperty(DiskRegion.CHECK_ENTRY_BALANCE_PROPERTY_NAME, "false");
    }
  }
  @Test
  public void testRollingEnabledRecoverValuesTruePersistWithOverFlowWithRollerTerminatingAfterCompletion() throws Exception {
    System.setProperty(DiskStoreImpl.COMPLETE_COMPACTION_BEFORE_TERMINATION_PROPERTY_NAME, "true");
    //System.setProperty(DiskRegion.ASSERT_ON_RECOVERY_PROPERTY_NAME, "true");
    try {
      this.rollingEnabledRecoverValuesTruePersistWithOverFlow();
    }finally {
      System.setProperty(DiskStoreImpl.COMPLETE_COMPACTION_BEFORE_TERMINATION_PROPERTY_NAME, "false");
      //System.setProperty(DiskRegion.ASSERT_ON_RECOVERY_PROPERTY_NAME, "false");
    }
  }
  private void rollingEnabledRecoverValuesTruePersistWithOverFlow() throws Exception { 
    String oldValue = System.getProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
    System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, "true");
    try {
    diskProps.setPersistBackup(true);
    diskProps.setRegionName("testRollingDisabledRecoverValuesTruePersistWithOverFlow");
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE_IN_BYTES);
    diskProps.setRolling(true);
    diskProps.setOverFlowCapacity(100);
    int previousRegionSize = 0;
    HashMap<String, VersionTag> tagmapInCache = null;
    //Num time start / close cycles
    for(int i =0; i < ITERATIONS; ++i) {
      long t1 = System.currentTimeMillis();
      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      long t2 = System.currentTimeMillis();
      System.out.println("Time taken to recover = "+ (t2-t1) + " for total number of entries= "+ region.size() + " with recover values as false  and rolling disabled ");
      if ((t2-t1) > 0) {
        System.out.println("Recovery rate is= "+ region.size()/ (t2-t1) + " per milliseconds");
      }
      int startKey =this.processRegionData();
      assertEquals(previousRegionSize,region.size());
      HashMap<String, VersionTag> tagmapFromRecover = saveVersionTags((LocalRegion)region);
      if (tagmapInCache!=null) {
        compareVersionTags(tagmapInCache, tagmapFromRecover);
      }
      int thisRegionSize = startOperations(startKey, value);
      previousRegionSize = thisRegionSize;
      tagmapInCache = saveVersionTags((LocalRegion)region);
      region.close();      
    }   
    } finally {
      if (oldValue != null) {
        System.setProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, oldValue);
      } else {
        System.clearProperty(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME);
      }
    }
  }
  
  //////////////////////////////////////////Set 4 end/////////////////
 
  
  @Test
  public void testRollingDisabledRecoverValuesFalseWithNotifyToRollAPICall() throws Exception { 
    diskProps.setPersistBackup(true);
    diskProps.setRegionName("testRollingDisabledRecoverValuesFalseWithNotifyToRoll");
    diskProps.setMaxOplogSize(MAX_OPLOG_SIZE_IN_BYTES);
    diskProps.setRolling(false);   
    int previousRegionSize = 0;
    HashMap<String, VersionTag> tagmapInCache = null;
    final boolean[] run = new boolean[] {true};
    //Num time start / close cycles
    for(int i =0; i < ITERATIONS; ++i) {
      region = DiskRegionHelperFactory.getSyncPersistOnlyRegion(cache, diskProps, Scope.LOCAL);
      int startKey =this.processRegionData();
      assertEquals(previousRegionSize,region.size());
      HashMap<String, VersionTag> tagmapFromRecover = saveVersionTags((LocalRegion)region);
      if (tagmapInCache!=null) {
        compareVersionTags(tagmapInCache, tagmapFromRecover);
      }
      Thread th = new Thread(new Runnable() {
        public void run() {
          while(run[0]) {
            try {
              Thread.sleep(1*1000);
            }catch(InterruptedException ie) {
              ie.printStackTrace();
            }
            ((LocalRegion)region).notifyToRoll();
            
          }
        }
      });
      int thisRegionSize = startOperations(startKey, value);
      previousRegionSize = thisRegionSize;
      run[0] = false;
      th.join();
      tagmapInCache = saveVersionTags((LocalRegion)region);
      region.close();      
    }    
   
  }

  public int processRegionData() throws Exception
  {
    System.out.println("Total entries in region at start = " + region.size());
    int startKey = 0;
    Iterator entryItr = region.entrySet().iterator();
    startKey = 0;
  
    while (entryItr.hasNext()) {
      Region.Entry entry = (Region.Entry)entryItr.next();
      String key = (String)entry.getKey();
      int indx = key.lastIndexOf('_');
      int temp = Integer.parseInt(key.substring(indx + 1));
      if (temp > startKey) {
        startKey = temp;
      }
      if (entry.getValue() instanceof String) {
        String val = (String)entry.getValue();
        if (!val.equals(value)) {
          throw new IllegalStateException("Values do not match");
        }
      } else if (entry.getValue() instanceof byte[]) {
        byte[] val = (byte[])entry.getValue();
//         System.out.println("Entry " + key + " had a byte array of size " + val.length
//                            + " whose first byte was " + val[0]);
        assertEquals(((byte[])value).length, val.length);
        for (int i=0; i < val.length; i++) {
          assertEquals("at offset " + i,
                       ((byte[])value)[i], val[i]);
        }
      } else {
        assertEquals(value, entry.getValue());
      }
    }
    return ++startKey;
   

  }
 

  /**
   * Disk region recovery test for Persist only with sync writes. Test has four
   * steps : STEP 1: Create cache. Create Region. Put entries. Close cache. STEP
   * 2: Create cache. Create Region with the same name as that of in STEP 1.
   * Delete some entries. 3) Recreate the deleted entries Close the Cache * 3:
   * Again Create cache. 4) check if the region creation is successful
   * 
   * @author Asif
   */
  public int  startOperations( final int startKey, final Object value)
      throws Exception  {

    
    final int NUM_THREADS = 5;
    final long OPS_PER_THREAD = OPS_PER_ITERATION / NUM_THREADS;
    Thread operations[] = new Thread[NUM_THREADS];
    System.out.println("Starting " + NUM_THREADS
                       + " threads to do each do " + OPS_PER_THREAD
                       + " operations");
    for (int i = 0; i < NUM_THREADS; ++i) {
      operations[i] = new Operation(i, region, value, startKey, OPS_PER_THREAD);
    }

    for (int i = 0; i < NUM_THREADS; ++i) {
      operations[i].start();
    }
//     synchronized(this) {
//       this.wait(operationTimeInSec*1000);
//     }
    
//     for (int i = 0; i < NUM_THREADS; ++i) {
//       ((Operation)operations[i]).signalHalt();
//     }

    for (int i = 0; i < NUM_THREADS; ++i) {
      operations[i].join();
    }    
    int regionSize = region.size();
    System.out.println("Total Region Size at end = " + region.size());   
    return regionSize;
  }

  class Operation extends Thread
  {
    final int id;

    volatile boolean run = true;

    int createKeyID = 1;

    final Region rgn;

    Object value;

    volatile int totalEntries = 0;
    final long maxOpCount;

    public Operation(int id, Region rgn, Object value, int startKey, long maxOpCount) {
      this.id = id;
      this.rgn = rgn;
      this.value = value;
      this.createKeyID = startKey;
      this.maxOpCount = maxOpCount;
    }

//     public void signalHalt()
//     {
//       this.run = false;
//     }

    public void run()
    {
      Random opRandom = new Random();
      Random keyRandom = new Random();
      long opCount = 0;
      while (opCount < this.maxOpCount) {
        int optype = opRandom.nextInt(3);
        switch (optype) {
        case 0:
          this.rgn.create(testId + "_" + id + "_" + createKeyID, value);
          opCount++;
          ++createKeyID;
          ++totalEntries;
          break;
        case 1: {
          int key = keyRandom.nextInt(createKeyID);
          try {
            this.rgn.put(testId + "_" + id + "_" + key, value);
            opCount++;
          }
          catch (EntryNotFoundException enfe) {

          }
          catch (EntryDestroyedException ede) {

          }
        }
          break;
        case 2: {
          int key = keyRandom.nextInt(createKeyID);
          try {
            this.rgn.destroy(testId + "_" + id + "_" + key);
            // don't count these as ops; they are tiny
            --totalEntries;
          }
          catch (EntryNotFoundException enfe) {

          }
          catch (EntryDestroyedException ede) {

          }
        }
          break;
        default:
          throw new IllegalStateException();

        }
      }
    }
  }

}