public static void main(String[] args) throws Exception {
    VerifiableProperties verifiableProperties = ToolUtils.getVerifiableProperties(args);
    ServerAdminToolConfig config = new ServerAdminToolConfig(verifiableProperties);
    ClusterMapConfig clusterMapConfig = new ClusterMapConfig(verifiableProperties);
    ClusterMap clusterMap =
        ((ClusterAgentsFactory) Utils.getObj(clusterMapConfig.clusterMapClusterAgentsFactory, clusterMapConfig,
            config.hardwareLayoutFilePath, config.partitionLayoutFilePath)).getClusterMap();
    SSLFactory sslFactory = new NettySslHttp2Factory(new SSLConfig(verifiableProperties));
    ServerAdminTool serverAdminTool = new ServerAdminTool(clusterMap, sslFactory, verifiableProperties);
    File file = new File(config.dataOutputFilePath);
    if (!file.exists() && !file.createNewFile()) {
      throw new IllegalStateException("Could not create " + file);
    }
    FileOutputStream outputFileStream = new FileOutputStream(config.dataOutputFilePath);
    DataNodeId dataNodeId = clusterMap.getDataNodeId(config.hostname, config.port);
    if (dataNodeId == null) {
      throw new IllegalArgumentException(
          "Could not find a data node corresponding to " + config.hostname + ":" + config.port);
    }
    switch (config.typeOfOperation) {
      case GetBlobProperties:
        BlobId blobId = new BlobId(config.blobId, clusterMap);
        Pair<ServerErrorCode, BlobProperties> bpResponse =
            serverAdminTool.getBlobProperties(dataNodeId, blobId, config.getOption, clusterMap);
        if (bpResponse.getFirst() == ServerErrorCode.NoError) {
          LOGGER.info("Blob properties for {} from {}: {}", blobId, dataNodeId, bpResponse.getSecond());
        } else {
          LOGGER.error("Failed to get blob properties for {} from {} with option {}. Error code is {}", blobId,
              dataNodeId, config.getOption, bpResponse.getFirst());
        }
        break;
      case GetUserMetadata:
        blobId = new BlobId(config.blobId, clusterMap);
        Pair<ServerErrorCode, ByteBuffer> umResponse =
            serverAdminTool.getUserMetadata(dataNodeId, blobId, config.getOption, clusterMap);
        if (umResponse.getFirst() == ServerErrorCode.NoError) {
          writeBufferToFile(umResponse.getSecond(), outputFileStream);
          LOGGER.info("User metadata for {} from {} written to {}", blobId, dataNodeId, config.dataOutputFilePath);
        } else {
          LOGGER.error("Failed to get user metadata for {} from {} with option {}. Error code is {}", blobId,
              dataNodeId, config.getOption, umResponse.getFirst());
        }
        break;
      case GetBlob:
        blobId = new BlobId(config.blobId, clusterMap);
        Pair<ServerErrorCode, BlobData> bResponse =
            serverAdminTool.getBlob(dataNodeId, blobId, config.getOption, clusterMap);
        if (bResponse.getFirst() == ServerErrorCode.NoError) {
          LOGGER.info("Blob type of {} from {} is {}", blobId, dataNodeId, bResponse.getSecond().getBlobType());
          ByteBuf buffer = bResponse.getSecond().content();
          try {
            writeByteBufToFile(buffer, outputFileStream);
          } finally {
            buffer.release();
          }
          LOGGER.info("Blob data for {} from {} written to {}", blobId, dataNodeId, config.dataOutputFilePath);
        } else {
          LOGGER.error("Failed to get blob data for {} from {} with option {}. Error code is {}", blobId, dataNodeId,
              config.getOption, bResponse.getFirst());
        }
        break;
      case BlobIndex:
        blobId = new BlobId(config.blobId, clusterMap);
        Pair<ServerErrorCode, String> biResponse = serverAdminTool.getBlobIndex(dataNodeId, blobId);
        if (biResponse.getFirst() == ServerErrorCode.NoError) {
          LOGGER.info("Blob index values of {} from {} is {}", blobId, dataNodeId, biResponse.getSecond());
        } else {
          LOGGER.error("Failed to get blob index values for {} from {}. Error code is {}", blobId, dataNodeId,
              biResponse.getFirst());
        }
        break;
      case ForceDelete:
        blobId = new BlobId(config.blobId, clusterMap);
        short lifeVersion = config.lifeVersion;
        ServerErrorCode errorCode = serverAdminTool.forceDeleteBlob(dataNodeId, blobId, lifeVersion);
        if (errorCode == ServerErrorCode.NoError) {
          LOGGER.info("Force delete {} {} from {} is successful.", blobId, lifeVersion, dataNodeId);
        } else {
          LOGGER.error("Failed to run force delete {} {} from {}. Error code is {}", blobId, lifeVersion, dataNodeId,
              errorCode);
        }
        break;
      case TriggerCompaction:
        if (config.partitionIds.length > 0 && !config.partitionIds[0].isEmpty()) {
          for (String partitionIdStr : config.partitionIds) {
            PartitionId partitionId = getPartitionIdFromStr(partitionIdStr, clusterMap);
            errorCode = serverAdminTool.triggerCompaction(dataNodeId, partitionId);
            if (errorCode == ServerErrorCode.NoError) {
              LOGGER.info("Compaction has been triggered for {} on {}", partitionId, dataNodeId);
            } else {
              LOGGER.error("From {}, received server error code {} for trigger compaction request on {}", dataNodeId,
                  errorCode, partitionId);
            }
          }
        } else {
          LOGGER.error("There were no partitions provided to trigger compaction on");
        }
        break;
      case RequestControl:
        if (config.partitionIds.length > 0 && !config.partitionIds[0].isEmpty()) {
          for (String partitionIdStr : config.partitionIds) {
            PartitionId partitionId = getPartitionIdFromStr(partitionIdStr, clusterMap);
            sendRequestControlRequest(serverAdminTool, dataNodeId, partitionId, config.requestTypeToControl,
                config.enableState);
          }
        } else {
          LOGGER.info("No partition list provided. Requesting enable status of {} to be set to {} on all partitions",
              config.requestTypeToControl, config.enableState);
          sendRequestControlRequest(serverAdminTool, dataNodeId, null, config.requestTypeToControl, config.enableState);
        }
        break;
      case ReplicationControl:
        List<String> origins = Collections.emptyList();
        if (config.origins.length > 0 && !config.origins[0].isEmpty()) {
          origins = Arrays.asList(config.origins);
        }
        if (config.partitionIds.length > 0 && !config.partitionIds[0].isEmpty()) {
          for (String partitionIdStr : config.partitionIds) {
            PartitionId partitionId = getPartitionIdFromStr(partitionIdStr, clusterMap);
            sendReplicationControlRequest(serverAdminTool, dataNodeId, partitionId, origins, config.enableState);
          }
        } else {
          LOGGER.info("No partition list provided. Requesting enable status for replication from {} to be set to {} on "
              + "all partitions", origins.isEmpty() ? "all DCs" : origins, config.enableState);
          sendReplicationControlRequest(serverAdminTool, dataNodeId, null, origins, config.enableState);
        }
        break;
      case CatchupStatus:
        if (config.partitionIds.length > 0 && !config.partitionIds[0].isEmpty()) {
          for (String partitionIdStr : config.partitionIds) {
            PartitionId partitionId = getPartitionIdFromStr(partitionIdStr, clusterMap);
            Pair<ServerErrorCode, Boolean> response =
                serverAdminTool.isCaughtUp(dataNodeId, partitionId, config.acceptableLagInBytes,
                    config.numReplicasCaughtUpPerPartition);
            if (response.getFirst() == ServerErrorCode.NoError) {
              LOGGER.info("Replicas are {} within {} bytes for {}", response.getSecond() ? "" : "NOT",
                  config.acceptableLagInBytes, partitionId);
            } else {
              LOGGER.error("From {}, received server error code {} for request for catchup status of {}", dataNodeId,
                  response.getFirst(), partitionId);
            }
          }
        } else {
          Pair<ServerErrorCode, Boolean> response =
              serverAdminTool.isCaughtUp(dataNodeId, null, config.acceptableLagInBytes,
                  config.numReplicasCaughtUpPerPartition);
          if (response.getFirst() == ServerErrorCode.NoError) {
            LOGGER.info("Replicas are {} within {} bytes for all partitions", response.getSecond() ? "" : "NOT",
                config.acceptableLagInBytes);
          } else {
            LOGGER.error("From {}, received server error code {} for request for catchup status of all partitions",
                dataNodeId, response.getFirst());
          }
        }
        break;
      case BlobStoreControl:
        if (config.partitionIds.length > 0 && !config.partitionIds[0].isEmpty()) {
          for (String partitionIdStr : config.partitionIds) {
            PartitionId partitionId = getPartitionIdFromStr(partitionIdStr, clusterMap);
            sendBlobStoreControlRequest(serverAdminTool, dataNodeId, partitionId,
                config.numReplicasCaughtUpPerPartition, config.storeControlRequestType);
          }
        } else {
          LOGGER.error("There were no partitions provided to be controlled (Start/Stop)");
        }
        break;
      default:
        throw new IllegalStateException("Recognized but unsupported operation: " + config.typeOfOperation);
    }
    serverAdminTool.close();
    outputFileStream.close();
    clusterMap.close();
    System.out.println("Server admin tool is safely closed");
    System.exit(0);
  }
Runs the server admin tool
@param args associated arguments.
@throws Exception
ambry-tools/src/main/java/com/github/ambry/tools/admin/ServerAdminTool.java