/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.internal.cache;

import static java.lang.System.lineSeparator;

import java.io.BufferedInputStream;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.PrintStream;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.Instantiator;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAlgorithm;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.compression.Compressor;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.InternalInstantiator;
import org.apache.geode.internal.InternalInstantiator.InstantiatorAttributesHolder;
import org.apache.geode.internal.cache.persistence.CanonicalIdHolder;
import org.apache.geode.internal.cache.persistence.DiskExceptionHandler;
import org.apache.geode.internal.cache.persistence.DiskInitFileInterpreter;
import org.apache.geode.internal.cache.persistence.DiskInitFileParser;
import org.apache.geode.internal.cache.persistence.DiskRegionView;
import org.apache.geode.internal.cache.persistence.DiskStoreID;
import org.apache.geode.internal.cache.persistence.PRPersistentConfig;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.cache.persistence.PersistentMemberPattern;
import org.apache.geode.internal.cache.versions.DiskRegionVersionVector;
import org.apache.geode.internal.cache.versions.RegionVersionHolder;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.VersioningIO;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * Does all the IF file work for a DiskStoreImpl.
 *
 *
 * @since GemFire prPersistSprint1
 */
public class DiskInitFile implements DiskInitFileInterpreter {
  private static final Logger logger = LogService.getLogger();

  public static final String IF_FILE_EXT = ".if";

  public static final byte IF_EOF_ID = 0;
  public static final byte END_OF_RECORD_ID = 21;
  static final int OPLOG_FILE_ID_REC_SIZE = 1 + 8 + 1;

  private final ReentrantLock lock = new ReentrantLock();

  /**
   * Written to IF Byte Format: 8: leastSigBits of UUID 8: mostSigBits of UUID 1: EndOfRecordMarker
   */
  public static final byte IFREC_DISKSTORE_ID = 56;

  /**
   * Written to IF Byte Format: 4: instantiatorID 4: classNameLength classNameLength: className
   * bytes 4: instClassNameLength instClassNameLength: instClassName bytes 1: EndOfRecordMarker
   */
  public static final byte IFREC_INSTANTIATOR_ID = 57;

  /**
   * Written to IF Byte Format: 4: classNameLength classNameLength: className bytes 1:
   * EndOfRecordMarker
   */
  public static final byte IFREC_DATA_SERIALIZER_ID = 58;
  /**
   * Written to IF Used to say that persistent member id is online. Byte Format: RegionId 4:
   * blobLength blobLength: member bytes 1: EndOfRecordMarker
   *
   * @since GemFire prPersistSprint1
   */
  public static final byte IFREC_ONLINE_MEMBER_ID = 59;
  /**
   * Written to IF Used to say that persistent member id is offline. Byte Format: RegionId 4:
   * blobLength blobLength: member bytes 1: EndOfRecordMarker
   *
   * @since GemFire prPersistSprint1
   */
  public static final byte IFREC_OFFLINE_MEMBER_ID = 60;
  /**
   * Written to IF Used to say that persistent member id no longer exists. Byte Format: RegionId 4:
   * blobLength blobLength: member bytes 1: EndOfRecordMarker
   *
   * @since GemFire prPersistSprint1
   */
  public static final byte IFREC_RM_MEMBER_ID = 61;
  /**
   * Written to IF. Used to record the persistent member id of this file. Byte Format: RegionId 4:
   * blobLength blobLength: member bytes 1: EndOfRecordMarker
   *
   * @since GemFire prPersistSprint1
   */
  public static final byte IFREC_MY_MEMBER_INITIALIZING_ID = 62;
  /**
   * Written to IF. Used to record the previous my member id completed initialization. Byte Format:
   * RegionId 1: EndOfRecordMarker
   *
   * @since GemFire prPersistSprint1
   */
  public static final byte IFREC_MY_MEMBER_INITIALIZED_ID = 63;

