@Override
  public short undelete(MessageInfo info) throws StoreException {
    checkStarted();
    final Timer.Context context = metrics.undeleteResponse.time();
    // The lifeVersion from message info is -1 when the undelete method is invoked by frontend request, we have to
    // get the legit lifeVersion before we can write undelete record to log segment.
    short revisedLifeVersion = info.getLifeVersion();
    try {
      StoreKey id = info.getStoreKey();
      Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset();
      short lifeVersionFromMessageInfo = info.getLifeVersion();

      List<IndexValue> values =
          index.findAllIndexValuesForKey(id, new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck));
      // Check if the undelete record is valid.
      index.validateSanityForUndelete(id, values, lifeVersionFromMessageInfo);
      IndexValue latestValue = values.get(0);
      IndexValue originalPut = values.get(values.size() - 1);
      if (!IndexValue.hasLifeVersion(revisedLifeVersion)) {
        revisedLifeVersion = (short) (latestValue.getLifeVersion() + 1);
      }
      MessageFormatInputStream stream =
          new UndeleteMessageFormatInputStream(id, info.getAccountId(), info.getContainerId(),
              info.getOperationTimeMs(), revisedLifeVersion);
      // Update info to add stream size;
      info =
          new MessageInfo(id, stream.getSize(), info.getAccountId(), info.getContainerId(), info.getOperationTimeMs(),
              revisedLifeVersion);
      ArrayList<MessageInfo> infoList = new ArrayList<>();
      infoList.add(info);
      MessageFormatWriteSet writeSet = new MessageFormatWriteSet(stream, infoList, false);
      if (!info.getStoreKey().isAccountContainerMatch(latestValue.getAccountId(), latestValue.getContainerId())) {
        if (config.storeValidateAuthorization) {
          throw new StoreException(
              "UNDELETE authorization failure. Key: " + info.getStoreKey() + " Actually accountId: "
                  + latestValue.getAccountId() + "Actually containerId: " + latestValue.getContainerId(),
              StoreErrorCodes.AuthorizationFailure);
        } else {
          logger.warn("UNDELETE authorization failure. Key: {} Actually accountId: {} Actually containerId: {}",
              info.getStoreKey(), latestValue.getAccountId(), latestValue.getContainerId());
          metrics.undeleteAuthorizationFailureCount.inc();
        }
      }
      synchronized (storeWriteLock) {
        Offset currentIndexEndOffset = index.getCurrentEndOffset();
        if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) {
          FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset);
          IndexValue value = index.findKey(info.getStoreKey(), fileSpan,
              EnumSet.of(PersistentIndex.IndexEntryType.DELETE, PersistentIndex.IndexEntryType.UNDELETE));
          if (value != null) {
            if (value.isUndelete() && value.getLifeVersion() == revisedLifeVersion) {
              // Might get an concurrent undelete from both replication and frontend.
              throw new IdUndeletedStoreException(
                  "Can't undelete id " + info.getStoreKey() + " in " + dataDir + " since concurrent operations",
                  value.getLifeVersion());
            } else {
              logger.warn("Revised lifeVersion is " + revisedLifeVersion + " last value is " + value);
              throw new StoreException(
                  "Cannot undelete id " + info.getStoreKey() + " since concurrent operation occurs",
                  StoreErrorCodes.LifeVersionConflict);
            }
          }
        }
        Offset endOffsetOfLastMessage = log.getEndOffset();
        writeSet.writeTo(log);
        logger.trace("Store : {} undelete mark written to log", dataDir);
        FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize());
        // we still use lifeVersion from message info here so that we can re-verify the sanity of undelete request in persistent index.
        IndexValue newUndelete = index.markAsUndeleted(info.getStoreKey(), fileSpan, null, info.getOperationTimeMs(),
            lifeVersionFromMessageInfo);
        blobStoreStats.handleNewUndeleteEntry(info.getStoreKey(), newUndelete, originalPut, latestValue);
      }
      onSuccess("UNDELETE");
      return revisedLifeVersion;
    } catch (StoreException e) {
      if (e.getErrorCode() == StoreErrorCodes.IOError) {
        onError();
      }
      throw e;
    } catch (Exception e) {
      throw new StoreException("Unknown error while trying to undelete blobs from store " + dataDir, e,
          StoreErrorCodes.UnknownError);
    } finally {
      context.stop();
    }
  }
Checks the used capacity of the store against the configured percentage thresholds to see if the store
should be read-only or read-write
ambry-store/src/main/java/com/github/ambry/store/BlobStore.java