@Override
public RenewDelegationTokenResponse renewDelegationToken(
    RenewDelegationTokenRequest request) throws IOException {
  if (!isAllowedDelegationTokenOp()) {
    throw new IOException(
        "Delegation Token can be renewed only with kerberos authentication");
  }

  org.apache.hadoop.yarn.api.records.Token protoToken = request.getDelegationToken();
  Token<MRDelegationTokenIdentifier> token =
      new Token<MRDelegationTokenIdentifier>(
          protoToken.getIdentifier().array(), protoToken.getPassword()
          .array(), new Text(protoToken.getKind()), new Text(
          protoToken.getService()));

  String user = UserGroupInformation.getCurrentUser().getShortUserName();
  long nextExpTime = jhsDTSecretManager.renewToken(token, user);
  RenewDelegationTokenResponse renewResponse = Records
      .newRecord(RenewDelegationTokenResponse.class);
  renewResponse.setNextExpirationTime(nextExpTime);
  return renewResponse;
}
 

@Override
public CancelDelegationTokenResponse cancelDelegationToken(
    CancelDelegationTokenRequest request) throws IOException {
  if (!isAllowedDelegationTokenOp()) {
    throw new IOException(
        "Delegation Token can be cancelled only with kerberos authentication");
  }

  org.apache.hadoop.yarn.api.records.Token protoToken = request.getDelegationToken();
  Token<MRDelegationTokenIdentifier> token =
      new Token<MRDelegationTokenIdentifier>(
          protoToken.getIdentifier().array(), protoToken.getPassword()
          .array(), new Text(protoToken.getKind()), new Text(
          protoToken.getService()));

  String user = UserGroupInformation.getCurrentUser().getUserName();
  jhsDTSecretManager.cancelToken(token, user);
  return Records.newRecord(CancelDelegationTokenResponse.class);
}
 

@SuppressWarnings("unchecked")
public Token<MRDelegationTokenIdentifier> selectToken(Text service,
    Collection<Token<? extends TokenIdentifier>> tokens) {
  if (service == null) {
    return null;
  }
  LOG.debug("Looking for a token with service " + service.toString());
  for (Token<? extends TokenIdentifier> token : tokens) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Token kind is " + token.getKind().toString()
          + " and the token's service name is " + token.getService());
    }
    if (MRDelegationTokenIdentifier.KIND_NAME.equals(token.getKind())
        && service.equals(token.getService())) {
      return (Token<MRDelegationTokenIdentifier>) token;
    }
  }
  return null;
}
 

@Override
public void storeToken(MRDelegationTokenIdentifier tokenId, Long renewDate)
    throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Storing token " + tokenId.getSequenceNumber());
  }

  ByteArrayOutputStream memStream = new ByteArrayOutputStream();
  DataOutputStream dataStream = new DataOutputStream(memStream);
  try {
    tokenId.write(dataStream);
    dataStream.writeLong(renewDate);
    dataStream.close();
    dataStream = null;
  } finally {
    IOUtils.cleanup(LOG, dataStream);
  }

  String dbKey = getTokenDatabaseKey(tokenId);
  try {
    db.put(bytes(dbKey), memStream.toByteArray());
  } catch (DBException e) {
    throw new IOException(e);
  }
}
 

@Override
public RenewDelegationTokenResponse renewDelegationToken(
    RenewDelegationTokenRequest request) throws IOException {
    if (!isAllowedDelegationTokenOp()) {
      throw new IOException(
          "Delegation Token can be renewed only with kerberos authentication");
    }

    org.apache.hadoop.yarn.api.records.Token protoToken = request.getDelegationToken();
    Token<MRDelegationTokenIdentifier> token =
        new Token<MRDelegationTokenIdentifier>(
            protoToken.getIdentifier().array(), protoToken.getPassword()
                .array(), new Text(protoToken.getKind()), new Text(
                protoToken.getService()));

    String user = UserGroupInformation.getCurrentUser().getShortUserName();
    long nextExpTime = jhsDTSecretManager.renewToken(token, user);
    RenewDelegationTokenResponse renewResponse = Records
        .newRecord(RenewDelegationTokenResponse.class);
    renewResponse.setNextExpirationTime(nextExpTime);
    return renewResponse;
}
 