  /**
   * Written to IF. Used to record create of region Byte Format: RegionId 4: nameLength nameLength:
   * name bytes 1: EndOfRecordMarker
   *
   * @since GemFire prPersistSprint1
   */
  public static final byte IFREC_CREATE_REGION_ID = 64;
  /**
   * Written to IF. Used to record begin of destroy of region Byte Format: RegionId 1:
   * EndOfRecordMarker
   *
   * @since GemFire prPersistSprint1
   */
  public static final byte IFREC_BEGIN_DESTROY_REGION_ID = 65;
  /**
   * Written to IF. Used to record clear of region Byte Format: RegionId 8: oplogEntryId 1:
   * EndOfRecordMarker
   *
   * @since GemFire prPersistSprint1
   */
  public static final byte IFREC_CLEAR_REGION_ID = 66;
  /**
   * Written to IF. Used to record that the end of a destroy region. Byte Format: RegionId 1:
   * EndOfRecordMarker
   *
   * @since GemFire prPersistSprint1
   */
  public static final byte IFREC_END_DESTROY_REGION_ID = 67;
  /**
   * Written to IF. Used to record that a region is about to be partially destroyed Byte Format:
   * RegionId 1: EndOfRecordMarker
   *
   * @since GemFire prPersistSprint1
   */
  public static final byte IFREC_BEGIN_PARTIAL_DESTROY_REGION_ID = 68;
  /**
   * Written to IF. Used to record that a region is partially destroyed Byte Format: RegionId 1:
   * EndOfRecordMarker
   *
   * @since GemFire prPersistSprint1
   */
  public static final byte IFREC_END_PARTIAL_DESTROY_REGION_ID = 69;

  /**
   * Records the creation of an oplog crf file Byte Format: 8: oplogId 1: EndOfRecord
   */
  public static final byte IFREC_CRF_CREATE = 70;

  /**
   * Records the creation of an oplog drf file Byte Format: 8: oplogId 1: EndOfRecord
   */
  public static final byte IFREC_DRF_CREATE = 71;

  /**
   * Records the deletion of an oplog crf file Byte Format: 8: oplogId 1: EndOfRecord
   */
  public static final byte IFREC_CRF_DELETE = 72;

  /**
   * Records the deletion of an oplog drf file Byte Format: 8: oplogId 1: EndOfRecord
   */
  public static final byte IFREC_DRF_DELETE = 73;

  /**
   * Written to IF. Used to record regions config Byte Format: RegionId 1: lruAlgorithm 1: lruAction
   * 4: lruLimit (int) // no need to ObjectSize during recovery since all data is in blob form 4:
   * concurrencyLevel (int) 4: initialCapacity (int) 4: loadFactor (float) 1: statisticsEnabled
   * (boolean) 1: isBucket (boolean) 1: EndOfRecordMarker
   *
   * Used to read the region configuration for 6.5 disk stores.
   *
   */
  public static final byte IFREC_REGION_CONFIG_ID = 74;

  /*
   * Written to IF Used to say that persistent member id is offline and has the same data on disk as
   * this member Byte Format: RegionId 4: blobLength blobLength: member bytes 1: EndOfRecordMarker
   *
   * @since GemFire prPersistSprint3
   */
  public static final byte IFREC_OFFLINE_AND_EQUAL_MEMBER_ID = 75;
  /**
   * Written to IF. Used to record regions config Byte Format: RegionId 1: lruAlgorithm 1: lruAction
   * 4: lruLimit (int) // no need to ObjectSize during recovery since all data is in blob form 4:
   * concurrencyLevel (int) 4: initialCapacity (int) 4: loadFactor (float) 1: statisticsEnabled
   * (boolean) 1: isBucket (boolean) 4: length of partitionName String bytes (int) length:actual
   * bytes 4: startingBucketId(int) 1: EndOfRecordMarker
   */
  public static final byte IFREC_REGION_CONFIG_ID_66 = 76;


  /**
   * Records the creation of an oplog krf file The presence of this record indicates that the krf
   * file is complete. Byte Format: 8: oplogId 1: EndOfRecord
   */
  public static final byte IFREC_KRF_CREATE = 77;

  /**
   * Records the creation of a persistent partitioned region configuration. Byte Format: variable:
   * pr name 4: total num buckets variable: colocated with 1: EndOfRecord
   */
  public static final byte IFREC_PR_CREATE = 78;

