private AdminResponse handleRequestControlRequest(DataInputStream requestStream, AdminRequest adminRequest)
      throws IOException {
    RequestControlAdminRequest controlRequest = RequestControlAdminRequest.readFrom(requestStream, adminRequest);
    RequestOrResponseType toControl = controlRequest.getRequestTypeToControl();
    ServerErrorCode error;
    Collection<PartitionId> partitionIds;
    if (!requestsDisableInfo.containsKey(toControl)) {
      metrics.badRequestError.inc();
      error = ServerErrorCode.BadRequest;
    } else {
      error = ServerErrorCode.NoError;
      if (controlRequest.getPartitionId() != null) {
        error = validateRequest(controlRequest.getPartitionId(), RequestOrResponseType.AdminRequest, false);
        partitionIds = Collections.singletonList(controlRequest.getPartitionId());
      } else {
        partitionIds = storeManager.getLocalPartitions();
      }
      if (!error.equals(ServerErrorCode.PartitionUnknown)) {
        controlRequestForPartitions(EnumSet.of(toControl), partitionIds, controlRequest.shouldEnable());
        for (PartitionId partitionId : partitionIds) {
          logger.info("Enable state for {} on {} is {}", toControl, partitionId,
              isRequestEnabled(toControl, partitionId));
        }
      }
    }
    return new AdminResponse(adminRequest.getCorrelationId(), adminRequest.getClientId(), error);
  }
Handles {@link com.github.ambry.protocol.AdminRequestOrResponseType#RequestControl}.
@param requestStream the serialized bytes of the request.
@param adminRequest the {@link AdminRequest} received.
@return the {@link AdminResponse} to the request.
@throws IOException if there is any I/O error reading from the {@code requestStream}.
ambry-server/src/main/java/com/github/ambry/server/AmbryServerRequests.java