@Override
public CancelDelegationTokenResponse cancelDelegationToken(
    CancelDelegationTokenRequest request) throws IOException {
    if (!isAllowedDelegationTokenOp()) {
      throw new IOException(
          "Delegation Token can be cancelled only with kerberos authentication");
    }

    org.apache.hadoop.yarn.api.records.Token protoToken = request.getDelegationToken();
    Token<MRDelegationTokenIdentifier> token =
        new Token<MRDelegationTokenIdentifier>(
            protoToken.getIdentifier().array(), protoToken.getPassword()
                .array(), new Text(protoToken.getKind()), new Text(
                protoToken.getService()));

    String user = UserGroupInformation.getCurrentUser().getUserName();
    jhsDTSecretManager.cancelToken(token, user);
    return Records.newRecord(CancelDelegationTokenResponse.class);
}
 

@Override
public void updateToken(MRDelegationTokenIdentifier tokenId,
    Long renewDate) throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Updating token " + tokenId.getSequenceNumber());
  }

  // Files cannot be atomically replaced, therefore we write a temporary
  // update file, remove the original token file, then rename the update
  // file to the token file. During recovery either the token file will be
  // used or if that is missing and an update file is present then the
  // update file is used.
  Path tokenPath = getTokenPath(tokenId);
  Path tmp = new Path(tokenPath.getParent(),
      UPDATE_TMP_FILE_PREFIX + tokenPath.getName());
  writeFile(tmp, buildTokenData(tokenId, renewDate));
  try {
    deleteFile(tokenPath);
  } catch (IOException e) {
    fs.delete(tmp, false);
    throw e;
  }
  if (!fs.rename(tmp, tokenPath)) {
    throw new IOException("Could not rename " + tmp + " to " + tokenPath);
  }
}
 

private MRDelegationTokenIdentifier loadToken(HistoryServerState state,
    Path tokenFile, long numTokenFileBytes) throws IOException {
  MRDelegationTokenIdentifier tokenId = new MRDelegationTokenIdentifier();
  long renewDate;
  byte[] tokenData = readFile(tokenFile, numTokenFileBytes);
  DataInputStream in =
      new DataInputStream(new ByteArrayInputStream(tokenData));
  try {
    tokenId.readFields(in);
    renewDate = in.readLong();
  } finally {
    IOUtils.cleanup(LOG, in);
  }
  state.tokenState.put(tokenId, renewDate);
  return tokenId;
}
 

@SuppressWarnings("unchecked")
public Token<MRDelegationTokenIdentifier> selectToken(Text service,
    Collection<Token<? extends TokenIdentifier>> tokens) {
  if (service == null) {
    return null;
  }
  LOG.debug("Looking for a token with service " + service.toString());
  for (Token<? extends TokenIdentifier> token : tokens) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Token kind is " + token.getKind().toString()
          + " and the token's service name is " + token.getService());
    }
    if (MRDelegationTokenIdentifier.KIND_NAME.equals(token.getKind())
        && service.equals(token.getService())) {
      return (Token<MRDelegationTokenIdentifier>) token;
    }
  }
  return null;
}
 

@Override
public void storeToken(MRDelegationTokenIdentifier tokenId, Long renewDate)
    throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Storing token " + tokenId.getSequenceNumber());
  }

  ByteArrayOutputStream memStream = new ByteArrayOutputStream();
  DataOutputStream dataStream = new DataOutputStream(memStream);
  try {
    tokenId.write(dataStream);
    dataStream.writeLong(renewDate);
    dataStream.close();
    dataStream = null;
  } finally {
    IOUtils.cleanup(LOG, dataStream);
  }

  String dbKey = getTokenDatabaseKey(tokenId);
  try {
    db.put(bytes(dbKey), memStream.toByteArray());
  } catch (DBException e) {
    throw new IOException(e);
  }
}
 

@Override
public RenewDelegationTokenResponse renewDelegationToken(
    RenewDelegationTokenRequest request) throws IOException {
    if (!isAllowedDelegationTokenOp()) {
      throw new IOException(
          "Delegation Token can be renewed only with kerberos authentication");
    }

    org.apache.hadoop.yarn.api.records.Token protoToken = request.getDelegationToken();
    Token<MRDelegationTokenIdentifier> token =
        new Token<MRDelegationTokenIdentifier>(
            protoToken.getIdentifier().array(), protoToken.getPassword()
                .array(), new Text(protoToken.getKind()), new Text(
                protoToken.getService()));

    String user = UserGroupInformation.getCurrentUser().getShortUserName();
    long nextExpTime = jhsDTSecretManager.renewToken(token, user);
    RenewDelegationTokenResponse renewResponse = Records
        .newRecord(RenewDelegationTokenResponse.class);
    renewResponse.setNextExpirationTime(nextExpTime);
    return renewResponse;
}
 