  /**
   * Records the deletion of persistent partitioned region. Byte Format: variable: pr name 1:
   * EndOfRecord
   */
  public static final byte IFREC_PR_DESTROY = 79;


  /**
   * Maps a member id (either a disk store ID or a distributed system id plus a byte) to a single
   * integer, which can be used in oplogs.
   *
   * Byte Format: 4: the number assigned to this id. variable: serialized object representing the
   * ID. variable: pr name 1: EndOfRecord
   */
  public static final byte IFREC_ADD_CANONICAL_MEMBER_ID = 80;
  /**
   * Written to IF Used to say that a disk store has been revoked Byte Format: variable: a
   * PersistentMemberPattern
   *
   * @since GemFire 7.0
   */
  public static final byte IFREC_REVOKE_DISK_STORE_ID = 81;

  /**
   * Written gemfire version to IF Byte Format: 1: version byte from Version.GFE_CURRENT.ordinal 1:
   * EndOfRecord
   *
   * @since GemFire 7.0
   */
  public static final byte IFREC_GEMFIRE_VERSION = 82;

  /**
   * Written to IF. Used to record clear of using an RVV Byte Format: RegionId variable: serialized
   * RVV 1: EndOfRecordMarker
   *
   * @since GemFire 7.0
   */
  public static final byte IFREC_CLEAR_REGION_WITH_RVV_ID = 83;

  /**
   * Written to IF. Used to record regions config Byte Format: RegionId 1: lruAlgorithm 1: lruAction
   * 4: lruLimit (int) // no need to ObjectSize during recovery since all data is in blob form 4:
   * concurrencyLevel (int) 4: initialCapacity (int) 4: loadFactor (float) 1: statisticsEnabled
   * (boolean) 1: isBucket (boolean) variable: partitionName (utf) 4: startingBucketId (int)
   * variable: compressorClassName (utf) 1: versioned (boolean) 1: EndOfRecordMarker
   *
   */
  public static final byte IFREC_REGION_CONFIG_ID_80 = 88;

  /**
   * Persist oplog file magic number. Written once at the beginning of every oplog file; CRF, DRF,
   * KRF, and IF. Followed by 6 byte magic number. Each oplog type has a different magic number
   * Followed by EndOfRecord Fix for bug 43824
   *
   * @since GemFire 8.0
   */
  public static final byte OPLOG_MAGIC_SEQ_ID = 89;

  /**
   * Written to IF. Used to record regions config Byte Format: RegionId 1: lruAlgorithm 1: lruAction
   * 4: lruLimit (int) // no need to ObjectSize during recovery since all data is in blob form 4:
   * concurrencyLevel (int) 4: initialCapacity (int) 4: loadFactor (float) 1: statisticsEnabled
   * (boolean) 1: isBucket (boolean) variable: partitionName (utf) 4: startingBucketId (int)
   * variable: compressorClassName (utf) 1: versioned (boolean) 1: offHeap (boolean) added in 9.0 1:
   * EndOfRecordMarker
   *
   * @since Geode 1.0
   */
  public static final byte IFREC_REGION_CONFIG_ID_90 = 90;

  private final DiskStoreImpl parent;

  private final File ifFile;
  private RandomAccessFile ifRAF;
  private boolean closed;
  // contains the ids of dataSerializers already written to IF
  private final IntOpenHashSet dsIds;
  // contains the ids of instantiators already written to IF
  private final IntOpenHashSet instIds;

  private final LongOpenHashSet crfIds;
  private final LongOpenHashSet drfIds;
  private final Set<Long> krfIds;

  /**
   * Map used to keep track of regions we know of from the DiskInitFile but that do not yet exist
   * (they have not yet been recovered or they have been closed).
   */
  private final Map<Long, PlaceHolderDiskRegion> drMap = new HashMap<>();
  private final Map<String, PlaceHolderDiskRegion> drMapByName = new HashMap<>();

  /**
   * Map of persistent partitioned regions configurations that are stored in this init file.
   */
  private final Map<String, PRPersistentConfig> prMap = new HashMap<>();

  private final InternalDataSerializer.RegistrationListener regListener;

  private int ifLiveRecordCount = 0;
  private int ifTotalRecordCount = 0;
  private boolean compactInProgress;
  // the recovered version
  private KnownVersion gfversion;


