import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public Token<TimelineDelegationTokenIdentifier> getDelegationToken(
    final String renewer) throws IOException, YarnException {
  PrivilegedExceptionAction<Token<TimelineDelegationTokenIdentifier>> getDTAction =
      new PrivilegedExceptionAction<Token<TimelineDelegationTokenIdentifier>>() {

        @Override
        public Token<TimelineDelegationTokenIdentifier> run()
            throws Exception {
          DelegationTokenAuthenticatedURL authUrl =
              new DelegationTokenAuthenticatedURL(authenticator,
                  connConfigurator);
          return (Token) authUrl.getDelegationToken(
              resURI.toURL(), token, renewer, doAsUser);
        }
      };
  return (Token<TimelineDelegationTokenIdentifier>) operateDelegationToken(getDTAction);
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public void cancelDelegationToken(
    final Token<TimelineDelegationTokenIdentifier> timelineDT)
        throws IOException, YarnException {
  final boolean isTokenServiceAddrEmpty =
      timelineDT.getService().toString().isEmpty();
  final String scheme = isTokenServiceAddrEmpty ? null
      : (YarnConfiguration.useHttps(this.getConfig()) ? "https" : "http");
  final InetSocketAddress address = isTokenServiceAddrEmpty ? null
      : SecurityUtil.getTokenServiceAddr(timelineDT);
  PrivilegedExceptionAction<Void> cancelDTAction =
      new PrivilegedExceptionAction<Void>() {

        @Override
        public Void run() throws Exception {
          // If the timeline DT to cancel is different than cached, replace it.
          // Token to set every time for retry, because when exception happens,
          // DelegationTokenAuthenticatedURL will reset it to null;
          if (!timelineDT.equals(token.getDelegationToken())) {
            token.setDelegationToken((Token) timelineDT);
          }
          DelegationTokenAuthenticatedURL authUrl =
              new DelegationTokenAuthenticatedURL(authenticator,
                  connConfigurator);
          // If the token service address is not available, fall back to use
          // the configured service address.
          final URI serviceURI = isTokenServiceAddrEmpty ? resURI
              : new URI(scheme, null, address.getHostName(),
              address.getPort(), RESOURCE_URI_STR, null, null);
          authUrl.cancelDelegationToken(serviceURI.toURL(), token, doAsUser);
          return null;
        }
      };
  operateDelegationToken(cancelDTAction);
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@Override
public void storeToken(TimelineDelegationTokenIdentifier tokenId,
    Long renewDate) throws IOException {
  DataOutputStream ds = null;
  WriteBatch batch = null;
  try {
    byte[] k = createTokenEntryKey(tokenId.getSequenceNumber());
    if (db.get(k) != null) {
      throw new IOException(tokenId + " already exists");
    }
    byte[] v = buildTokenData(tokenId, renewDate);
    ByteArrayOutputStream bs = new ByteArrayOutputStream();
    ds = new DataOutputStream(bs);
    ds.writeInt(tokenId.getSequenceNumber());
    batch = db.createWriteBatch();
    batch.put(k, v);
    batch.put(LATEST_SEQUENCE_NUMBER_KEY, bs.toByteArray());
    db.write(batch);
  } catch (DBException e) {
    throw new IOException(e);
  } finally {
    IOUtils.cleanup(LOG, ds);
    IOUtils.cleanup(LOG, batch);
  }
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public Token<TimelineDelegationTokenIdentifier> getDelegationToken(
    final String renewer) throws IOException, YarnException {
  boolean isProxyAccess =
      UserGroupInformation.getCurrentUser().getAuthenticationMethod()
      == UserGroupInformation.AuthenticationMethod.PROXY;
  final String doAsUser = isProxyAccess ?
      UserGroupInformation.getCurrentUser().getShortUserName() : null;
  PrivilegedExceptionAction<Token<TimelineDelegationTokenIdentifier>> getDTAction =
      new PrivilegedExceptionAction<Token<TimelineDelegationTokenIdentifier>>() {

        @Override
        public Token<TimelineDelegationTokenIdentifier> run()
            throws Exception {
          DelegationTokenAuthenticatedURL authUrl =
              new DelegationTokenAuthenticatedURL(authenticator,
                  connConfigurator);
          return (Token) authUrl.getDelegationToken(
              resURI.toURL(), token, renewer, doAsUser);
        }
      };
  return (Token<TimelineDelegationTokenIdentifier>) operateDelegationToken(getDTAction);
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public long renewDelegationToken(
    final Token<TimelineDelegationTokenIdentifier> timelineDT)
        throws IOException, YarnException {
  final boolean isTokenServiceAddrEmpty =
      timelineDT.getService().toString().isEmpty();
  final String scheme = isTokenServiceAddrEmpty ? null
      : (YarnConfiguration.useHttps(this.getConfig()) ? "https" : "http");
  final InetSocketAddress address = isTokenServiceAddrEmpty ? null
      : SecurityUtil.getTokenServiceAddr(timelineDT);
  PrivilegedExceptionAction<Long> renewDTAction =
      new PrivilegedExceptionAction<Long>() {

        @Override
        public Long run() throws Exception {
          // If the timeline DT to renew is different than cached, replace it.
          // Token to set every time for retry, because when exception happens,
          // DelegationTokenAuthenticatedURL will reset it to null;
          if (!timelineDT.equals(token.getDelegationToken())) {
            token.setDelegationToken((Token) timelineDT);
          }
          DelegationTokenAuthenticatedURL authUrl =
              new DelegationTokenAuthenticatedURL(authenticator,
                  connConfigurator);
          // If the token service address is not available, fall back to use
          // the configured service address.
          final URI serviceURI = isTokenServiceAddrEmpty ? resURI
              : new URI(scheme, null, address.getHostName(),
              address.getPort(), RESOURCE_URI_STR, null, null);
          return authUrl
              .renewDelegationToken(serviceURI.toURL(), token, doAsUser);
        }
      };
  return (Long) operateDelegationToken(renewDTAction);
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@Test
public void testParseTimelineDelegationTokenIdentifierRenewer() throws IOException {
  // Server side when generation a timeline DT
  Configuration conf = new YarnConfiguration();
  conf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTH_TO_LOCAL,
      "RULE:[2:[email protected]$0]([nr][email protected]*EXAMPLE.COM)s/.*/yarn/");
  HadoopKerberosName.setConfiguration(conf);
  Text owner = new Text("owner");
  Text renewer = new Text("rm/[email protected]");
  Text realUser = new Text("realUser");
  TimelineDelegationTokenIdentifier token =
      new TimelineDelegationTokenIdentifier(owner, renewer, realUser);
  Assert.assertEquals(new Text("yarn"), token.getRenewer());
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@Override
protected void storeNewToken(TimelineDelegationTokenIdentifier tokenId,
    long renewDate) {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Storing token " + tokenId.getSequenceNumber());
  }
  try {
    if (stateStore != null) {
      stateStore.storeToken(tokenId, renewDate);
    }
  } catch (IOException e) {
    LOG.error("Unable to store token " + tokenId.getSequenceNumber(), e);
  }
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@Override
protected void removeStoredToken(TimelineDelegationTokenIdentifier tokenId)
    throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Storing token " + tokenId.getSequenceNumber());
  }
  try {
    if (stateStore != null) {
      stateStore.removeToken(tokenId);
    }
  } catch (IOException e) {
    LOG.error("Unable to remove token " + tokenId.getSequenceNumber(), e);
  }
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@Override
protected void updateStoredToken(TimelineDelegationTokenIdentifier tokenId,
    long renewDate) {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Updating token " + tokenId.getSequenceNumber());
  }
  try {
    if (stateStore != null) {
      stateStore.updateToken(tokenId, renewDate);
    }
  } catch (IOException e) {
    LOG.error("Unable to update token " + tokenId.getSequenceNumber(), e);
  }
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
public void recover(TimelineServiceState state) throws IOException {
  LOG.info("Recovering " + getClass().getSimpleName());
  for (DelegationKey key : state.getTokenMasterKeyState()) {
    addKey(key);
  }
  this.delegationTokenSequenceNumber = state.getLatestSequenceNumber();
  for (Entry<TimelineDelegationTokenIdentifier, Long> entry :
      state.getTokenState().entrySet()) {
    addPersistedDelegationToken(entry.getKey(), entry.getValue());
  }
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
public TimelineDelegationTokenIdentifier getTokenIdentifier()
    throws IOException {
  ByteArrayInputStream in =
      new ByteArrayInputStream(builder.getTokenIdentifier().toByteArray());
  TimelineDelegationTokenIdentifier identifer =
      new TimelineDelegationTokenIdentifier();
  identifer.readFields(new DataInputStream(in));
  return identifer;
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@Override
public void updateToken(TimelineDelegationTokenIdentifier tokenId,
    Long renewDate) throws IOException {
  try {
    byte[] k = createTokenEntryKey(tokenId.getSequenceNumber());
    if (db.get(k) == null) {
      throw new IOException(tokenId + " doesn't exist");
    }
    byte[] v = buildTokenData(tokenId, renewDate);
    db.put(k, v);
  } catch (DBException e) {
    throw new IOException(e);
  }
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@Override
public void removeToken(TimelineDelegationTokenIdentifier tokenId)
    throws IOException {
  try {
    byte[] key = createTokenEntryKey(tokenId.getSequenceNumber());
    db.delete(key);
  } catch (DBException e) {
    throw new IOException(e);
  }
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
private static byte[] buildTokenData(
    TimelineDelegationTokenIdentifier tokenId, Long renewDate)
    throws IOException {
  TimelineDelegationTokenIdentifierData data =
      new TimelineDelegationTokenIdentifierData(tokenId, renewDate);
  return data.toByteArray();
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@Override
public void storeToken(TimelineDelegationTokenIdentifier tokenId,
    Long renewDate) throws IOException {
  if (state.tokenState.containsKey(tokenId)) {
    throw new IOException("token " + tokenId + " was stored twice");
  }
  state.tokenState.put(tokenId, renewDate);
  state.latestSequenceNumber = tokenId.getSequenceNumber();
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@Override
public void updateToken(TimelineDelegationTokenIdentifier tokenId,
    Long renewDate) throws IOException {
  if (!state.tokenState.containsKey(tokenId)) {
    throw new IOException("token " + tokenId + " not in store");
  }
  state.tokenState.put(tokenId, renewDate);
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
private void addTimelineDelegationToken(
    ContainerLaunchContext clc) throws YarnException, IOException {
  Credentials credentials = new Credentials();
  DataInputByteBuffer dibb = new DataInputByteBuffer();
  ByteBuffer tokens = clc.getTokens();
  if (tokens != null) {
    dibb.reset(tokens);
    credentials.readTokenStorageStream(dibb);
    tokens.rewind();
  }
  // If the timeline delegation token is already in the CLC, no need to add
  // one more
  for (org.apache.hadoop.security.token.Token<? extends TokenIdentifier> token : credentials
      .getAllTokens()) {
    if (token.getKind().equals(TimelineDelegationTokenIdentifier.KIND_NAME)) {
      return;
    }
  }
  org.apache.hadoop.security.token.Token<TimelineDelegationTokenIdentifier>
      timelineDelegationToken = getTimelineDelegationToken();
  if (timelineDelegationToken == null) {
    return;
  }
  credentials.addToken(timelineService, timelineDelegationToken);
  if (LOG.isDebugEnabled()) {
    LOG.debug("Add timline delegation token into credentials: "
        + timelineDelegationToken);
  }
  DataOutputBuffer dob = new DataOutputBuffer();
  credentials.writeTokenStorageToStream(dob);
  tokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
  clc.setTokens(tokens);
}
 

import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier; //导入依赖的package包/类
@VisibleForTesting
org.apache.hadoop.security.token.Token<TimelineDelegationTokenIdentifier>
    getTimelineDelegationToken() throws IOException, YarnException {
      try {
        return timelineClient.getDelegationToken(timelineDTRenewer);
      } catch (Exception e ) {
        if (timelineServiceBestEffort) {
          LOG.warn("Failed to get delegation token from the timeline server: "
              + e.getMessage());
          return null;
        }
        throw e;
      }
}