class OMAllocateBlockRequest extends OMKeyRequest {

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

    OzoneManagerProtocolProtos.AllocateBlockRequest allocateBlockRequest =
        getOmRequest().getAllocateBlockRequest();

    OzoneManagerProtocolProtos.KeyArgs keyArgs =
        allocateBlockRequest.getKeyArgs();

    OzoneManagerProtocolProtos.KeyLocation blockLocation =
        allocateBlockRequest.getKeyLocation();
    Preconditions.checkNotNull(blockLocation);

    String volumeName = keyArgs.getVolumeName();
    String bucketName = keyArgs.getBucketName();
    String keyName = keyArgs.getKeyName();
    long clientID = allocateBlockRequest.getClientID();

    OMMetrics omMetrics = ozoneManager.getMetrics();
    omMetrics.incNumBlockAllocateCalls();

    AuditLogger auditLogger = ozoneManager.getAuditLogger();

    Map<String, String> auditMap = buildKeyArgsAuditMap(keyArgs);
    auditMap.put(OzoneConsts.CLIENT_ID, String.valueOf(clientID));

    OMMetadataManager omMetadataManager = ozoneManager.getMetadataManager();
    String openKeyName = null;

    OMResponse.Builder omResponse = OmResponseUtil.getOMResponseBuilder(
        getOmRequest());
    OMClientResponse omClientResponse = null;

    OmKeyInfo openKeyInfo = null;
    IOException exception = null;
    OmVolumeArgs omVolumeArgs = null;
    OmBucketInfo omBucketInfo = null;

    try {
      keyArgs = resolveBucketLink(ozoneManager, keyArgs, auditMap);
      volumeName = keyArgs.getVolumeName();
      bucketName = keyArgs.getBucketName();

      // check Acl
      checkKeyAclsInOpenKeyTable(ozoneManager, volumeName, bucketName, keyName,
          IAccessAuthorizer.ACLType.WRITE, allocateBlockRequest.getClientID());

      validateBucketAndVolume(omMetadataManager, volumeName,
          bucketName);

      // Here we don't acquire bucket/volume lock because for a single client
      // allocateBlock is called in serial fashion.

      openKeyName = omMetadataManager.getOpenKey(volumeName, bucketName,
          keyName, clientID);
      openKeyInfo = omMetadataManager.getOpenKeyTable().get(openKeyName);
      if (openKeyInfo == null) {
        throw new OMException("Open Key not found " + openKeyName,
            KEY_NOT_FOUND);
      }

      List<OmKeyLocationInfo> newLocationList = Collections.singletonList(
          OmKeyLocationInfo.getFromProtobuf(blockLocation));
      omVolumeArgs = getVolumeInfo(omMetadataManager, volumeName);
      omBucketInfo = getBucketInfo(omMetadataManager, volumeName, bucketName);
      // check volume quota
      long preAllocatedSpace = newLocationList.size()
          * ozoneManager.getScmBlockSize()
          * openKeyInfo.getFactor().getNumber();
      checkVolumeQuotaInBytes(omVolumeArgs, preAllocatedSpace);
      // Append new block
      openKeyInfo.appendNewBlocks(newLocationList, false);

      // Set modification time.
      openKeyInfo.setModificationTime(keyArgs.getModificationTime());

      // Set the UpdateID to current transactionLogIndex
      openKeyInfo.setUpdateID(trxnLogIndex, ozoneManager.isRatisEnabled());

      // Add to cache.
      omMetadataManager.getOpenKeyTable().addCacheEntry(
          new CacheKey<>(openKeyName),
          new CacheValue<>(Optional.of(openKeyInfo), trxnLogIndex));

      // update usedBytes atomically.
      omVolumeArgs.getUsedBytes().add(preAllocatedSpace);
      omBucketInfo.getUsedBytes().add(preAllocatedSpace);

      omResponse.setAllocateBlockResponse(AllocateBlockResponse.newBuilder()
          .setKeyLocation(blockLocation).build());
      omClientResponse = new OMAllocateBlockResponse(omResponse.build(),
          openKeyInfo, clientID, omVolumeArgs, omBucketInfo);

      LOG.debug("Allocated block for Volume:{}, Bucket:{}, OpenKey:{}",
          volumeName, bucketName, openKeyName);
    } catch (IOException ex) {
      omMetrics.incNumBlockAllocateCallFails();
      exception = ex;
      omClientResponse = new OMAllocateBlockResponse(createErrorOMResponse(
          omResponse, exception));
      LOG.error("Allocate Block failed. Volume:{}, Bucket:{}, OpenKey:{}. " +
            "Exception:{}", volumeName, bucketName, openKeyName, exception);
    } finally {
      addResponseToDoubleBuffer(trxnLogIndex, omClientResponse,
          omDoubleBufferHelper);
    }

    auditLog(auditLogger, buildAuditMessage(OMAction.ALLOCATE_BLOCK, auditMap,
        exception, getOmRequest().getUserInfo()));

    return omClientResponse;
  }

  public  OMAllocateBlockRequest(OMRequest omRequest);

}