  /**
   * Used to calculate the highest oplog entry id we have seen in a clear entry.
   */
  private long clearOplogEntryIdHWM = DiskStoreImpl.INVALID_ID;

  /**
   * Container for canonical ids held in the disk store. Member ids are canonicalized so they can be
   * written as an integer in the oplogs.
   */
  private final CanonicalIdHolder canonicalIdHolder = new CanonicalIdHolder();

  /**
   * Set of members that have been revoked. We keep track of the revoked members so that we can
   * indicate to the user a member has been revoked, rather is simply conflicting
   */
  private final Set<PersistentMemberPattern> revokedMembers =
      new HashSet<>();

  private transient long nextSeekPosition;

  private transient boolean gotEOF;

  private void lock(boolean useBackupLock) {
    if (useBackupLock) {
      getDiskStore().getBackupLock().lock();
    }
    lock.lock();
  }

  private void unlock(boolean useBackupLock) {
    if (useBackupLock) {
      getDiskStore().getBackupLock().unlock();
    }
    lock.unlock();
  }

  private void recoverFromFailedCompaction() {
    File tmpFile = getTempFile();
    if (tmpFile.exists()) {
      // if the temp init file exists then we must have crashed during a compaction.
      // In this case we need to destroy the non temp file and rename the temp file.
      if (ifFile.exists()) {
        if (!ifFile.delete()) {
          throw new IllegalStateException("Could not delete " + ifFile);
        }
        if (!tmpFile.renameTo(ifFile)) {
          throw new IllegalStateException("Could not rename " + tmpFile + " to " + ifFile);
        }
      }
    }
  }

  private DiskStoreImpl getDiskStore() {
    return parent;
  }

  public KnownVersion currentRecoveredGFVersion() {
    return gfversion;
  }

  DiskStoreID recover() {
    recoverFromFailedCompaction();
    if (!ifFile.exists()) {
      // nothing to recover
      // Instead of calling randomUUID which uses SecureRandom which can be slow
      // return UUID.randomUUID();
      // create a UUID using the cheaper Random class.
      return new DiskStoreID(UUID.randomUUID());
    }
    DiskStoreID result = null;
    try {
      try (final FileInputStream fis = new FileInputStream(ifFile);
          final CountingDataInputStream dis =
              new CountingDataInputStream(new BufferedInputStream(fis, 8 * 1024),
                  ifFile.length())) {
        DiskInitFileParser parser = new DiskInitFileParser(dis, this);
        result = parser.parse();

        gotEOF = parser.gotEOF();
        nextSeekPosition = dis.getCount();
        if (logger.isTraceEnabled(LogMarker.PERSIST_RECOVERY_VERBOSE)) {
          logger.trace(LogMarker.PERSIST_RECOVERY_VERBOSE, "liveRecordCount={} totalRecordCount={}",
              ifLiveRecordCount, ifTotalRecordCount);
        }
      }
      for (PlaceHolderDiskRegion drv : drMap.values()) {
        if (drv.getMyPersistentID() != null || drv.getMyInitializingID() != null) {
          // Prepare each region we found in the init file for early recovery.
          if (drv.isBucket() || !getDiskStore().getOwnedByRegion()) {
            getDiskStore().getStats().incUncreatedRecoveredRegions(1);
            drv.setRecoveredEntryMap(RegionMapFactory.createVM(drv, getDiskStore(),
                getDiskStore().getInternalRegionArguments()));
            if (!getDiskStore().isOffline()) {
              // schedule it for recovery since we want to recovery region data early now
              getDiskStore().scheduleForRecovery(drv);
            }
            // else if we are validating or offlineCompacting
            // then the scheduleForRecovery is called later in DiskStoreImpl
            // this helps fix bug 42043
          }
        }
      }
    } catch (EOFException ex) {
      // ignore since a partial record write can be caused by a crash
      // throw new
      // DiskAccessException(String.format("Failed to read file during recovery from %s",
      // this.ifFile.getPath()), ex, this.parent);
    } catch (ClassNotFoundException | IOException ex) {
      throw new DiskAccessException(
          String.format("Failed to read file during recovery from %s",
              ifFile.getPath()),
          ex, parent);
    } catch (CancelException e) {
      if (logger.isDebugEnabled()) {
        logger.debug("Oplog::readOplog:Error in recovery as Cache was closed", e);
      }
    } catch (RegionDestroyedException e) {
      if (logger.isDebugEnabled()) {
        logger.debug("Oplog::readOplog:Error in recovery as Region was destroyed", e);
      }
    } catch (IllegalStateException ex) {
      if (!parent.isClosing()) {
        throw ex;
      }
    }
    return result;
  }

