@Override
  public UpdateNodeResourceResponse updateNodeResource(UpdateNodeResourceRequest request)
      throws YarnException, IOException {

    // parameter verification.
    if (request == null) {
      routerMetrics.incrUpdateNodeResourceFailedRetrieved();
      RouterServerUtil.logAndThrowException("Missing UpdateNodeResource request.", null);
    }

    String subClusterId = request.getSubClusterId();
    if (StringUtils.isBlank(subClusterId)) {
      routerMetrics.incrUpdateNodeResourceFailedRetrieved();
      RouterServerUtil.logAndThrowException("Missing UpdateNodeResource SubClusterId.", null);
    }

    try {
      long startTime = clock.getTime();
      RMAdminProtocolMethod remoteMethod = new RMAdminProtocolMethod(
          new Class[]{UpdateNodeResourceRequest.class}, new Object[]{request});
      Collection<UpdateNodeResourceResponse> updateNodeResourceResps =
          remoteMethod.invokeConcurrent(this, UpdateNodeResourceResponse.class, subClusterId);
      if (CollectionUtils.isNotEmpty(updateNodeResourceResps)) {
        long stopTime = clock.getTime();
        routerMetrics.succeededUpdateNodeResourceRetrieved(stopTime - startTime);
        return UpdateNodeResourceResponse.newInstance();
      }
    } catch (YarnException e) {
      routerMetrics.incrUpdateNodeResourceFailedRetrieved();
      RouterServerUtil.logAndThrowException(e,
          "Unable to updateNodeResource due to exception. " + e.getMessage());
    }

    routerMetrics.incrUpdateNodeResourceFailedRetrieved();
    throw new YarnException("Unable to updateNodeResource.");
  }
@Test
  public void testUpdateNodeResourceNormalRequest() throws Exception {
    // case 1, test the existing subCluster (SC-1).
    Map<NodeId, ResourceOption> nodeResourceMap = new HashMap<>();
    NodeId nodeId = NodeId.newInstance("127.0.0.1", 1);
    ResourceOption resourceOption =
        ResourceOption.newInstance(Resource.newInstance(2 * GB, 1), -1);
    nodeResourceMap.put(nodeId, resourceOption);
    UpdateNodeResourceRequest request =
        UpdateNodeResourceRequest.newInstance(nodeResourceMap, "SC-1");
    UpdateNodeResourceResponse response = interceptor.updateNodeResource(request);
    assertNotNull(response);

    // case 2, test the non-exist subCluster.
    UpdateNodeResourceRequest request1 =
        UpdateNodeResourceRequest.newInstance(nodeResourceMap, "SC-NON");
    LambdaTestUtils.intercept(Exception.class, "subClusterId = SC-NON is not an active subCluster.",
        () -> interceptor.updateNodeResource(request1));
  }