private void readModifyEntryWithKey(CountingDataInputStream dis, byte opcode,
      OplogEntryIdSet deletedIds, boolean recoverValue,
      KnownVersion version, ByteArrayDataInput in) throws IOException {
    long oplogOffset = -1;

    byte userBits = dis.readByte();

    int idByteCount = (opcode - OPLOG_MOD_ENTRY_WITH_KEY_1ID) + 1;
    long oplogKeyId = getModEntryId(dis, idByteCount);
    long drId = DiskInitFile.readDiskRegionID(dis);
    DiskRecoveryStore drs = getOplogSet().getCurrentlyRecovering(drId);

    // read version
    VersionTag tag = null;
    if (EntryBits.isWithVersions(userBits)) {
      tag = readVersionsFromOplog(dis);
    } else if (getParent().isUpgradeVersionOnly() && drs != null) {
      tag = createDummyTag(drs);
      userBits = EntryBits.setWithVersions(userBits, true);
    }
    if (drs != null
        && !drs.getDiskRegionView().getFlags().contains(DiskRegionFlag.IS_WITH_VERSIONING)) {
      // 50044 Remove version tag from entry if we don't want versioning for
      // this region
      tag = null;
      userBits = EntryBits.setWithVersions(userBits, false);
    }
    OkToSkipResult skipResult = okToSkipModifyRecord(deletedIds, drId, drs, oplogKeyId, true, tag);
    if (skipResult.skip()) {
      if (!isPhase2()) {
        incSkipped();
        stats.incRecoveryRecordsSkipped();
      }
    } else if (recoverValue && !getParent().isOfflineCompacting()) {
      recoverValue = recoverLruValue(drs);
    }

    byte[] objValue = null;
    int valueLength = 0;
    CompactionRecord p2cr = null;
    long crOffset;
    if (EntryBits.isAnyInvalid(userBits) || EntryBits.isTombstone(userBits)) {
      if (EntryBits.isInvalid(userBits)) {
        objValue = DiskEntry.INVALID_BYTES;
      } else if (EntryBits.isTombstone(userBits)) {
        objValue = DiskEntry.TOMBSTONE_BYTES;
      } else {
        objValue = DiskEntry.LOCAL_INVALID_BYTES;
      }
      crOffset = dis.getCount();
      if (!skipResult.skip()) {
        if (isPhase2()) {
          p2cr = (CompactionRecord) getRecoveryMap().get(oplogKeyId);
          if (p2cr != null && p2cr.getOffset() != crOffset) {
            skipResult = OkToSkipResult.SKIP_RECORD;
          }
        }
      }
    } else {
      int len = dis.readInt();
      oplogOffset = dis.getCount();
      crOffset = oplogOffset;
      valueLength = len;
      if (!skipResult.skip()) {
        if (isPhase2()) {
          p2cr = (CompactionRecord) getRecoveryMap().get(oplogKeyId);
          if (p2cr != null && p2cr.getOffset() != crOffset) {
            skipResult = OkToSkipResult.SKIP_RECORD;
          }
        }
      }
      if (!skipResult.skip() && recoverValue) {
        byte[] valueBytes = new byte[len];
        dis.readFully(valueBytes);
        objValue = valueBytes;
        validateValue(valueBytes, userBits, version, in);
      } else {
        forceSkipBytes(dis, len);
      }
    }

    int keyLen = dis.readInt();

    incTotalCount();
    if (skipResult.skip()) {
      if (skipResult.skipKey()) {
        forceSkipBytes(dis, keyLen);
      } else {
        byte[] keyBytes = new byte[keyLen];
        dis.readFully(keyBytes);
        skippedKeyBytes.put(oplogKeyId, keyBytes);
      }
      readEndOfRecord(dis);
      if (logger.isTraceEnabled(LogMarker.PERSIST_RECOVERY_VERBOSE)) {
        logger.trace(LogMarker.PERSIST_RECOVERY_VERBOSE,
            "skipping readModEntryWK init oplogKeyId=<{}> drId={}", oplogKeyId, drId);
      }
    } else {
      // read the key
      byte[] keyBytes = null;
      if (isPhase2()) {
        forceSkipBytes(dis, keyLen);
      } else {
        keyBytes = new byte[keyLen];
        dis.readFully(keyBytes);
      }
      readEndOfRecord(dis);
      if (drs != null && tag != null) {
        // Update the RVV with the new entry
        // This must be done after reading the end of record to make sure
        // we don't have a corrupt record. See bug #45538
        drs.recordRecoveredVersionTag(tag);
      }
      assert oplogKeyId >= 0;
      if (getParent().isOfflineCompacting()) {
        if (isPhase1()) {
          CompactionRecord cr = new CompactionRecord(keyBytes, crOffset);
          getRecoveryMap().put(oplogKeyId, cr);
          if (drs != null) {
            drs.getDiskRegionView().incRecoveredEntryCount();
          }
          stats.incRecoveredEntryCreates();
        } else { // phase2
          Assert.assertNotNull(p2cr, "First pass did not find create a compaction record");
          getOplogSet().getChild().copyForwardForOfflineCompact(oplogKeyId, p2cr.getKeyBytes(),
              objValue, userBits, drId, tag);
          if (logger.isTraceEnabled(LogMarker.PERSIST_RECOVERY_VERBOSE)) {
            logger.trace(LogMarker.PERSIST_RECOVERY_VERBOSE,
                "readModifyEntryWithKey copyForward oplogKeyId=<{}>", oplogKeyId);
          }
          // add it to the deletedIds set so we will ignore it in earlier oplogs
          deletedIds.add(oplogKeyId);
        }
      } else {
        Object key = deserializeKey(keyBytes, version, in);
        Object oldValue = getRecoveryMap().put(oplogKeyId, key);
        if (oldValue != null) {
          throw new AssertionError(
              String.format(
                  "Oplog::readNewEntry: Create is present in more than one Oplog. This should not be possible. The Oplog Key ID for this entry is %s.",
                  oplogKeyId));
        }
        // Check the actual region to see if it has this key from
        // a previous recovered oplog.
        if (drs != null) {
          DiskEntry de = drs.getDiskEntry(key);
          if (de == null) {
            DiskRegionView drv = drs.getDiskRegionView();
            // and create an entry
            DiskEntry.RecoveredEntry re = createRecoveredEntry(objValue, valueLength, userBits,
                getOplogId(), oplogOffset, oplogKeyId, recoverValue, version, in);
            if (tag != null) {
              re.setVersionTag(tag);
            }
            if (logger.isTraceEnabled(LogMarker.PERSIST_RECOVERY_VERBOSE)) {
              logger.trace(LogMarker.PERSIST_RECOVERY_VERBOSE,
                  "readModEntryWK init oplogKeyId=<{}> drId={} key={} oplogOffset={} userBits={} valueLen={} tag={}",
                  oplogKeyId, drId, key, oplogOffset, userBits, valueLength, tag);
            }
            initRecoveredEntry(drv, drs.initializeRecoveredEntry(key, re));
            drs.getDiskRegionView().incRecoveredEntryCount();
            stats.incRecoveredEntryCreates();

          } else {
            DiskId curdid = de.getDiskId();
            assert curdid
                .getOplogId() != getOplogId() : "Mutiple ModEntryWK in the same oplog for getOplogId()="
                    + getOplogId() + " , curdid.getOplogId()=" + curdid.getOplogId()
                    + " , for drId="
                    + drId + " , key=" + key;
            if (logger.isTraceEnabled(LogMarker.PERSIST_RECOVERY_VERBOSE)) {
              logger.trace(LogMarker.PERSIST_RECOVERY_VERBOSE,
                  "ignore readModEntryWK because getOplogId()={} != curdid.getOplogId()={} for drId={} key={}",
                  getOplogId(), curdid.getOplogId(), drId, key);
            }
          }
        }
      }
    }
  }
Reads an oplog entry of type ModifyWithKey

@param dis DataInputStream from which the oplog is being read
@param opcode byte whether the id is short/int/long
geode-core/src/main/java/org/apache/geode/internal/cache/Oplog.java