  @Override
  public void cmnClearRegion(long drId, long clearOplogEntryId) {
    DiskRegionView drv = getDiskRegionById(drId);
    if (drv.getClearOplogEntryId() == DiskStoreImpl.INVALID_ID) {
      ifLiveRecordCount++;
    }
    // otherwise previous clear is cancelled so don't change liveRecordCount
    ifTotalRecordCount++;
    drv.setClearOplogEntryId(clearOplogEntryId);
    if (clearOplogEntryId > clearOplogEntryIdHWM) {
      clearOplogEntryIdHWM = clearOplogEntryId;
    }
  }

  @Override
  public void cmnClearRegion(long drId,
      ConcurrentHashMap<DiskStoreID, RegionVersionHolder<DiskStoreID>> memberToVersion) {
    DiskRegionView drv = getDiskRegionById(drId);
    if (drv.getClearRVV() == null) {
      ifLiveRecordCount++;
    }
    // otherwise previous clear is cancelled so don't change liveRecordCount
    ifTotalRecordCount++;

    DiskStoreID ownerId = parent.getDiskStoreID();
    // Create a fake RVV for clear purposes. We only need to memberToVersion information
    RegionVersionHolder<DiskStoreID> ownerExceptions = memberToVersion.remove(ownerId);
    long ownerVersion = ownerExceptions == null ? 0 : ownerExceptions.getVersion();
    DiskRegionVersionVector rvv =
        new DiskRegionVersionVector(ownerId, memberToVersion, ownerVersion,
            new ConcurrentHashMap<>(), 0L, false, ownerExceptions);
    drv.setClearRVV(rvv);
  }

  private int liveRegions = 0; // added for bug 41618

  public boolean hasLiveRegions() {
    lock(false);
    try {
      return liveRegions > 0;
    } finally {
      unlock(false);
    }
  }

  @Override
  public void cmnCreateRegion(long drId, String regName) {
    recoverDiskRegion(drId, regName);
    liveRegions++;
    ifLiveRecordCount++;
    ifTotalRecordCount++;
  }

  @Override
  public void cmnRegionConfig(long drId, byte lruAlgorithm, byte lruAction, int lruLimit,
      int concurrencyLevel, int initialCapacity, float loadFactor, boolean statisticsEnabled,
      boolean isBucket, EnumSet<DiskRegionFlag> flags, String partitionName, int startingBucketId,
      String compressorClassName, boolean offHeap) {
    DiskRegionView dr = getDiskRegionById(drId);
    if (dr != null) {
      // We don't want gateway queues to turn on versioning. Unfortunately, the only
      // way to indentify that a region is a gateway queue is by the region
      // name.
      if (!(dr.getName().contains("_SERIAL_GATEWAY_SENDER_QUEUE")
          || dr.getName().contains("_PARALLEL__GATEWAY__SENDER__QUEUE"))) {
        flags.add(DiskRegionFlag.IS_WITH_VERSIONING);
      }
      dr.setConfig(lruAlgorithm, lruAction, lruLimit, concurrencyLevel, initialCapacity, loadFactor,
          statisticsEnabled, isBucket, flags, partitionName, startingBucketId, compressorClassName,
          offHeap);

      // Just count this as a live record even though it is possible
      // that we have an extra one due to the config changing while
      // we were offline.
      ifLiveRecordCount++;
      ifTotalRecordCount++;
    } else {
      if (logger.isTraceEnabled(LogMarker.PERSIST_RECOVERY_VERBOSE)) {
        logger.trace(LogMarker.PERSIST_RECOVERY_VERBOSE, "bad disk region id!");
      } else {
        throw new IllegalStateException("bad disk region id");
      }
    }
  }

