/*
 * 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 java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
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.longs.Long2ObjectMap;
import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
import it.unimi.dsi.fastutil.objects.ObjectIterator;
import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.SerializationException;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.EntryDestroyedException;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.distributed.OplogCancelledException;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.InternalStatisticsDisabledException;
import org.apache.geode.internal.Sendable;
import org.apache.geode.internal.cache.DiskInitFile.DiskRegionFlag;
import org.apache.geode.internal.cache.DiskStoreImpl.OplogCompactor;
import org.apache.geode.internal.cache.DiskStoreImpl.OplogEntryIdSet;
import org.apache.geode.internal.cache.DistributedRegion.DiskPosition;
import org.apache.geode.internal.cache.entries.DiskEntry;
import org.apache.geode.internal.cache.entries.DiskEntry.Helper.Flushable;
import org.apache.geode.internal.cache.entries.DiskEntry.Helper.ValueWrapper;
import org.apache.geode.internal.cache.eviction.EvictionController;
import org.apache.geode.internal.cache.eviction.EvictionList;
import org.apache.geode.internal.cache.persistence.BytesAndBits;
import org.apache.geode.internal.cache.persistence.DiskRecoveryStore;
import org.apache.geode.internal.cache.persistence.DiskRegionView;
import org.apache.geode.internal.cache.persistence.DiskStoreID;
import org.apache.geode.internal.cache.persistence.UninterruptibleFileChannel;
import org.apache.geode.internal.cache.persistence.UninterruptibleRandomAccessFile;
import org.apache.geode.internal.cache.versions.CompactVersionHolder;
import org.apache.geode.internal.cache.versions.RegionVersionHolder;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.cache.versions.VersionHolder;
import org.apache.geode.internal.cache.versions.VersionSource;
import org.apache.geode.internal.cache.versions.VersionStamp;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.offheap.OffHeapHelper;
import org.apache.geode.internal.offheap.ReferenceCountHelper;
import org.apache.geode.internal.offheap.StoredObject;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.offheap.annotations.Retained;
import org.apache.geode.internal.sequencelog.EntryLogger;
import org.apache.geode.internal.serialization.ByteArrayDataInput;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.Versioning;
import org.apache.geode.internal.serialization.VersioningIO;
import org.apache.geode.internal.shared.NativeCalls;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.pdx.internal.PdxWriterImpl;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Implements an operation log to write to disk. As of prPersistSprint2 this file only supports
 * persistent regions. For overflow only regions see {@link OverflowOplog}.
 *
 * @since GemFire 5.1
 */
public class Oplog implements CompactableOplog, Flushable {
  private static final Logger logger = LogService.getLogger();

  /* System property to override the disk store write buffer size. */
  public final String WRITE_BUFFER_SIZE_SYS_PROP_NAME = "WRITE_BUF_SIZE";

  /** Extension of the oplog file * */
  public static final String CRF_FILE_EXT = ".crf";
  public static final String DRF_FILE_EXT = ".drf";
  public static final String KRF_FILE_EXT = ".krf";

  /** The file which will be created on disk * */
  private File diskFile;

  /** boolean marked true when this oplog is closed * */
  private volatile boolean closed;

  private final OplogFile crf = new OplogFile();
  private final OplogFile drf = new OplogFile();
  private final KRFile krf = new KRFile();

  /** The stats for this store */
  private final DiskStoreStats stats;

  /** The store that owns this Oplog* */
  private final DiskStoreImpl parent;

  /**
   * The oplog set this oplog is part of
   */
  private final PersistentOplogSet oplogSet;

  /** oplog id * */
  protected final long oplogId;

  /** recovered gemfire version * */
  protected KnownVersion gfversion;

  /**
   * Recovered version of the data. Usually this is same as {@link #gfversion} except for the case
   * of upgrading disk store from previous version in which case the keys/values are carried forward
   * as is and need to be interpreted in load by latest product code if required.
   */
  protected KnownVersion dataVersion;

  /** Directory in which the file is present* */
  private DirectoryHolder dirHolder;

