class OMKeysRenameRequest extends OMKeyRequest {

  @Override
  @SuppressWarnings("methodlength")
  public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
      long trxnLogIndex, OzoneManagerDoubleBufferHelper omDoubleBufferHelper) {

    RenameKeysRequest renameKeysRequest = getOmRequest().getRenameKeysRequest();
    RenameKeysArgs renameKeysArgs = renameKeysRequest.getRenameKeysArgs();
    String volumeName = renameKeysArgs.getVolumeName();
    String bucketName = renameKeysArgs.getBucketName();
    OMClientResponse omClientResponse = null;

    List<RenameKeysMap> unRenamedKeys = new ArrayList<>();

    // fromKeyName -> toKeyName
    Map<String, String> renamedKeys = new HashMap<>();

    Map<String, OmKeyInfo> fromKeyAndToKeyInfo = new HashMap<>();
    OMMetrics omMetrics = ozoneManager.getMetrics();
    omMetrics.incNumKeyRenames();

    AuditLogger auditLogger = ozoneManager.getAuditLogger();

    OMResponse.Builder omResponse = OmResponseUtil.getOMResponseBuilder(
        getOmRequest());

    OMMetadataManager omMetadataManager = ozoneManager.getMetadataManager();
    IOException exception = null;
    OmKeyInfo fromKeyValue = null;
    Result result = null;
    Map<String, String> auditMap = new LinkedHashMap<>();
    String fromKeyName = null;
    String toKeyName = null;
    boolean acquiredLock = false;
    boolean renameStatus = true;

    try {
      ResolvedBucket bucket = ozoneManager.resolveBucketLink(
          Pair.of(volumeName, bucketName), this);
      bucket.audit(auditMap);
      volumeName = bucket.realVolume();
      bucketName = bucket.realBucket();
      acquiredLock =
          omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK,
              volumeName, bucketName);

      for (RenameKeysMap renameKey : renameKeysArgs.getRenameKeysMapList()) {

        fromKeyName = renameKey.getFromKeyName();
        toKeyName = renameKey.getToKeyName();
        RenameKeysMap.Builder unRenameKey = RenameKeysMap.newBuilder();

        if (toKeyName.length() == 0 || fromKeyName.length() == 0) {
          renameStatus = false;
          unRenamedKeys.add(
              unRenameKey.setFromKeyName(fromKeyName).setToKeyName(toKeyName)
                  .build());
          LOG.error("Key name is empty fromKeyName {} toKeyName {}",
              fromKeyName, toKeyName);
          continue;
        }

        try {
          // check Acls to see if user has access to perform delete operation
          // on old key and create operation on new key
          checkKeyAcls(ozoneManager, volumeName, bucketName, fromKeyName,
              IAccessAuthorizer.ACLType.DELETE, OzoneObj.ResourceType.KEY);
          checkKeyAcls(ozoneManager, volumeName, bucketName, toKeyName,
              IAccessAuthorizer.ACLType.CREATE, OzoneObj.ResourceType.KEY);
        } catch (Exception ex) {
          renameStatus = false;
          unRenamedKeys.add(
              unRenameKey.setFromKeyName(fromKeyName).setToKeyName(toKeyName)
                  .build());
          LOG.error("Acl check failed for fromKeyName {} toKeyName {}",
              fromKeyName, toKeyName, ex);
          continue;
        }

        // Check if toKey exists
        String fromKey = omMetadataManager.getOzoneKey(volumeName, bucketName,
            fromKeyName);
        String toKey =
            omMetadataManager.getOzoneKey(volumeName, bucketName, toKeyName);
        OmKeyInfo toKeyValue = omMetadataManager.getKeyTable().get(toKey);

        if (toKeyValue != null) {

          renameStatus = false;
          unRenamedKeys.add(
              unRenameKey.setFromKeyName(fromKeyName).setToKeyName(toKeyName)
                  .build());
          LOG.error("Received a request name of new key {} already exists",
              toKeyName);
        }

        // fromKeyName should exist
        fromKeyValue = omMetadataManager.getKeyTable().get(fromKey);
        if (fromKeyValue == null) {
          renameStatus = false;
          unRenamedKeys.add(
              unRenameKey.setFromKeyName(fromKeyName).setToKeyName(toKeyName)
                  .build());
          LOG.error("Received a request to rename a Key does not exist {}",
              fromKey);
          continue;
        }

        fromKeyValue.setUpdateID(trxnLogIndex, ozoneManager.isRatisEnabled());

        fromKeyValue.setKeyName(toKeyName);

        //Set modification time
        fromKeyValue.setModificationTime(Time.now());

        // Add to cache.
        // fromKey should be deleted, toKey should be added with newly updated
        // omKeyInfo.
        Table<String, OmKeyInfo> keyTable = omMetadataManager.getKeyTable();
        keyTable.addCacheEntry(new CacheKey<>(fromKey),
            new CacheValue<>(Optional.absent(), trxnLogIndex));
        keyTable.addCacheEntry(new CacheKey<>(toKey),
            new CacheValue<>(Optional.of(fromKeyValue), trxnLogIndex));
        renamedKeys.put(fromKeyName, toKeyName);
        fromKeyAndToKeyInfo.put(fromKeyName, fromKeyValue);
      }

      OmRenameKeys newOmRenameKeys =
          new OmRenameKeys(volumeName, bucketName, null, fromKeyAndToKeyInfo);
      omClientResponse = new OMKeysRenameResponse(omResponse
          .setRenameKeysResponse(RenameKeysResponse.newBuilder()
              .setStatus(renameStatus)
              .addAllUnRenamedKeys(unRenamedKeys))
          .setStatus(renameStatus ? OK : PARTIAL_RENAME)
          .setSuccess(renameStatus).build(),
          newOmRenameKeys);

      result = Result.SUCCESS;
    } catch (IOException ex) {
      result = Result.FAILURE;
      exception = ex;
      createErrorOMResponse(omResponse, ex);

      omResponse.setRenameKeysResponse(RenameKeysResponse.newBuilder()
          .setStatus(renameStatus).addAllUnRenamedKeys(unRenamedKeys).build());
      omClientResponse = new OMKeysRenameResponse(omResponse.build());

    } finally {
      addResponseToDoubleBuffer(trxnLogIndex, omClientResponse,
          omDoubleBufferHelper);
      if (acquiredLock) {
        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
            bucketName);
      }
    }

    auditMap = buildAuditMap(auditMap, renamedKeys, unRenamedKeys);
    auditLog(auditLogger, buildAuditMessage(OMAction.RENAME_KEYS, auditMap,
        exception, getOmRequest().getUserInfo()));

    switch (result) {
    case SUCCESS:
      LOG.debug("Rename Keys is successfully completed for auditMap:{}.",
          auditMap);
      break;
    case FAILURE:
      ozoneManager.getMetrics().incNumKeyRenameFails();
      LOG.error("Rename keys failed for auditMap:{}.", auditMap);
      break;
    default:
      LOG.error("Unrecognized Result for OMKeysRenameRequest: {}",
          renameKeysRequest);
    }

    return omClientResponse;
  }

  public  OMKeysRenameRequest(OMRequest omRequest);

}