  @Override
  public boolean cmnPRCreate(String name, PRPersistentConfig config) {
    if (prMap.put(name, config) == null) {
      ifLiveRecordCount++;
      ifTotalRecordCount++;
      liveRegions++;
      return true;
    }

    return false;
  }

  @Override
  public void cmnGemfireVersion(KnownVersion version) {
    gfversion = version;
  }

  @Override
  public boolean cmnPRDestroy(String name) {
    if (prMap.remove(name) != null) {
      ifLiveRecordCount--;
      ifTotalRecordCount++;
      liveRegions--;
      return true;
    }
    return false;
  }

  @Override
  public void cmnAddCanonicalMemberId(int id, Object object) {
    canonicalIdHolder.addMapping(id, object);
    ifLiveRecordCount++;
    ifTotalRecordCount++;
  }

  @Override
  public void cmnRmMemberId(long drId, PersistentMemberID pmid) {
    DiskRegionView dr = getDiskRegionById(drId);
    if (dr != null) {
      if (!dr.rmOnlineMember(pmid)) {
        if (!dr.rmOfflineMember(pmid)) {
          dr.rmEqualMember(pmid);
        }
      }
      // since we removed a member don't inc the live count
      // In fact decrement it by one since both this record
      // and the previous one are both garbage.
      ifLiveRecordCount--;
      ifTotalRecordCount++;
    } else {
      if (logger.isTraceEnabled(LogMarker.PERSIST_RECOVERY_VERBOSE)) {
        logger.trace(LogMarker.PERSIST_RECOVERY_VERBOSE, "bad disk region id!");
      } else {
        throw new IllegalStateException("bad disk region id");
      }
    }
  }

  @Override
  public void cmnOfflineMemberId(long drId, PersistentMemberID pmid) {
    DiskRegionView dr = getDiskRegionById(drId);
    if (dr != null) {
      dr.addOfflineMember(pmid);
      if (dr.rmOnlineMember(pmid) || dr.rmEqualMember(pmid)) {
        ifLiveRecordCount--;
      }
      ifLiveRecordCount++;
      ifTotalRecordCount++;
    } else {
      if (logger.isTraceEnabled(LogMarker.PERSIST_RECOVERY_VERBOSE)) {
        logger.trace(LogMarker.PERSIST_RECOVERY_VERBOSE, "bad disk region id!");
      } else {
        throw new IllegalStateException("bad disk region id");
      }
    }
  }

  @Override
  public void cmdOfflineAndEqualMemberId(long drId, PersistentMemberID pmid) {
    DiskRegionView dr = getDiskRegionById(drId);
    if (dr != null) {
      dr.addOfflineAndEqualMember(pmid);
      if (dr.rmOnlineMember(pmid) || dr.rmOfflineMember(pmid)) {
        ifLiveRecordCount--;
      }
      ifLiveRecordCount++;
      ifTotalRecordCount++;
    } else {
      if (logger.isTraceEnabled(LogMarker.PERSIST_RECOVERY_VERBOSE)) {
        logger.trace(LogMarker.PERSIST_RECOVERY_VERBOSE, "bad disk region id!");
      } else {
        throw new IllegalStateException("bad disk region id");
      }
    }
  }

  @Override
  public void cmnOnlineMemberId(long drId, PersistentMemberID pmid) {
    DiskRegionView dr = getDiskRegionById(drId);
    if (dr != null) {
      dr.addOnlineMember(pmid);
      if (dr.rmOfflineMember(pmid) || dr.rmEqualMember(pmid)) {
        ifLiveRecordCount--;
      }
      ifLiveRecordCount++;
      ifTotalRecordCount++;
    } else {
      if (logger.isTraceEnabled(LogMarker.PERSIST_RECOVERY_VERBOSE)) {
        logger.trace(LogMarker.PERSIST_RECOVERY_VERBOSE, "bad disk region id!");
      } else {
        throw new IllegalStateException("bad disk region id");
      }
    }
  }