  /** The max Oplog size (user configurable) * */
  private final long maxOplogSize;
  private long maxCrfSize;
  private long maxDrfSize;

  private final AtomicBoolean hasDeletes = new AtomicBoolean();

  private boolean firstRecord = true;

  /**
   * This system property instructs that writes be synchronously written to disk and not to file
   * system. (Use rwd instead of rw - RandomAccessFile property)
   */
  private static final boolean SYNC_WRITES =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "syncWrites");

  /**
   * The HighWaterMark of recentValues.
   */
  private final AtomicLong totalCount = new AtomicLong(0);

  /**
   * The number of records in this oplog that contain the most recent value of the entry.
   */
  private final AtomicLong totalLiveCount = new AtomicLong(0);

  /**
   * Set to true once compact is called on this oplog.
   *
   * @since GemFire prPersistSprint1
   */
  private volatile boolean compacting = false;

  /**
   * Set to true after the first drf recovery.
   */
  private boolean haveRecoveredDrf = true;

  /**
   * Set to true after the first crf recovery.
   */
  private boolean haveRecoveredCrf = true;

  private OpState opState;

  /**
   * Written to CRF, and DRF.
   */
  private static final byte OPLOG_EOF_ID = 0;

  private static final byte END_OF_RECORD_ID = 21;
  /**
   * Written to CRF and DRF. Followed by 16 bytes which is the leastSigBits and mostSigBits of a
   * UUID for the disk store we belong to. 1: EndOfRecord Is written once at the beginning of every
   * oplog file.
   */
  private static final byte OPLOG_DISK_STORE_ID = 62;

  static final int OPLOG_DISK_STORE_REC_SIZE = 1 + 16 + 1;
  /**
   * Written to CRF. Followed by 8 bytes which is the BASE_ID to use for any NEW_ENTRY records. 1:
   * EndOfRecord Only needs to be written once per oplog and must preceed any OPLOG_NEW_ENTRY_0ID
   * records.
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_NEW_ENTRY_BASE_ID = 63;

  static final int OPLOG_NEW_ENTRY_BASE_REC_SIZE = 1 + 8 + 1;

  /**
   * Written to CRF. The OplogEntryId is +1 the previous new_entry OplogEntryId. Byte Format: 1:
   * userBits RegionId 4: valueLength (optional depending on bits) valueLength: value bytes
   * (optional depending on bits) 4: keyLength keyLength: key bytes 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_NEW_ENTRY_0ID = 64;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 1 byte. Byte Format: 1: userBits 1: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_1ID = 65;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 2 bytes. Byte Format: 1: userBits 2: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_2ID = 66;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 3 bytes. Byte Format: 1: userBits 3: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_3ID = 67;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 4 bytes. Byte Format: 1: userBits 4: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_4ID = 68;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 5 bytes. Byte Format: 1: userBits 5: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_5ID = 69;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 6 bytes. Byte Format: 1: userBits 6: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_6ID = 70;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 7 bytes. Byte Format: 1: userBits 7: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_7ID = 71;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 8 bytes. Byte Format: 1: userBits 8: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_8ID = 72;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 1 byte. Byte Format: 1: userBits 1: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 4: keyLength keyLength: key bytes 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_WITH_KEY_1ID = 73;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 2 bytes. Byte Format: 1: userBits 2: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 4: keyLength keyLength: key bytes 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_WITH_KEY_2ID = 74;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 3 bytes. Byte Format: 1: userBits 3: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 4: keyLength keyLength: key bytes 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_WITH_KEY_3ID = 75;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 4 bytes. Byte Format: 1: userBits 4: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 4: keyLength keyLength: key bytes 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_WITH_KEY_4ID = 76;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 5 bytes. Byte Format: 1: userBits 5: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 4: keyLength keyLength: key bytes 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_WITH_KEY_5ID = 77;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 6 bytes. Byte Format: 1: userBits 6: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 4: keyLength keyLength: key bytes 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_WITH_KEY_6ID = 78;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 7 bytes. Byte Format: 1: userBits 7: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 4: keyLength keyLength: key bytes 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_WITH_KEY_7ID = 79;

  /**
   * Written to CRF. The OplogEntryId is relative to the previous mod_entry OplogEntryId. The signed
   * difference is encoded in 8 bytes. Byte Format: 1: userBits 8: OplogEntryId RegionId 4:
   * valueLength (optional depending on bits) valueLength: value bytes (optional depending on bits)
   * 4: keyLength keyLength: key bytes 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_MOD_ENTRY_WITH_KEY_8ID = 80;

  /**
   * Written to DRF. The OplogEntryId is relative to the previous del_entry OplogEntryId. The signed
   * difference is encoded in 1 byte. Byte Format: 1: OplogEntryId 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_DEL_ENTRY_1ID = 81;

  /**
   * Written to DRF. The OplogEntryId is relative to the previous del_entry OplogEntryId. The signed
   * difference is encoded in 2 bytes. Byte Format: 2: OplogEntryId 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_DEL_ENTRY_2ID = 82;
  /**
   * Written to DRF. The OplogEntryId is relative to the previous del_entry OplogEntryId. The signed
   * difference is encoded in 3 bytes. Byte Format: 3: OplogEntryId 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_DEL_ENTRY_3ID = 83;

  /**
   * Written to DRF. The OplogEntryId is relative to the previous del_entry OplogEntryId. The signed
   * difference is encoded in 4 bytes. Byte Format: 4: OplogEntryId 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_DEL_ENTRY_4ID = 84;

  /**
   * Written to DRF. The OplogEntryId is relative to the previous del_entry OplogEntryId. The signed
   * difference is encoded in 5 bytes. Byte Format: 5: OplogEntryId 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_DEL_ENTRY_5ID = 85;

  /**
   * Written to DRF. The OplogEntryId is relative to the previous del_entry OplogEntryId. The signed
   * difference is encoded in 6 bytes. Byte Format: 6: OplogEntryId 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_DEL_ENTRY_6ID = 86;

  /**
   * Written to DRF. The OplogEntryId is relative to the previous del_entry OplogEntryId. The signed
   * difference is encoded in 7 bytes. Byte Format: 7: OplogEntryId 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_DEL_ENTRY_7ID = 87;

  /**
   * Written to DRF. The OplogEntryId is relative to the previous del_entry OplogEntryId. The signed
   * difference is encoded in 8 bytes. Byte Format: 8: OplogEntryId 1: EndOfRecord
   *
   * @since GemFire prPersistSprint1
   */
  private static final byte OPLOG_DEL_ENTRY_8ID = 88;

  /**
   * The maximum size of a DEL_ENTRY record in bytes. Currenty this is 10; 1 for opcode and 8 for
   * oplogEntryId and 1 for END_OF_RECORD_ID
   */
  private static final int MAX_DELETE_ENTRY_RECORD_BYTES = 1 + 8 + 1;

  /**
   * Written to beginning of each CRF. Contains the RVV for all regions in the CRF. Byte Format 8:
   * number of regions (variable length encoded number) for each region 4: number of members
   * (variable length encoded number) for each member 4: canonical member id (variable length
   * encoded number) 8: version id (variable length encoded number) 4: number of exceptions
   * (variable length encoded number) variable: exceptions
   */
  private static final byte OPLOG_RVV = 89;

  /**
   * When detected conflict, besides persisting the golden copy by modify(), also persist the
   * conflict operation's region version and member id. and failedWritten to beginning of each CRF.
   * Contains the RVV for all regions in the CRF. Byte Format regionId versions
   */
  private static final byte OPLOG_CONFLICT_VERSION = 90;

  /**
   * persist Gemfire version string into crf, drf, krf Byte Format variable gemfire version string,
   * such as 7.0.0.beta EndOfRecord
   */
  private static final byte OPLOG_GEMFIRE_VERSION = 91;

  static final int OPLOG_GEMFIRE_VERSION_REC_SIZE = 1 + 3 + 1;

  /**
   * Persist oplog file magic number. Written once at the beginning of every oplog file; CRF, DRF,
   * KRF, IF and IRF. 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
   */
  static final byte OPLOG_MAGIC_SEQ_ID = 92;

  public static final int OPLOG_MAGIC_SEQ_REC_SIZE = 1 + OPLOG_TYPE.getLen() + 1;

  /** Compact this oplogs or no. A client configurable property * */
  private final boolean compactOplogs;

  /**
   * This object is used to correctly identify the OpLog size so as to cause a switch of oplogs
   */
  final Object lock = new Object();

  final ByteBuffer[] bbArray = new ByteBuffer[2];

  private boolean lockedForKRFcreate = false;

  /**
   * Set to true when this oplog will no longer be written to. Never set to false once it becomes
   * true.
   */
  private boolean doneAppending = false;

  /**
   * Used to track all information's about live entries that region has in this oplog.
   */
  private final RegionMap regionMap = new RegionMap();

  /**
   * Creates new {@code Oplog} for the given region.
   *
   * @param oplogId int identifying the new oplog
   * @param dirHolder The directory in which to create new Oplog
   *
   * @throws DiskAccessException if the disk files can not be initialized
   */
  Oplog(long oplogId, PersistentOplogSet parent, DirectoryHolder dirHolder) {
    if (oplogId > DiskId.MAX_OPLOG_ID) {
      throw new IllegalStateException(
          "Too many oplogs. The oplog id can not exceed " + DiskId.MAX_OPLOG_ID);
    }
    this.oplogId = oplogId;
    oplogSet = parent;
    this.parent = parent.getParent();
    this.dirHolder = dirHolder;
    // Pretend we have already seen the first record.
    // This will cause a large initial record to force a switch
    // which allows the maxDirSize to be checked.
    firstRecord = false;
    opState = new OpState();
    long maxOplogSizeParam = getParent().getMaxOplogSizeInBytes();
    long availableSpace = this.dirHolder.getAvailableSpace();
    if (availableSpace < maxOplogSizeParam) {
      if (DiskStoreImpl.PREALLOCATE_OPLOGS && !DiskStoreImpl.SET_IGNORE_PREALLOCATE) {
        throw new DiskAccessException(
            String.format("Could not create and pre grow file in dir %s with size=%s",
                this.dirHolder,
                maxOplogSizeParam),
            new IOException("not enough space left to create and pre grow oplog files, available="
                + availableSpace + ", required=" + maxOplogSizeParam),
            getParent());
      }
      maxOplogSize = availableSpace;
      if (logger.isDebugEnabled()) {
        logger.debug(
            "Reducing maxOplogSize to {} because that is all the room remaining in the directory.",
            availableSpace);
      }
    } else {
      maxOplogSize = maxOplogSizeParam;
    }
    setMaxCrfDrfSize();
    stats = getParent().getStats();
    compactOplogs = getParent().getAutoCompact();

    closed = false;
    String n = getParent().getName();
    diskFile = new File(this.dirHolder.getDir(), oplogSet.getPrefix() + n + "_" + oplogId);
    try {
      createDrf(null);
      createCrf(null);
      // open krf for offline compaction
      if (getParent().isOfflineCompacting()) {
        krfFileCreate();
      }
    } catch (Exception ex) {
      close();

      getParent().getCancelCriterion().checkCancelInProgress(ex);
      if (ex instanceof DiskAccessException) {
        throw (DiskAccessException) ex;
      }
      throw new DiskAccessException(
          String.format("Failed creating operation log because: %s", ex),
          getParent());
    }
  }

  /**
   * A copy constructor used for creating a new oplog based on the previous Oplog. This constructor
   * is invoked only from the function switchOplog
   *
   * @param oplogId integer identifying the new oplog
   * @param dirHolder The directory in which to create new Oplog
   * @param prevOplog The previous oplog
   */
  private Oplog(long oplogId, DirectoryHolder dirHolder, Oplog prevOplog) {
    if (oplogId > DiskId.MAX_OPLOG_ID) {
      throw new IllegalStateException(
          "Too many oplogs. The oplog id can not exceed " + DiskId.MAX_OPLOG_ID);
    }
    this.oplogId = oplogId;
    parent = prevOplog.parent;
    oplogSet = prevOplog.oplogSet;
    this.dirHolder = dirHolder;
    opState = new OpState();
    long maxOplogSizeParam = getParent().getMaxOplogSizeInBytes();
    long availableSpace = this.dirHolder.getAvailableSpace();
    if (prevOplog.compactOplogs) {
      maxOplogSize = maxOplogSizeParam;
    } else {
      if (availableSpace < maxOplogSizeParam) {
        maxOplogSize = availableSpace;
        if (logger.isDebugEnabled()) {
          logger.debug(
              "Reducing maxOplogSize to {} because that is all the room remaining in the directory.",
              availableSpace);
        }
      } else {
        maxOplogSize = maxOplogSizeParam;
      }
    }
    setMaxCrfDrfSize();
    stats = prevOplog.stats;
    compactOplogs = prevOplog.compactOplogs;
    // copy over the previous Oplog's data version since data is not being
    // transformed at this point
    dataVersion = prevOplog.getDataVersionIfOld();

    closed = false;
    String n = getParent().getName();
    diskFile = new File(this.dirHolder.getDir(), oplogSet.getPrefix() + n + "_" + oplogId);
    try {
      createDrf(prevOplog.drf);
      createCrf(prevOplog.crf);
      // open krf for offline compaction
      if (getParent().isOfflineCompacting()) {
        krfFileCreate();
      }
    } catch (Exception ex) {
      close();

      getParent().getCancelCriterion().checkCancelInProgress(ex);
      if (ex instanceof DiskAccessException) {
        throw (DiskAccessException) ex;
      }
      throw new DiskAccessException(
          String.format("Failed creating operation log because: %s", ex),
          getParent());
    }
  }

  public Object getLock() {
    return lock;
  }

  public void replaceIncompatibleEntry(DiskRegionView dr, DiskEntry old, DiskEntry repl) {
    boolean useNextOplog = false;
    // No need to get the backup lock prior to synchronizing (correct lock order) since the
    // synchronized block does not attempt to get the backup lock (incorrect lock order)
    synchronized (lock) {
      if (getOplogSet().getChild() != this) {
        // make sure to only call replaceIncompatibleEntry for child, because
        // this.lock
        // can only sync with compaction thread on child oplog
        useNextOplog = true;
      } else {
        // This method is use in recovery only and will not be called by
        // compaction.
        // It's only called before or after compaction. It will replace
        // DiskEntry
        // in DiskRegion without modifying DiskId (such as to a new oplogId),
        // Not to change the entry count in oplog either. While doing that,
        // this.lock will lock the current child to sync with compaction thread.
        // If replace thread got this.lock, DiskEntry "old" will not be removed
        // from
        // current oplog (maybe not child). If compaction thread got this.lock,
        // DiskEntry "old" should have been moved to child oplog when replace
        // thread
        // processes it.

        // See #48032. A new region entry has been put into the region map, but
        // we
        // also have to replace it in the oplog live entries that are used to
        // write
        // the krf. If we don't, we will recover the wrong (old) value.
        getOrCreateDRI(dr).replaceLive(old, repl);
        if (logger.isTraceEnabled(LogMarker.PERSIST_RECOVERY_VERBOSE)) {
          logger.trace(LogMarker.PERSIST_RECOVERY_VERBOSE,
              "replacing incompatible entry key = {} old = {} new = {} oldDiskId = {} new diskId = {} tag = {} in child oplog #{}",
              old.getKey(), System.identityHashCode(old), System.identityHashCode(repl),
              old.getDiskId(), repl.getDiskId(), old.getVersionStamp(), getOplogId());
        }
      }
    }
    if (useNextOplog) {
      if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) {
        CacheObserverHolder.getInstance().afterSwitchingOplog();
      }
      Assert.assertTrue(getOplogSet().getChild() != this);
      getOplogSet().getChild().replaceIncompatibleEntry(dr, old, repl);
    }
  }

  private void writeDiskStoreRecord(OplogFile olf, OPLOG_TYPE type) throws IOException {
    opState = new OpState();
    opState.initialize(type);
    writeOpLogBytes(olf, false, true);
    olf.currSize += getOpStateSize();
    dirHolder.incrementTotalOplogSize(getOpStateSize());
    clearOpState();

    opState.initialize(getParent().getDiskStoreID());
    writeOpLogBytes(olf, false, true); // fix for bug 41928
    olf.currSize += getOpStateSize();
    dirHolder.incrementTotalOplogSize(getOpStateSize());
  }

  private void writeGemfireVersionRecord(OplogFile olf) throws IOException {
    if (gfversion == null) {
      gfversion = KnownVersion.CURRENT;
    }
    KnownVersion dataVersion = getDataVersionIfOld();
    if (dataVersion == null) {
      dataVersion = KnownVersion.CURRENT;
    }
    // if gfversion and dataVersion are not same, then write a special token
    // version and then write both, else write gfversion as before
    // this is for backward compatibility with 7.0
    opState = new OpState();
    if (gfversion == dataVersion) {
      writeProductVersionRecord(gfversion, olf);
    } else {
      writeProductVersionRecord(KnownVersion.TOKEN, olf);
      clearOpState();
      writeProductVersionRecord(gfversion, olf);
      clearOpState();
      writeProductVersionRecord(dataVersion, olf);
    }
  }

  private void writeProductVersionRecord(KnownVersion version, OplogFile olf) throws IOException {
    opState.initialize(version.ordinal());
    writeOpLogBytes(olf, false, true);
    olf.currSize += getOpStateSize();
    dirHolder.incrementTotalOplogSize(getOpStateSize());
  }

  /**
   * Write an RVV record containing all of the live disk regions.
   */
  private void writeRVVRecord(OplogFile olf, boolean writeGCRVV) throws IOException {
    writeRVVRecord(olf, getParent().getAllDiskRegions(), writeGCRVV);
  }

  /**
   * Write the RVV record for the given regions.
   *
   * @param olf the oplog to write to
   * @param diskRegions the set of disk regions we should write the RVV of
   * @param writeGCRVV true to write write the GC RVV
   */
  private void writeRVVRecord(OplogFile olf, Map<Long, AbstractDiskRegion> diskRegions,
      boolean writeGCRVV) throws IOException {
    opState = new OpState();
    opState.initialize(diskRegions, writeGCRVV);
    writeOpLogBytes(olf, false, true); // fix for bug 41928
    olf.currSize += getOpStateSize();
    dirHolder.incrementTotalOplogSize(getOpStateSize());
  }

  private boolean wroteNewEntryBase = false;

  /**
   * Write a OPLOG_NEW_ENTRY_BASE_ID to this oplog. Must be called before any OPLOG_NEW_ENTRY_0ID
   * records are written to this oplog.
   */
  private boolean writeNewEntryBaseRecord(boolean async) throws IOException {
    if (wroteNewEntryBase) {
      return false;
    }
    wroteNewEntryBase = true;
    long newEntryBase = getOplogSet().getOplogEntryId();

    OpState saved = opState;
    try {
      opState = new OpState();
      opState.initialize(newEntryBase);
      writeOpLogBytes(crf, async, false/* no need to flush this record */);
      dirHolder.incrementTotalOplogSize(getOpStateSize());
    } finally {
      opState = saved;
    }
    return true;
  }

  /**
   * Return true if this oplog has a drf but does not have a crf
   */
  boolean isDrfOnly() {
    return drf.f != null && (crf.f == null || !crf.f.exists());
  }

  /**
   * This constructor will get invoked only in case of persistent region when it is recovering an
   * oplog.
   */
  Oplog(long oplogId, PersistentOplogSet parent) {
    // @todo have the crf and drf use different directories.
    if (oplogId > DiskId.MAX_OPLOG_ID) {
      throw new IllegalStateException(