@Override
public CancelDelegationTokenResponse cancelDelegationToken(
    CancelDelegationTokenRequest request) throws IOException {
    if (!isAllowedDelegationTokenOp()) {
      throw new IOException(
          "Delegation Token can be cancelled only with kerberos authentication");
    }

    org.apache.hadoop.yarn.api.records.Token protoToken = request.getDelegationToken();
    Token<MRDelegationTokenIdentifier> token =
        new Token<MRDelegationTokenIdentifier>(
            protoToken.getIdentifier().array(), protoToken.getPassword()
                .array(), new Text(protoToken.getKind()), new Text(
                protoToken.getService()));

    String user = UserGroupInformation.getCurrentUser().getUserName();
    jhsDTSecretManager.cancelToken(token, user);
    return Records.newRecord(CancelDelegationTokenResponse.class);
}
 

@Override
public void updateToken(MRDelegationTokenIdentifier tokenId,
    Long renewDate) throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Updating token " + tokenId.getSequenceNumber());
  }

  // Files cannot be atomically replaced, therefore we write a temporary
  // update file, remove the original token file, then rename the update
  // file to the token file. During recovery either the token file will be
  // used or if that is missing and an update file is present then the
  // update file is used.
  Path tokenPath = getTokenPath(tokenId);
  Path tmp = new Path(tokenPath.getParent(),
      UPDATE_TMP_FILE_PREFIX + tokenPath.getName());
  writeFile(tmp, buildTokenData(tokenId, renewDate));
  try {
    deleteFile(tokenPath);
  } catch (IOException e) {
    fs.delete(tmp, false);
    throw e;
  }
  if (!fs.rename(tmp, tokenPath)) {
    throw new IOException("Could not rename " + tmp + " to " + tokenPath);
  }
}
 

private MRDelegationTokenIdentifier loadToken(HistoryServerState state,
    Path tokenFile, long numTokenFileBytes) throws IOException {
  MRDelegationTokenIdentifier tokenId = new MRDelegationTokenIdentifier();
  long renewDate;
  byte[] tokenData = readFile(tokenFile, numTokenFileBytes);
  DataInputStream in =
      new DataInputStream(new ByteArrayInputStream(tokenData));
  try {
    tokenId.readFields(in);
    renewDate = in.readLong();
  } finally {
    IOUtils.cleanup(LOG, in);
  }
  state.tokenState.put(tokenId, renewDate);
  return tokenId;
}
 

private void loadToken(HistoryServerState state, byte[] data)
    throws IOException {
  MRDelegationTokenIdentifier tokenId = new MRDelegationTokenIdentifier();
  long renewDate;
  DataInputStream in = new DataInputStream(new ByteArrayInputStream(data));
  try {
    tokenId.readFields(in);
    renewDate = in.readLong();
  } finally {
    IOUtils.cleanup(LOG, in);
  }
  state.tokenState.put(tokenId, renewDate);
}
 

@Override
public void removeToken(MRDelegationTokenIdentifier tokenId)
    throws IOException {
  String dbKey = getTokenDatabaseKey(tokenId);
  try {
    db.delete(bytes(dbKey));
  } catch (DBException e) {
    throw new IOException(e);
  }
}
 

@Override
public void storeToken(MRDelegationTokenIdentifier tokenId,
    Long renewDate) throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Storing token " + tokenId.getSequenceNumber());
  }

  Path tokenPath = getTokenPath(tokenId);
  if (fs.exists(tokenPath)) {
    throw new IOException(tokenPath + " already exists");
  }

  createNewFile(tokenPath, buildTokenData(tokenId, renewDate));
}
 

@Override
public void removeToken(MRDelegationTokenIdentifier tokenId)
    throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Removing token " + tokenId.getSequenceNumber());
  }
  deleteFile(getTokenPath(tokenId));
}
 