  @Override
  public void cmnDataSerializerId(Class<? extends DataSerializer> dsc) {
    if (dsc != null) {
      DataSerializer ds = InternalDataSerializer.register(dsc, /* dsId, */ true);
      dsIds.add(ds.getId());
    }
    ifLiveRecordCount++;
    ifTotalRecordCount++;
  }

  @Override
  public void cmnInstantiatorId(int id, Class<?> c, Class<?> ic) {
    if (c != null && ic != null) {
      InternalInstantiator.register(c, ic, id, true);
      instIds.add(id);
    }
    ifLiveRecordCount++;
    ifTotalRecordCount++;
  }

  @Override
  public void cmnInstantiatorId(int id, String cn, String icn) {
    if (cn != null && icn != null) {
      InternalInstantiator.register(cn, icn, id, true);
      instIds.add(id);
    }
    ifLiveRecordCount++;
    ifTotalRecordCount++;
  }

  @Override
  public void cmnCrfCreate(long oplogId) {
    crfIds.add(oplogId);
    ifLiveRecordCount++;
    ifTotalRecordCount++;
  }

  @Override
  public void cmnDrfCreate(long oplogId) {
    drfIds.add(oplogId);
    ifLiveRecordCount++;
    ifTotalRecordCount++;
  }

  @Override
  public void cmnKrfCreate(long oplogId) {
    krfIds.add(oplogId);
    ifLiveRecordCount++;
    ifTotalRecordCount++;
  }

  @Override
  public boolean cmnCrfDelete(long oplogId) {
    if (krfIds.remove(oplogId)) {
      ifLiveRecordCount--;
      ifTotalRecordCount++;
    }
    if (crfIds.remove(oplogId)) {
      ifLiveRecordCount--;
      ifTotalRecordCount++;
      return true;
    } else {
      return false;
    }
  }

  @Override
  public boolean cmnDrfDelete(long oplogId) {
    if (drfIds.remove(oplogId)) {
      ifLiveRecordCount--;
      ifTotalRecordCount++;
      return true;
    } else {
      return false;
    }
  }

  public boolean isCRFOplogIdPresent(long crfId) {
    return crfIds.contains(crfId);
  }

  public boolean isDRFOplogIdPresent(long drfId) {
    return drfIds.contains(drfId);
  }

  public void verifyOplogs(LongOpenHashSet foundCrfs, LongOpenHashSet foundDrfs) {
    verifyOplogs(foundCrfs, foundDrfs, crfIds, drfIds);
  }

  public void verifyOplogs(LongOpenHashSet foundCrfs, LongOpenHashSet foundDrfs,
      LongOpenHashSet expectedCrfIds, LongOpenHashSet expectedDrfIds) {
    LongOpenHashSet missingCrfs = calcMissing(foundCrfs, expectedCrfIds);
    LongOpenHashSet missingDrfs = calcMissing(foundDrfs, expectedDrfIds);
    // Note that finding extra ones is ok; it is possible we died just
    // after creating one but before we could record it in the if file
    // Or died just after deleting it but before we could record it in the if file.
    boolean failed = false;
    String msg = null;
    if (!missingCrfs.isEmpty()) {
      failed = true;
      msg = "*.crf files with these ids: " + Arrays.toString(missingCrfs.toArray());
    }
    if (!missingDrfs.isEmpty()) {
      failed = true;
      if (msg == null) {
        msg = "";
      } else {
        msg += ", ";
      }
      msg += "*.drf files with these ids: " + Arrays.toString(missingDrfs.toArray());
    }
    if (failed) {
      msg = "The following required files could not be found: " + msg + ".";
      throw new IllegalStateException(msg);
    }
  }

  private LongOpenHashSet calcMissing(LongOpenHashSet found, LongOpenHashSet expected) {
    LongOpenHashSet missing = new LongOpenHashSet(expected);
    missing.removeAll(found);
    return missing;
  }

  boolean hasKrf(long oplogId) {
    return krfIds.contains(oplogId);
  }

  DiskRegionView takeDiskRegionByName(String name) {
    lock(false);
    try {
      DiskRegionView result = drMapByName.remove(name);
      if (result != null) {
        drMap.remove(result.getId());
      }
      return result;
    } finally {
      unlock(false);
    }
  }

