@Override
  public GetDelegationTokenResponse getDelegationToken(
      GetDelegationTokenRequest request) throws YarnException, IOException {

    if (request == null || request.getRenewer() == null) {
      routerMetrics.incrGetDelegationTokenFailedRetrieved();
      String msg = "Missing getDelegationToken request or Renewer.";
      RouterAuditLogger.logFailure(user.getShortUserName(), GET_DELEGATIONTOKEN, UNKNOWN,
          TARGET_CLIENT_RM_SERVICE, msg);
      RouterServerUtil.logAndThrowException(msg, null);
    }

    try {
      // Verify that the connection is kerberos authenticated
      if (!RouterServerUtil.isAllowedDelegationTokenOp()) {
        routerMetrics.incrGetDelegationTokenFailedRetrieved();
        String msg = "Delegation Token can be issued only with kerberos authentication.";
        RouterAuditLogger.logFailure(user.getShortUserName(), GET_DELEGATIONTOKEN, UNKNOWN,
            TARGET_CLIENT_RM_SERVICE, msg);
        throw new IOException(msg);
      }

      long startTime = clock.getTime();
      UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
      Text owner = new Text(ugi.getUserName());
      Text realUser = null;
      if (ugi.getRealUser() != null) {
        realUser = new Text(ugi.getRealUser().getUserName());
      }

      RMDelegationTokenIdentifier tokenIdentifier =
          new RMDelegationTokenIdentifier(owner, new Text(request.getRenewer()), realUser);
      Token<RMDelegationTokenIdentifier> realRMDToken =
          new Token<>(tokenIdentifier, this.getTokenSecretManager());

      org.apache.hadoop.yarn.api.records.Token routerRMDTToken =
          BuilderUtils.newDelegationToken(realRMDToken.getIdentifier(),
              realRMDToken.getKind().toString(),
              realRMDToken.getPassword(), realRMDToken.getService().toString());

      long stopTime = clock.getTime();
      routerMetrics.succeededGetDelegationTokenRetrieved((stopTime - startTime));
      RouterAuditLogger.logSuccess(user.getShortUserName(), GET_DELEGATIONTOKEN,
          TARGET_CLIENT_RM_SERVICE);
      return GetDelegationTokenResponse.newInstance(routerRMDTToken);
    } catch(IOException e) {
      routerMetrics.incrGetDelegationTokenFailedRetrieved();
      RouterAuditLogger.logFailure(user.getShortUserName(), GET_DELEGATIONTOKEN, UNKNOWN,
          TARGET_CLIENT_RM_SERVICE, "getDelegationToken error, errMsg = " + e.getMessage());
      throw new YarnException(e);
    }
  }
@Test
  public void testGetDelegationToken() throws IOException, YarnException {

    // We design such a unit test to check
    // that the execution of the GetDelegationToken method is as expected.
    //
    // 1. Apply for a DelegationToken for renewer1,
    // the Router returns the DelegationToken of the user, and the KIND of the token is
    // RM_DELEGATION_TOKEN
    //
    // 2. We maintain the compatibility with RMDelegationTokenIdentifier,
    // we can serialize the token into RMDelegationTokenIdentifier.
    //
    // 3. We can get the issueDate, and compare the data in the StateStore,
    // the data should be consistent.

    // Step1. We apply for DelegationToken for renewer1
    // Both response & delegationToken cannot be empty
    GetDelegationTokenRequest request = mock(GetDelegationTokenRequest.class);
    when(request.getRenewer()).thenReturn("renewer1");
    GetDelegationTokenResponse response = interceptor.getDelegationToken(request);
    Assert.assertNotNull(response);
    Token delegationToken = response.getRMDelegationToken();
    Assert.assertNotNull(delegationToken);
    Assert.assertEquals("RM_DELEGATION_TOKEN", delegationToken.getKind());

    // Step2. Serialize the returned Token as RMDelegationTokenIdentifier.
    org.apache.hadoop.security.token.Token<RMDelegationTokenIdentifier> token =
        ConverterUtils.convertFromYarn(delegationToken, (Text) null);
    RMDelegationTokenIdentifier rMDelegationTokenIdentifier = token.decodeIdentifier();
    Assert.assertNotNull(rMDelegationTokenIdentifier);

    // Step3. Verify the returned data of the token.
    String renewer = rMDelegationTokenIdentifier.getRenewer().toString();
    long issueDate = rMDelegationTokenIdentifier.getIssueDate();
    long maxDate = rMDelegationTokenIdentifier.getMaxDate();
    Assert.assertEquals("renewer1", renewer);

    long tokenMaxLifetime = this.getConf().getLong(
        YarnConfiguration.RM_DELEGATION_TOKEN_MAX_LIFETIME_KEY,
        YarnConfiguration.RM_DELEGATION_TOKEN_MAX_LIFETIME_DEFAULT);
    Assert.assertEquals(issueDate + tokenMaxLifetime, maxDate);

    RouterRMDTSecretManagerState managerState = stateStore.getRouterRMSecretManagerState();
    Assert.assertNotNull(managerState);

    Map<RMDelegationTokenIdentifier, RouterStoreToken> delegationTokenState =
        managerState.getTokenState();
    Assert.assertNotNull(delegationTokenState);
    Assert.assertTrue(delegationTokenState.containsKey(rMDelegationTokenIdentifier));

    long tokenRenewInterval = this.getConf().getLong(
        YarnConfiguration.RM_DELEGATION_TOKEN_RENEW_INTERVAL_KEY,
        YarnConfiguration.RM_DELEGATION_TOKEN_RENEW_INTERVAL_DEFAULT);
    RouterStoreToken resultRouterStoreToken = delegationTokenState.get(rMDelegationTokenIdentifier);
    Assert.assertNotNull(resultRouterStoreToken);
    long renewDate = resultRouterStoreToken.getRenewDate();
    Assert.assertEquals(issueDate + tokenRenewInterval, renewDate);
  }