private byte[] buildTokenData(MRDelegationTokenIdentifier tokenId,
    Long renewDate) throws IOException {
  ByteArrayOutputStream memStream = new ByteArrayOutputStream();
  DataOutputStream dataStream = new DataOutputStream(memStream);
  try {
    tokenId.write(dataStream);
    dataStream.writeLong(renewDate);
    dataStream.close();
    dataStream = null;
  } finally {
    IOUtils.cleanup(LOG, dataStream);
  }
  return memStream.toByteArray();
}
 

private void loadTokenFromBucket(int bucketId,
    HistoryServerState state, Path tokenFile, long numTokenFileBytes)
        throws IOException {
  MRDelegationTokenIdentifier token =
      loadToken(state, tokenFile, numTokenFileBytes);
  int tokenBucketId = getBucketId(token);
  if (tokenBucketId != bucketId) {
    throw new IOException("Token " + tokenFile
        + " should be in bucket " + tokenBucketId + ", found in bucket "
        + bucketId);
  }
}
 

@Override
protected void storeNewToken(MRDelegationTokenIdentifier tokenId,
    long renewDate) {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Storing token " + tokenId.getSequenceNumber());
  }
  try {
    store.storeToken(tokenId, renewDate);
  } catch (IOException e) {
    LOG.error("Unable to store token " + tokenId.getSequenceNumber(), e);
  }
}
 

@Override
protected void removeStoredToken(MRDelegationTokenIdentifier tokenId)
    throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Storing token " + tokenId.getSequenceNumber());
  }
  try {
    store.removeToken(tokenId);
  } catch (IOException e) {
    LOG.error("Unable to remove token " + tokenId.getSequenceNumber(), e);
  }
}
 

@Override
protected void updateStoredToken(MRDelegationTokenIdentifier tokenId,
    long renewDate) {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Updating token " + tokenId.getSequenceNumber());
  }
  try {
    store.updateToken(tokenId, renewDate);
  } catch (IOException e) {
    LOG.error("Unable to update token " + tokenId.getSequenceNumber(), e);
  }
}
 

public void recover(HistoryServerState state) throws IOException {
  LOG.info("Recovering " + getClass().getSimpleName());
  for (DelegationKey key : state.tokenMasterKeyState) {
    addKey(key);
  }
  for (Entry<MRDelegationTokenIdentifier, Long> entry :
      state.tokenState.entrySet()) {
    addPersistedDelegationToken(entry.getKey(), entry.getValue());
  }
}
 

@Override
public void storeToken(MRDelegationTokenIdentifier tokenId, Long renewDate)
    throws IOException {
  if (state.tokenState.containsKey(tokenId)) {
    throw new IOException("token " + tokenId + " was stored twice");
  }
  state.tokenState.put(tokenId, renewDate);
}
 

@Override
public void updateToken(MRDelegationTokenIdentifier tokenId, Long renewDate)
    throws IOException {
  if (!state.tokenState.containsKey(tokenId)) {
    throw new IOException("token " + tokenId + " not in store");
  }
  state.tokenState.put(tokenId, renewDate);
}
 

private void loadToken(HistoryServerState state, byte[] data)
    throws IOException {
  MRDelegationTokenIdentifier tokenId = new MRDelegationTokenIdentifier();
  long renewDate;
  DataInputStream in = new DataInputStream(new ByteArrayInputStream(data));
  try {
    tokenId.readFields(in);
    renewDate = in.readLong();
  } finally {
    IOUtils.cleanup(LOG, in);
  }
  state.tokenState.put(tokenId, renewDate);
}
 

@Override
public void removeToken(MRDelegationTokenIdentifier tokenId)
    throws IOException {
  String dbKey = getTokenDatabaseKey(tokenId);
  try {
    db.delete(bytes(dbKey));
  } catch (DBException e) {
    throw new IOException(e);
  }
}
 

@Override
public void storeToken(MRDelegationTokenIdentifier tokenId,
    Long renewDate) throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Storing token " + tokenId.getSequenceNumber());
  }

  Path tokenPath = getTokenPath(tokenId);
  if (fs.exists(tokenPath)) {
    throw new IOException(tokenPath + " already exists");
  }

  createNewFile(tokenPath, buildTokenData(tokenId, renewDate));
}
 

@Override
public void removeToken(MRDelegationTokenIdentifier tokenId)
    throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Removing token " + tokenId.getSequenceNumber());
  }
  deleteFile(getTokenPath(tokenId));
}