  Map<Long, PlaceHolderDiskRegion> getDRMap() {
    lock(false);
    try {
      return new HashMap<>(drMap);
    } finally {
      unlock(false);
    }
  }

  DiskRegion createDiskRegion(DiskStoreImpl dsi, String name, boolean isBucket,
      boolean isPersistBackup, boolean overflowEnabled, boolean isSynchronous,
      DiskRegionStats stats, CancelCriterion cancel, DiskExceptionHandler exceptionHandler,
      RegionAttributes<?, ?> ra, EnumSet<DiskRegionFlag> flags, String partitionName,
      int startingBucketId, Compressor compressor, boolean offHeap) {
    lock(true);
    try {
      // need to call the constructor and addDiskRegion while synced
      DiskRegion result = new DiskRegion(dsi, name, isBucket, isPersistBackup, overflowEnabled,
          isSynchronous, stats, cancel, exceptionHandler, ra, flags, partitionName,
          startingBucketId, compressor == null ? null : compressor.getClass().getName(), offHeap);
      dsi.addDiskRegion(result);
      return result;
    } finally {
      unlock(true);
    }
  }

  DiskRegionView getDiskRegionByName(String name) {
    lock(false);
    try {
      return drMapByName.get(name);
    } finally {
      unlock(false);
    }
  }

  DiskRegionView getDiskRegionByPrName(String name) {
    lock(false);
    try {
      for (PlaceHolderDiskRegion dr : drMapByName.values()) {
        if (dr.isBucket()) {
          if (name.equals(dr.getPrName())) {
            return dr;
          }
        }
      }
      return null;
    } finally {
      unlock(false);
    }
  }

  private DiskRegionView getDiskRegionById(long drId) {
    DiskRegionView result = drMap.get(drId);
    if (result == null) {
      result = parent.getById(drId);
    }
    return result;
  }


  private void recoverDiskRegion(long drId, String regName) {
    // Whatever the last create region drId we see we remember
    // in the DiskStore. Note that this could be a region that is destroyed
    // (we will not know until we see a later destroy region record)
    // but since drId's can wrap around into negative numbers whatever
    // the last one we see is the right one to remember.
    parent.recoverRegionId(drId);
    PlaceHolderDiskRegion dr = new PlaceHolderDiskRegion(parent, drId, regName);
    Object old = drMap.put(drId, dr);
    assert old == null;
    PlaceHolderDiskRegion oldDr = drMapByName.put(regName, dr);
    if (oldDr != null) {
      drMap.remove(oldDr.getId()); // fix for bug 42043
    }
    // don't schedule for recovery until we know it was not destroyed
  }

  /**
   * Maximum number of bytes used to encode a DiskRegion id.
   */
  static final int DR_ID_MAX_BYTES = 9;

  private void writeIFRecord(byte b, DiskRegionView dr) {
    assert lock.isHeldByCurrentThread();
    try {
      ByteBuffer bb = getIFWriteBuffer(1 + DR_ID_MAX_BYTES + 1);
      bb.put(b);
      putDiskRegionID(bb, dr.getId());
      bb.put(END_OF_RECORD_ID);
      writeIFRecord(bb, false); // don't do stats for these small records
    } catch (IOException ex) {
      DiskAccessException dae = new DiskAccessException(
          String.format("Failed writing data to initialization file because: %s", ex),
          parent);
      if (!compactInProgress) {
        parent.handleDiskAccessException(dae);
      }
      throw dae;
    }
  }

  private void writeIFRecord(byte b, DiskRegionView dr, long v) {
    assert lock.isHeldByCurrentThread();
    try {
      ByteBuffer bb = getIFWriteBuffer(1 + DR_ID_MAX_BYTES + 8 + 1);
      bb.put(b);
      putDiskRegionID(bb, dr.getId());
      bb.putLong(v);
      bb.put(END_OF_RECORD_ID);
      writeIFRecord(bb, false); // don't do stats for these small records
    } catch (IOException ex) {
      DiskAccessException dae = new DiskAccessException(
          String.format("Failed writing data to initialization file because: %s", ex),
          parent);
      if (!compactInProgress) {
        parent.handleDiskAccessException(dae);
