class Get70 extends BaseCommand {

  @Override
  public void cmdExecute(final Message clientMessage, final ServerConnection serverConnection,
      final SecurityService securityService, long startparam) throws IOException {
    long start = startparam;
    Part regionNamePart = null, keyPart = null, valuePart = null;
    String regionName = null;
    Object callbackArg = null, key = null;
    CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
    CacheServerStats stats = serverConnection.getCacheServerStats();
    String errMessage = null;

    serverConnection.setAsTrue(REQUIRES_RESPONSE);
    // requiresResponse = true;
    {
      long oldStart = start;
      start = DistributionStats.getStatTime();
      stats.incReadGetRequestTime(start - oldStart);
    }
    // Retrieve the data from the message parts
    int parts = clientMessage.getNumberOfParts();
    regionNamePart = clientMessage.getPart(0);
    keyPart = clientMessage.getPart(1);
    // valuePart = null; (redundant assignment)
    if (parts > 2) {
      valuePart = clientMessage.getPart(2);
      try {
        callbackArg = valuePart.getObject();
      } catch (Exception e) {
        writeException(clientMessage, e, false, serverConnection);
        // responded = true;
        serverConnection.setAsTrue(RESPONDED);
        return;
      }
    }
    regionName = regionNamePart.getCachedString();
    try {
      key = keyPart.getStringOrObject();
    } catch (Exception e) {
      writeException(clientMessage, e, false, serverConnection);
      // responded = true;
      serverConnection.setAsTrue(RESPONDED);
      return;
    }
    if (logger.isDebugEnabled()) {
      logger.debug("{}: Received 7.0 get request ({} bytes) from {} for region {} key {} txId {}",
          serverConnection.getName(), clientMessage.getPayloadLength(),
          serverConnection.getSocketString(), regionName, key, clientMessage.getTransactionId());
    }

    // Process the get request
    if (key == null || regionName == null) {
      if ((key == null) && (regionName == null)) {
        errMessage =
            "The input region name and key for the get request are null.";
      } else if (key == null) {
        errMessage = "The input key for the get request is null.";
      } else if (regionName == null) {
        errMessage = "The input region name for the get request is null.";
      }
      logger.warn("{}: {}", serverConnection.getName(), errMessage);
      writeErrorResponse(clientMessage, MessageType.REQUESTDATAERROR, errMessage, serverConnection);
      serverConnection.setAsTrue(RESPONDED);
      return;
    }

    Region region = serverConnection.getCache().getRegion(regionName);
    if (region == null) {
      String reason = String.format("%s was not found during get request", regionName);
      writeRegionDestroyedEx(clientMessage, regionName, reason, serverConnection);
      serverConnection.setAsTrue(RESPONDED);
      return;
    }

    GetOperationContext getContext = null;
    try {
      // for integrated security
      securityService.authorize(Resource.DATA, Operation.READ, regionName, key);

      AuthorizeRequest authzRequest = serverConnection.getAuthzRequest();
      if (authzRequest != null) {
        getContext = authzRequest.getAuthorize(regionName, key, callbackArg);
        callbackArg = getContext.getCallbackArg();
      }
    } catch (NotAuthorizedException ex) {
      writeException(clientMessage, ex, false, serverConnection);
      serverConnection.setAsTrue(RESPONDED);
      return;
    }

    // Get the value and update the statistics. Do not deserialize
    // the value if it is a byte[].
    Entry entry;
    try {
      entry = getEntry(region, key, callbackArg, serverConnection);
    } catch (Exception e) {
      writeException(clientMessage, e, false, serverConnection);
      serverConnection.setAsTrue(RESPONDED);
      return;
    }

    @Retained
    final Object originalData = entry.value;
    Object data = originalData;
    try {
      boolean isObject = entry.isObject;
      VersionTag versionTag = entry.versionTag;
      boolean keyNotPresent = entry.keyNotPresent;

      try {
        AuthorizeRequestPP postAuthzRequest = serverConnection.getPostAuthzRequest();
        if (postAuthzRequest != null) {
          try {
            getContext = postAuthzRequest.getAuthorize(regionName, key, data, isObject, getContext);
            GetOperationContextImpl gci = (GetOperationContextImpl) getContext;
            Object newData = gci.getRawValue();
            if (newData != data) {
              // user changed the value
              isObject = getContext.isObject();
              data = newData;
            }
          } finally {
            if (getContext != null) {
              ((GetOperationContextImpl) getContext).release();
            }
          }
        }
      } catch (NotAuthorizedException ex) {
        writeException(clientMessage, ex, false, serverConnection);
        serverConnection.setAsTrue(RESPONDED);
        return;
      }

      // post process
      data = securityService.postProcess(regionName, key, data, entry.isObject);

      long oldStart = start;
      start = DistributionStats.getStatTime();
      stats.incProcessGetTime(start - oldStart);

      if (region instanceof PartitionedRegion) {
        PartitionedRegion pr = (PartitionedRegion) region;
        if (pr.getNetworkHopType() != PartitionedRegion.NETWORK_HOP_NONE) {
          writeResponseWithRefreshMetadata(data, callbackArg, clientMessage, isObject,
              serverConnection, pr, pr.getNetworkHopType(), versionTag, keyNotPresent);
          pr.clearNetworkHopData();
        } else {
          writeResponse(data, callbackArg, clientMessage, isObject, versionTag, keyNotPresent,
              serverConnection);
        }
      } else {
        writeResponse(data, callbackArg, clientMessage, isObject, versionTag, keyNotPresent,
            serverConnection);
      }
    } finally {
      OffHeapHelper.release(originalData);
    }

    serverConnection.setAsTrue(RESPONDED);
    if (logger.isDebugEnabled()) {
      logger.debug("{}: Wrote get response back to {} for region {} {}", serverConnection.getName(),
          serverConnection.getSocketString(), regionName, entry);
    }
    stats.incWriteGetResponseTime(DistributionStats.getStatTime() - start);

    CachePerfStats regionPerfStats = ((InternalRegion) region).getRegionPerfStats();
    if (regionPerfStats != null) {
      regionPerfStats.endGetForClient(startparam, entry.keyNotPresent);
    }
  }

}