@Override
  public void updateTtl(List<MessageInfo> infosToUpdate) throws StoreException {
    checkStarted();
    checkDuplicates(infosToUpdate);
    final Timer.Context context = metrics.ttlUpdateResponse.time();
    try {
      List<IndexValue> indexValuesToUpdate = new ArrayList<>();
      List<Short> lifeVersions = new ArrayList<>();
      Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset();
      for (MessageInfo info : infosToUpdate) {
        if (info.getExpirationTimeInMs() != Utils.Infinite_Time) {
          throw new StoreException("BlobStore only supports removing the expiration time",
              StoreErrorCodes.UpdateNotAllowed);
        }
        IndexValue value =
            index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck));
        if (value == null) {
          throw new StoreException("Cannot update TTL of " + info.getStoreKey() + " since it's not in the index",
              StoreErrorCodes.IDNotFound);
        } else if (!info.getStoreKey().isAccountContainerMatch(value.getAccountId(), value.getContainerId())) {
          if (config.storeValidateAuthorization) {
            throw new StoreException(
                "UPDATE authorization failure. Key: " + info.getStoreKey() + " AccountId in store: "
                    + value.getAccountId() + " ContainerId in store: " + value.getContainerId(),
                StoreErrorCodes.AuthorizationFailure);
          } else {
            logger.warn("UPDATE authorization failure. Key: {} AccountId in store: {} ContainerId in store: {}",
                info.getStoreKey(), value.getAccountId(), value.getContainerId());
            metrics.ttlUpdateAuthorizationFailureCount.inc();
          }
        } else if (value.isDelete()) {
          throw new StoreException(
              "Cannot update TTL of " + info.getStoreKey() + " since it is already deleted in the index.",
              StoreErrorCodes.IDDeleted);
        } else if (value.isTtlUpdate()) {
          throw new StoreException("TTL of " + info.getStoreKey() + " is already updated in the index.",
              StoreErrorCodes.AlreadyUpdated);
        } else if (!IndexValue.hasLifeVersion(info.getLifeVersion()) && value.getExpiresAtMs() != Utils.Infinite_Time
            && value.getExpiresAtMs() < info.getOperationTimeMs() + ttlUpdateBufferTimeMs) {
          // When the request is from frontend, make sure it's not too close to expiry date.
          // When the request is from replication, we don't care about the operation time.
          throw new StoreException(
              "TTL of " + info.getStoreKey() + " cannot be updated because it is too close to expiry. Op time (ms): "
                  + info.getOperationTimeMs() + ". ExpiresAtMs: " + value.getExpiresAtMs(),
              StoreErrorCodes.UpdateNotAllowed);
        }
        indexValuesToUpdate.add(value);
        lifeVersions.add(value.getLifeVersion());
      }
      synchronized (storeWriteLock) {
        Offset currentIndexEndOffset = index.getCurrentEndOffset();
        if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) {
          FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset);
          for (MessageInfo info : infosToUpdate) {
            IndexValue value =
                index.findKey(info.getStoreKey(), fileSpan, EnumSet.allOf(PersistentIndex.IndexEntryType.class));
            if (value != null) {
              if (value.isDelete()) {
                throw new StoreException(
                    "Cannot update TTL of " + info.getStoreKey() + " since it is already deleted in the index.",
                    StoreErrorCodes.IDDeleted);
              } else if (value.isTtlUpdate()) {
                throw new StoreException("TTL of " + info.getStoreKey() + " is already updated in the index.",
                    StoreErrorCodes.AlreadyUpdated);
              }
            }
          }
        }
        List<InputStream> inputStreams = new ArrayList<>(infosToUpdate.size());
        List<MessageInfo> updatedInfos = new ArrayList<>(infosToUpdate.size());
        int i = 0;
        for (MessageInfo info : infosToUpdate) {
          MessageFormatInputStream stream =
              new TtlUpdateMessageFormatInputStream(info.getStoreKey(), info.getAccountId(), info.getContainerId(),
                  info.getExpirationTimeInMs(), info.getOperationTimeMs(), lifeVersions.get(i));
          // we only need change the stream size.
          updatedInfos.add(
              new MessageInfo(info.getStoreKey(), stream.getSize(), info.getAccountId(), info.getContainerId(),
                  info.getOperationTimeMs(), info.getLifeVersion()));
          inputStreams.add(stream);
          i++;
        }
        Offset endOffsetOfLastMessage = log.getEndOffset();
        MessageFormatWriteSet writeSet =
            new MessageFormatWriteSet(new SequenceInputStream(Collections.enumeration(inputStreams)), updatedInfos,
                false);
        writeSet.writeTo(log);
        logger.trace("Store : {} ttl update mark written to log", dataDir);
        int correspondingPutIndex = 0;
        for (MessageInfo info : updatedInfos) {
          FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize());
          // Ttl update should aways use the same lifeVersion as it's previous value of the same key, that's why we are
          // using LIFE_VERSION_FROM_FRONTEND here no matter the lifeVersion from the message info.
          IndexValue ttlUpdateValue =
              index.markAsPermanent(info.getStoreKey(), fileSpan, null, info.getOperationTimeMs(),
                  MessageInfo.LIFE_VERSION_FROM_FRONTEND);
          endOffsetOfLastMessage = fileSpan.getEndOffset();
          blobStoreStats.handleNewTtlUpdateEntry(info.getStoreKey(), ttlUpdateValue,
              indexValuesToUpdate.get(correspondingPutIndex++));
        }
        logger.trace("Store : {} ttl update has been marked in the index ", dataDir);
      }
      onSuccess("TTL_UPDATE");
    } catch (StoreException e) {
      if (e.getErrorCode() == StoreErrorCodes.IOError) {
        onError();
      }
      throw e;
    } catch (Exception e) {
      throw new StoreException("Unknown error while trying to update ttl of 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