Author: ddas Date: Thu Jun 2 05:03:36 2011 New Revision: 1130409 URL: http://svn.apache.org/viewvc?rev=1130409&view=rev Log: MAPREDUCE-2452. Moves the cancellation of delegation tokens to a separate thread. Contributed by Devaraj Das.
Modified: hadoop/common/branches/branch-0.20-security/CHANGES.txt hadoop/common/branches/branch-0.20-security/src/mapred/org/apache/hadoop/mapreduce/security/token/DelegationTokenRenewal.java Modified: hadoop/common/branches/branch-0.20-security/CHANGES.txt URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security/CHANGES.txt?rev=1130409&r1=1130408&r2=1130409&view=diff ============================================================================== --- hadoop/common/branches/branch-0.20-security/CHANGES.txt (original) +++ hadoop/common/branches/branch-0.20-security/CHANGES.txt Thu Jun 2 05:03:36 2011 @@ -4,6 +4,9 @@ Release 0.20.205.0 - unreleased BUG FIXES + MAPREDUCE-2452. Moves the cancellation of delegation tokens to a separate + thread. (ddas) + HDFS-1377. Quota bug for partial blocks allows quotas to be violated. (eli) MAPREDUCE-2451. Log the details from health check script at the Modified: hadoop/common/branches/branch-0.20-security/src/mapred/org/apache/hadoop/mapreduce/security/token/DelegationTokenRenewal.java URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security/src/mapred/org/apache/hadoop/mapreduce/security/token/DelegationTokenRenewal.java?rev=1130409&r1=1130408&r2=1130409&view=diff ============================================================================== --- hadoop/common/branches/branch-0.20-security/src/mapred/org/apache/hadoop/mapreduce/security/token/DelegationTokenRenewal.java (original) +++ hadoop/common/branches/branch-0.20-security/src/mapred/org/apache/hadoop/mapreduce/security/token/DelegationTokenRenewal.java Thu Jun 2 05:03:36 2011 @@ -33,6 +33,7 @@ import java.util.Iterator; import java.util.Set; import java.util.Timer; import java.util.TimerTask; +import java.util.concurrent.LinkedBlockingQueue; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -48,6 +49,7 @@ import org.apache.hadoop.security.Creden import org.apache.hadoop.security.token.Token; import org.apache.hadoop.security.token.TokenIdentifier; import org.apache.hadoop.security.token.SecretManager.InvalidToken; +import org.apache.hadoop.util.StringUtils; //@InterfaceAudience.Private @@ -100,10 +102,75 @@ public class DelegationTokenRenewal { // global single timer (daemon) private static Timer renewalTimer = new Timer(true); + //delegation token canceler thread + private static DelegationTokenCancelThread dtCancelThread = + new DelegationTokenCancelThread(); + static { + dtCancelThread.start(); + } + + //managing the list of tokens using Map // jobId=>List<tokens> private static Set<DelegationTokenToRenew> delegationTokens = Collections.synchronizedSet(new HashSet<DelegationTokenToRenew>()); + + private static class DelegationTokenCancelThread extends Thread { + private static class TokenWithConf { + Token<DelegationTokenIdentifier> token; + Configuration conf; + TokenWithConf(Token<DelegationTokenIdentifier> token, + Configuration conf) { + this.token = token; + this.conf = conf; + } + } + private LinkedBlockingQueue<TokenWithConf> queue = + new LinkedBlockingQueue<TokenWithConf>(); + + public DelegationTokenCancelThread() { + super("Delegation Token Canceler"); + setDaemon(true); + } + public void cancelToken(Token<DelegationTokenIdentifier> token, + Configuration conf) { + TokenWithConf tokenWithConf = new TokenWithConf(token, conf); + while (!queue.offer(tokenWithConf)) { + LOG.warn("Unable to add token " + token + " for cancellation. " + + "Will retry.."); + try { + Thread.sleep(100); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + } + + public void run() { + while (true) { + TokenWithConf tokenWithConf = null; + try { + tokenWithConf = queue.take(); + DistributedFileSystem dfs = getDFSForToken(tokenWithConf.token, + tokenWithConf.conf); + if (LOG.isDebugEnabled()) { + LOG.debug("Canceling token " + tokenWithConf.token.getService() + + " for dfs=" + dfs); + } + dfs.cancelDelegationToken(tokenWithConf.token); + } catch (IOException e) { + LOG.warn("Failed to cancel token " + tokenWithConf.token + " " + + StringUtils.stringifyException(e)); + } catch (InterruptedException ie) { + return; + } catch (Throwable t) { + LOG.warn("Got exception " + StringUtils.stringifyException(t) + + ". Exiting.."); + System.exit(-1); + } + } + } + } //adding token private static void addTokenToList(DelegationTokenToRenew t) { delegationTokens.add(t); @@ -309,15 +376,7 @@ public class DelegationTokenRenewal { Configuration conf = t.conf; if(token.getKind().equals(kindHdfs)) { - try { - DistributedFileSystem dfs = getDFSForToken(token, conf); - if (LOG.isDebugEnabled()) - LOG.debug("canceling token " + token.getService() + " for dfs=" + - dfs); - dfs.cancelDelegationToken(token); - } catch (Exception e) { - LOG.warn("Failed to cancel " + token, e); - } + dtCancelThread.cancelToken(token, conf); } }