commit 5dd9e6a03d8f026c4ed4af85013bff0ca2cc34a5

Author: Boris Shkolnik <borya@yahoo-inc.com>
Date:   Mon Feb 1 22:52:13 2010 -0800

    MAPREDUCE:1383 from https://issues.apache.org/jira/secure/attachment/12434455/MAPREDUCE-1383-BP20-7.patch
    
    +++ b/YAHOO-CHANGES.txt
    +    MAPREDUCE-1383. Automates fetching of delegation tokens in File*Formats
    +    Distributed Cache and Distcp. Also, provides a config mapreduce.job.hdfs-servers
    +    that the jobs can populate with a comma separated list of namenodes.
    +    The job client automatically fetches delegation tokens from those namenodes.
    +


git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.20-security-patches@1077141 13f79535-47bb-0310-9956-ffa450edef68

build.xml

@@ -367,7 +367,7 @@
      
   </target>
 
-  <target name="compile-mapred-classes" depends="compile-core-classes">
+  <target name="compile-mapred-classes" depends="compile-core-classes,compile-hdfs-classes">
     <jsp-compile
      uriroot="${src.webapps}/task"
      outputdir="${build.src}"

src/mapred/org/apache/hadoop/filecache/TrackerDistributedCacheManager.java

@@ -42,6 +42,7 @@ import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.mapreduce.JobContext;
 import org.apache.hadoop.util.RunJar;
+import org.apache.hadoop.mapreduce.security.TokenCache;
 
 /**
  * Manages a single machine's instance of a cross-job
@@ -725,4 +726,32 @@ public class TrackerDistributedCacheManager {
   static void setFileVisibilities(Configuration conf, String booleans) {
     conf.set(JobContext.CACHE_FILE_VISIBILITIES, booleans);
   }
+  
+  /**
+   * For each archive or cache file - get the corresponding delegation token
+   * @param job
+   * @throws IOException
+   */
+  public static void getDelegationTokens(Configuration job) throws IOException {
+    URI[] tarchives = DistributedCache.getCacheArchives(job);
+    URI[] tfiles = DistributedCache.getCacheFiles(job);
+
+    int size = (tarchives!=null? tarchives.length : 0) + (tfiles!=null ? tfiles.length :0);
+    Path[] ps = new Path[size];
+
+    int i = 0;
+    if (tarchives != null) {
+      for (i=0; i < tarchives.length; i++) {
+        ps[i] = new Path(tarchives[i].toString());
+      }
+    }
+
+    if (tfiles != null) {
+      for(int j=0; j< tfiles.length; j++) {
+        ps[i+j] = new Path(tfiles[j].toString());
+      }
+    }
+
+    TokenCache.obtainTokensForNamenodes(ps, job);
+  }
 }

src/mapred/org/apache/hadoop/mapred/Child.java

@@ -26,25 +26,20 @@ import java.net.InetSocketAddress;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSError;
 import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.FileUtil;
-import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.ipc.RPC;
-import org.apache.hadoop.mapred.JvmTask;
-import org.apache.hadoop.mapreduce.JobContext;
-import org.apache.hadoop.mapreduce.security.token.JobTokenIdentifier;
+import org.apache.hadoop.mapreduce.security.TokenCache;
+import org.apache.hadoop.mapreduce.security.TokenStorage;
 import org.apache.hadoop.mapreduce.security.token.JobTokenSecretManager;
 import org.apache.hadoop.metrics.MetricsContext;
 import org.apache.hadoop.metrics.MetricsUtil;
 import org.apache.hadoop.metrics.jvm.JvmMetrics;
 import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.mapreduce.security.TokenCache;
-import org.apache.hadoop.mapreduce.security.TokenStorage;
-import org.apache.log4j.LogManager;
+import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.util.Shell;
 import org.apache.hadoop.util.StringUtils;
+import org.apache.log4j.LogManager;
 
 /** 
  * The main() for child processes. 
@@ -72,8 +67,7 @@ class Child {
 
     // file name is passed thru env
     String jobTokenFile = System.getenv().get("JOB_TOKEN_FILE");
-    defaultConf.set(JobContext.JOB_TOKEN_FILE, jobTokenFile);
-    TokenStorage ts = TokenCache.loadTaskTokenStorage(defaultConf);
+    TokenStorage ts = TokenCache.loadTaskTokenStorage(jobTokenFile, defaultConf);
     LOG.debug("loading token. # keys =" +ts.numberOfSecretKeys() + 
         "; from file=" + jobTokenFile);
 
@@ -155,7 +149,7 @@ class Child {
         JobConf job = new JobConf(task.getJobFile());
 
         // set job shuffle token
-         Token<JobTokenIdentifier> jt = (Token<JobTokenIdentifier>)ts.getJobToken();
+        Token<? extends TokenIdentifier> jt = ts.getJobToken();
         // set the jobTokenFile into task
         task.setJobTokenSecret(JobTokenSecretManager.createSecretKey(jt.getPassword()));
 

src/mapred/org/apache/hadoop/mapred/FileInputFormat.java

@@ -36,6 +36,7 @@ import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.PathFilter;
+import org.apache.hadoop.mapreduce.security.TokenCache;
 import org.apache.hadoop.net.NetworkTopology;
 import org.apache.hadoop.net.Node;
 import org.apache.hadoop.net.NodeBase;
@@ -152,6 +153,9 @@ public abstract class FileInputFormat<K, V> implements InputFormat<K, V> {
       throw new IOException("No input paths specified in job");
     }
 
+    // get tokens for all the required FileSystems..
+    TokenCache.obtainTokensForNamenodes(dirs, job);
+    
     List<FileStatus> result = new ArrayList<FileStatus>();
     List<IOException> errors = new ArrayList<IOException>();
     

src/mapred/org/apache/hadoop/mapred/FileOutputFormat.java

@@ -24,6 +24,7 @@ import java.text.NumberFormat;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.mapreduce.security.TokenCache;
 import org.apache.hadoop.util.Progressable;
 
 /** A base class for {@link OutputFormat}. */
@@ -106,6 +107,10 @@ public abstract class FileOutputFormat<K, V> implements OutputFormat<K, V> {
       // normalize the output directory
       outDir = fs.makeQualified(outDir);
       setOutputPath(job, outDir);
+      
+      // get delegation token for the outDir's file system
+      TokenCache.obtainTokensForNamenodes(new Path[] {outDir}, job);
+      
       // check its existence
       if (fs.exists(outDir)) {
         throw new FileAlreadyExistsException("Output directory " + outDir + 

src/mapred/org/apache/hadoop/mapred/JobClient.java

@@ -64,7 +64,7 @@ import org.apache.hadoop.mapreduce.InputFormat;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.JobContext;
 import org.apache.hadoop.mapreduce.JobSubmissionFiles;
-import org.apache.hadoop.mapreduce.security.TokenStorage;
+import org.apache.hadoop.mapreduce.security.TokenCache;
 import org.apache.hadoop.mapreduce.split.JobSplitWriter;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -72,6 +72,8 @@ import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.util.ToolRunner;
+import org.codehaus.jackson.JsonParseException;
+import org.codehaus.jackson.map.JsonMappingException;
 import org.codehaus.jackson.map.ObjectMapper;
 
 /**
@@ -639,6 +641,8 @@ public class JobClient extends Configured implements MRConstants, Tool  {
     TrackerDistributedCacheManager.determineTimestamps(job);
     //  set the public/private visibility of the archives and files
     TrackerDistributedCacheManager.determineCacheVisibilities(job);
+    // get DelegationTokens for cache files
+    TrackerDistributedCacheManager.getDelegationTokens(job);
 
     String originalJarPath = job.getJar();
 
@@ -726,7 +730,12 @@ public class JobClient extends Configured implements MRConstants, Tool  {
     job.set("mapreduce.job.dir", submitJobDir.toString());
     JobStatus status = null;
     try {
+      
       copyAndConfigureFiles(job, submitJobDir);
+      
+      // get delegation token for the dir
+      TokenCache.obtainTokensForNamenodes(new Path [] {submitJobDir}, job);
+      
       Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);
       int reduces = job.getNumReduceTasks();
       JobContext context = new JobContext(job, jobId);
@@ -756,29 +765,13 @@ public class JobClient extends Configured implements MRConstants, Tool  {
         out.close();
       }
       
-      // create TokenStorage object with user secretKeys
-      String tokensFileName = job.get("tokenCacheFile");
-      TokenStorage tokenStorage = null;
-      if(tokensFileName != null) {
-        LOG.info("loading secret keys from " + tokensFileName);
-        String localFileName = new Path(tokensFileName).toUri().getPath();
-        tokenStorage = new TokenStorage();
-        // read JSON
-        ObjectMapper mapper = new ObjectMapper();
-        Map<String, String> nm = 
-          mapper.readValue(new File(localFileName), Map.class);
-        
-        for(Map.Entry<String, String> ent: nm.entrySet()) {
-          LOG.debug("adding secret key alias="+ent.getKey());
-          tokenStorage.addSecretKey(new Text(ent.getKey()), ent.getValue().getBytes());
-        }
-      }
 
       //
       // Now, actually submit the job (using the submit name)
       //
+      populateTokenCache(job);
       status = jobSubmitClient.submitJob(
-          jobId, submitJobDir.toString(), tokenStorage);
+         jobId, submitJobDir.toString(), TokenCache.getTokenStorage());
       if (status != null) {
         return new NetworkedJob(status);
       } else {
@@ -1789,5 +1782,45 @@ public class JobClient extends Configured implements MRConstants, Tool  {
     int res = ToolRunner.run(new JobClient(), argv);
     System.exit(res);
   }
+  
+  //get secret keys and tokens and store them into TokenCache
+  @SuppressWarnings("unchecked")
+  private void populateTokenCache(Configuration conf) throws IOException{
+    // create TokenStorage object with user secretKeys
+    String tokensFileName = conf.get("tokenCacheFile");
+    if(tokensFileName != null) {
+      LOG.info("loading user's secret keys from " + tokensFileName);
+      String localFileName = new Path(tokensFileName).toUri().getPath();
+
+      boolean json_error = false;
+      try {
+        // read JSON
+        ObjectMapper mapper = new ObjectMapper();
+        Map<String, String> nm = 
+          mapper.readValue(new File(localFileName), Map.class);
+
+        for(Map.Entry<String, String> ent: nm.entrySet()) {
+          TokenCache.setSecretKey(new Text(ent.getKey()), ent.getValue().getBytes());
+        }
+      } catch (JsonMappingException e) {
+        json_error = true;
+      } catch (JsonParseException e) {
+        json_error = true;
+      }
+      if(json_error)
+        LOG.warn("couldn't parse Token Cache JSON file with user secret keys");
+    }
+
+    // add the delegation tokens from configuration
+    String [] nameNodes = conf.getStrings(JobContext.JOB_NAMENODES);
+    LOG.info("adding the following namenodes' delegation tokens:" + Arrays.toString(nameNodes));
+    if(nameNodes != null) {
+      Path [] ps = new Path[nameNodes.length];
+      for(int i=0; i< nameNodes.length; i++) {
+        ps[i] = new Path(nameNodes[i]);
+      }
+      TokenCache.obtainTokensForNamenodes(ps, conf);
+    }
+  }
 }
 

src/mapred/org/apache/hadoop/mapred/JobInProgress.java

@@ -43,7 +43,7 @@ import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapred.CleanupQueue.PathDeletionContext;
 import org.apache.hadoop.mapred.JobHistory.Values;
 import org.apache.hadoop.mapreduce.security.token.JobTokenIdentifier;
-import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
+import org.apache.hadoop.mapreduce.security.TokenCache;
 import org.apache.hadoop.mapreduce.security.TokenStorage;
 import org.apache.hadoop.mapreduce.security.token.JobTokenIdentifier;
 import org.apache.hadoop.metrics.MetricsContext;
@@ -3090,7 +3090,7 @@ class JobInProgress {
    */
   private void generateAndStoreTokens() throws IOException {
     Path jobDir = jobtracker.getSystemDirectoryForJob(jobId);
-    Path keysFile = new Path(jobDir, SecureShuffleUtils.JOB_TOKEN_FILENAME);
+    Path keysFile = new Path(jobDir, TokenCache.JOB_TOKEN_HDFS_FILE);
     // we need to create this file using the jobtracker's filesystem
     FSDataOutputStream os = jobtracker.getFileSystem().create(keysFile);
     //create JobToken file and write token to it

src/mapred/org/apache/hadoop/mapred/TaskRunner.java

@@ -36,6 +36,7 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.filecache.DistributedCache;
 import org.apache.hadoop.filecache.TaskDistributedCacheManager;
 import org.apache.hadoop.filecache.TrackerDistributedCacheManager;
+import org.apache.hadoop.mapreduce.security.TokenCache;
 import org.apache.hadoop.mapreduce.server.tasktracker.Localizer;
 import org.apache.hadoop.fs.FSError;
 import org.apache.hadoop.fs.FileSystem;
@@ -497,7 +498,7 @@ abstract class TaskRunner extends Thread {
     }
     env.put("LD_LIBRARY_PATH", ldLibraryPath.toString());
 
-    String jobTokenFile = conf.get(JobContext.JOB_TOKEN_FILE);
+    String jobTokenFile = conf.get(TokenCache.JOB_TOKEN_FILENAME);
     LOG.debug("putting jobToken file name into environment fn=" + jobTokenFile);
     env.put("JOB_TOKEN_FILE", jobTokenFile);
     

src/mapred/org/apache/hadoop/mapred/TaskTracker.java

@@ -927,6 +927,7 @@ public class TaskTracker
                               new LocalDirAllocator("mapred.local.dir");
 
   // intialize the job directory
+  @SuppressWarnings("unchecked")
   private void localizeJob(TaskInProgress tip) 
   throws IOException, InterruptedException {
     Path localJarFile = null;
@@ -955,7 +956,8 @@ public class TaskTracker
         rjob.keepJobFiles = ((localJobConf.getKeepTaskFilesPattern() != null) ||
                              localJobConf.getKeepFailedTaskFiles());
 
-        TokenStorage ts = TokenCache.loadTokens(rjob.jobConf);
+        TokenStorage ts = TokenCache.loadTokens(rjob.jobConf.get(TokenCache.JOB_TOKEN_FILENAME), rjob.jobConf);
+
         Token<JobTokenIdentifier> jt = (Token<JobTokenIdentifier>)ts.getJobToken(); 
         getJobTokenSecretManager().addTokenForJob(jobId.toString(), jt);
  
@@ -3707,7 +3709,7 @@ public class TaskTracker
         throws IOException {
       // check if the tokenJob file is there..
       Path skPath = new Path(systemDirectory, 
-          jobId.toString()+"/"+SecureShuffleUtils.JOB_TOKEN_FILENAME);
+          jobId.toString()+"/"+TokenCache.JOB_TOKEN_HDFS_FILE);
       
       FileStatus status = null;
       long jobTokenSize = -1;
@@ -3724,7 +3726,7 @@ public class TaskTracker
       // Download job_token
       systemFS.copyToLocalFile(skPath, localJobTokenFile);      
       // set it into jobConf to transfer the name to TaskRunner
-      jobConf.set(JobContext.JOB_TOKEN_FILE,localJobTokenFile.toString());
+      jobConf.set(TokenCache.JOB_TOKEN_FILENAME, localJobTokenFile.toString());
     }
 
 }

src/mapred/org/apache/hadoop/mapreduce/JobContext.java

@@ -48,7 +48,8 @@ public class JobContext {
   protected final org.apache.hadoop.mapred.JobConf conf;
   private final JobID jobId;
 
-  public static final String JOB_TOKEN_FILE = "mapreduce.job.jobTokenFile";
+  public static final String JOB_NAMENODES = "mapreduce.job.hdfs-servers";
+  public static final String JOB_JOBTRACKER_ID = "mapreduce.job.kerberos.jtprinicipal";
 
   public static final String CACHE_FILE_VISIBILITIES = 
     "mapreduce.job.cache.files.visibilities";

src/mapred/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.java

@@ -35,6 +35,7 @@ import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.mapreduce.JobContext;
 import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.security.TokenCache;
 import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.StringUtils;
 
@@ -185,6 +186,9 @@ public abstract class FileInputFormat<K, V> extends InputFormat<K, V> {
     if (dirs.length == 0) {
       throw new IOException("No input paths specified in job");
     }
+    
+    // get tokens for all the required FileSystems..
+    TokenCache.obtainTokensForNamenodes(dirs, job.getConfiguration());
 
     List<IOException> errors = new ArrayList<IOException>();
     

src/mapred/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.java

@@ -35,6 +35,7 @@ import org.apache.hadoop.mapreduce.RecordWriter;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.hadoop.mapreduce.TaskID;
 import org.apache.hadoop.mapreduce.TaskInputOutputContext;
+import org.apache.hadoop.mapreduce.security.TokenCache;
 
 /** A base class for {@link OutputFormat}s that read from {@link FileSystem}s.*/
 public abstract class FileOutputFormat<K, V> extends OutputFormat<K, V> {
@@ -119,6 +120,10 @@ public abstract class FileOutputFormat<K, V> extends OutputFormat<K, V> {
     if (outDir == null) {
       throw new InvalidJobConfException("Output directory not set.");
     }
+    
+    // get delegation token for outDir's file system
+    TokenCache.obtainTokensForNamenodes(new Path[] {outDir}, job.getConfiguration());
+
     if (outDir.getFileSystem(job.getConfiguration()).exists(outDir)) {
       throw new FileAlreadyExistsException("Output directory " + outDir + 
                                            " already exists");

src/mapred/org/apache/hadoop/mapreduce/security/SecureShuffleUtils.java

@@ -38,10 +38,6 @@ import org.apache.hadoop.record.Utils;
 public class SecureShuffleUtils {
   public static final String HTTP_HEADER_URL_HASH = "UrlHash";
   public static final String HTTP_HEADER_REPLY_URL_HASH = "ReplyHash";
-  /**
-   * file name used on HDFS for generated job token
-   */
-  public static final String JOB_TOKEN_FILENAME = "jobToken";
   
   /**
    * Base64 encoded hash of msg

src/mapred/org/apache/hadoop/mapreduce/security/TokenCache.java

@@ -19,16 +19,23 @@
 package org.apache.hadoop.mapreduce.security;
 
 import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.URI;
 import java.util.Collection;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.hadoop.hdfs.security.token.DelegationTokenIdentifier;
+import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.TokenIdentifier;
 
@@ -40,6 +47,16 @@ import org.apache.hadoop.security.token.TokenIdentifier;
 public class TokenCache {
   
   private static final Log LOG = LogFactory.getLog(TokenCache.class);
+  /**
+   * file name used on HDFS for generated job token
+   */
+  public static final String JOB_TOKEN_HDFS_FILE = "jobToken";
+
+  /**
+   * conf setting for job tokens cache file name
+   */
+
+  public static final String JOB_TOKEN_FILENAME = "mapreduce.job.jobTokenFile";
 
   private static TokenStorage tokenStorage;
   
@@ -71,6 +88,16 @@ public class TokenCache {
     getTokenStorage().setToken(new Text(namenode), t);
   }
   
+  /**
+   * 
+   * @param namenode
+   * @return delegation token
+   */
+  @SuppressWarnings("unchecked")
+  public static Token<DelegationTokenIdentifier> getDelegationToken(String namenode) {
+    return (Token<DelegationTokenIdentifier>)getTokenStorage().getToken(new Text(namenode));
+  }
+
   /**
    * auxiliary method 
    * @return all the available tokens
@@ -109,12 +136,12 @@ public class TokenCache {
    * @throws IOException
    */
   //@InterfaceAudience.Private
-  public static TokenStorage loadTaskTokenStorage(JobConf conf)
+  public static TokenStorage loadTaskTokenStorage(String fileName, JobConf conf)
   throws IOException {
     if(tokenStorage != null)
       return tokenStorage;
     
-    tokenStorage = loadTokens(conf);
+    tokenStorage = loadTokens(fileName, conf);
     
     return tokenStorage;
   }
@@ -125,9 +152,8 @@ public class TokenCache {
    * @throws IOException
    */
   //@InterfaceAudience.Private
-  public static TokenStorage loadTokens(JobConf conf) 
+  public static TokenStorage loadTokens(String jobTokenFile, JobConf conf) 
   throws IOException {
-    String jobTokenFile = conf.get(JobContext.JOB_TOKEN_FILE);
     Path localJobTokenFile = new Path (jobTokenFile);
     FileSystem localFS = FileSystem.getLocal(conf);
     FSDataInputStream in = localFS.open(localJobTokenFile);
@@ -135,9 +161,62 @@ public class TokenCache {
     TokenStorage ts = new TokenStorage();
     ts.readFields(in);
 
-    LOG.info("Task: Loaded jobTokenFile from: "+localJobTokenFile.toUri().getPath() 
+    if(LOG.isDebugEnabled()) {
+      LOG.debug("Task: Loaded jobTokenFile from: "+localJobTokenFile.toUri().getPath() 
         +"; num of sec keys  = " + ts.numberOfSecretKeys());
+    }
     in.close();
     return ts;
   }
+
+  static String buildDTServiceName(URI uri) {
+    int port = uri.getPort();
+    if(port == -1) 
+      port = NameNode.DEFAULT_PORT;
+    
+    // build the service name string "/ip:port"
+    // for whatever reason using NetUtils.createSocketAddr(target).toString()
+    // returns "localhost/ip:port"
+    StringBuffer sb = new StringBuffer();
+    sb.append(NetUtils.normalizeHostName(uri.getHost())).append(":").append(port);
+    return sb.toString();
+  }
+    
+  /**
+   * get Delegation for each distinct dfs for given paths.
+   * @param ps
+   * @param conf
+   * @throws IOException
+   */
+  public static void obtainTokensForNamenodes(Path [] ps, Configuration conf) 
+  throws IOException {
+    // get jobtracker principal id (for the renewer)
+    Text jtCreds = new Text(conf.get(JobContext.JOB_JOBTRACKER_ID, ""));
+
+    for(Path p: ps) {
+      FileSystem fs = FileSystem.get(p.toUri(), conf);
+      if(fs instanceof DistributedFileSystem) {
+        DistributedFileSystem dfs = (DistributedFileSystem)fs;
+        URI uri = fs.getUri();
+        String fs_addr = buildDTServiceName(uri);
+        
+        // see if we already have the token
+        Token<DelegationTokenIdentifier> token =
+          TokenCache.getDelegationToken(fs_addr);
+        if(token != null) {
+          LOG.debug("DT for " + token.getService()  + " is already present");
+          continue;
+        }
+        // get the token
+        token = dfs.getDelegationToken(jtCreds);
+        if(token==null)
+          throw new IOException("Token from " + fs_addr + " is null");
+
+        token.setService(new Text(fs_addr));
+        TokenCache.addDelegationToken(fs_addr, token);
+        LOG.info("getting dt for " + p.toString() + ";uri="+ fs_addr +
+            ";t.service="+token.getService());
+      }
+    }
+  }
 }

src/mapred/org/apache/hadoop/mapreduce/security/TokenStorage.java

@@ -40,7 +40,7 @@ import org.apache.hadoop.security.token.TokenIdentifier;
 //@InterfaceAudience.Private
 public class TokenStorage implements Writable {
 
-  private static final Text SHUFFLE_JOB_TOKEN = new Text("ShuffleJobToken");
+  private static final Text JOB_TOKEN = new Text("ShuffleJobToken");
 
   private  Map<Text, byte[]> secretKeysMap = new HashMap<Text, byte[]>();
   private  Map<Text, Token<? extends TokenIdentifier>> tokenMap = 
@@ -74,7 +74,7 @@ public class TokenStorage implements Writable {
    */
   //@InterfaceAudience.Private
   public void setJobToken(Token<? extends TokenIdentifier> t) {
-    setToken(SHUFFLE_JOB_TOKEN, t);
+    setToken(JOB_TOKEN, t);
   }
   
   /**
@@ -83,7 +83,7 @@ public class TokenStorage implements Writable {
    */
   //@InterfaceAudience.Private
   public Token<? extends TokenIdentifier> getJobToken() {
-    return getToken(SHUFFLE_JOB_TOKEN);
+    return getToken(JOB_TOKEN);
   }
   
   /**

src/test/org/apache/hadoop/mapred/TestTaskTrackerLocalization.java

@@ -21,31 +21,28 @@ import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
 import java.util.jar.JarOutputStream;
 import java.util.zip.ZipEntry;
 
+import junit.framework.TestCase;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.mapreduce.server.tasktracker.Localizer;
-import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
-import org.apache.hadoop.mapreduce.security.token.JobTokenIdentifier;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.util.Shell;
-import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.mapred.JvmManager.JvmEnv;
 import org.apache.hadoop.mapred.TaskController.JobInitializationContext;
 import org.apache.hadoop.mapred.TaskController.TaskControllerContext;
 import org.apache.hadoop.mapred.TaskTracker.TaskInProgress;
 import org.apache.hadoop.mapred.UtilsForTests.InlineCleanupQueue;
-
-import junit.framework.TestCase;
+import org.apache.hadoop.mapreduce.security.TokenCache;
+import org.apache.hadoop.mapreduce.security.token.JobTokenIdentifier;
+import org.apache.hadoop.mapreduce.server.tasktracker.Localizer;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.util.Shell;
 
 /**
  * Test to verify localization of a job and localization of a task on a
@@ -207,7 +204,7 @@ public class TestTaskTrackerLocalization extends TestCase {
     if(!dir.exists())
       assertTrue("faild to create dir="+dir.getAbsolutePath(), dir.mkdirs());
 
-    File jobTokenFile = new File(dir, SecureShuffleUtils.JOB_TOKEN_FILENAME);
+    File jobTokenFile = new File(dir, TokenCache.JOB_TOKEN_HDFS_FILE);
     FileOutputStream fos = new FileOutputStream(jobTokenFile);
     java.io.DataOutputStream out = new java.io.DataOutputStream(fos);
     Token<JobTokenIdentifier> jt = new Token<JobTokenIdentifier>();

src/test/org/apache/hadoop/security/TestTokenCache.java → src/test/org/apache/hadoop/mapreduce/security/TestTokenCache.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.hadoop.security;
+package org.apache.hadoop.mapreduce.security;
 
 
 import static org.junit.Assert.assertEquals;
@@ -23,7 +23,9 @@ import static org.junit.Assert.fail;
 
 import java.io.File;
 import java.io.IOException;
+import java.net.URI;
 import java.security.NoSuchAlgorithmException;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -32,8 +34,12 @@ import javax.crypto.spec.SecretKeySpec;
 
 import org.apache.commons.codec.binary.Base64;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.examples.SleepJob;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.security.token.DelegationTokenIdentifier;
+import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.io.Text;
@@ -41,16 +47,20 @@ import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.MiniMRCluster;
 import org.apache.hadoop.mapred.OutputCollector;
 import org.apache.hadoop.mapred.Reporter;
-import org.apache.hadoop.mapreduce.Job;
-//import org.apache.hadoop.mapreduce.SleepJob;
-import org.apache.hadoop.examples.SleepJob;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.security.TokenCache;
+import org.apache.hadoop.mapreduce.security.TokenStorage;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
+import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.ToolRunner;
 import org.codehaus.jackson.map.ObjectMapper;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
-import org.apache.hadoop.mapreduce.security.TokenStorage;
-import org.apache.hadoop.mapreduce.security.TokenCache;
+import static org.junit.Assert.*;
+
 
 public class TestTokenCache {
   private static final int NUM_OF_KEYS = 10;
@@ -67,15 +77,29 @@ public class TestTokenCache {
       // get token storage and a key
       TokenStorage ts = TokenCache.getTokenStorage();
       byte[] key1 = TokenCache.getSecretKey(new Text("alias1"));
-      
+      Collection<Token<? extends TokenIdentifier>> dts = TokenCache.getAllTokens();
+      int dts_size = 0;
+      if(dts != null)
+        dts_size = dts.size();
+
       System.out.println("inside MAP: ts==NULL?=" + (ts==null) + 
           "; #keys = " + (ts==null? 0:ts.numberOfSecretKeys()) + 
           ";jobToken = " +  (ts==null? "n/a":ts.getJobToken()) +
-          "; alias1 key=" + new String(key1));
+          "; alias1 key=" + new String(key1) + 
+          "; dts size= " + dts_size);
     
-      if(key1 == null || ts == null || ts.numberOfSecretKeys() != NUM_OF_KEYS) {
+      for(Token<? extends TokenIdentifier> t : dts) {
+        System.out.println(t.getKind() + "=" + StringUtils.byteToHexString(t.getPassword()));
+      }
+
+      if(dts.size() != 2) { // one job token and one delegation token
         throw new RuntimeException("tokens are not available"); // fail the test
-      } 
+      }
+
+      if(key1 == null || ts == null || ts.numberOfSecretKeys() != NUM_OF_KEYS) {
+        throw new RuntimeException("secret keys are not available"); // fail the test
+      }
+      
       super.map(key, value, output, reporter);
     }
     
@@ -174,6 +198,14 @@ public class TestTokenCache {
     // make sure JT starts
     jConf = mrCluster.createJobConf();
     
+    // provide namenodes names for the job to get the delegation tokens for
+    //String nnUri = dfsCluster.getNameNode().getUri(namenode).toString();
+    NameNode nn = dfsCluster.getNameNode();
+    URI nnUri = NameNode.getUri(nn.getNameNodeAddress());
+    jConf.set(JobContext.JOB_NAMENODES, nnUri + "," + nnUri.toString());
+    // job tracker principle id..
+    jConf.set(JobContext.JOB_JOBTRACKER_ID, "jt_id");
+
     // using argument to pass the file name
     String[] args = {
        "-tokenCacheFile", tokenFileName.toString(), 
@@ -214,4 +246,39 @@ public class TestTokenCache {
     }
     assertEquals("local job res is not 0", res, 0);
   }
+  
+  @Test
+  public void testGetTokensForNamenodes() throws IOException {
+    FileSystem fs = dfsCluster.getFileSystem();
+
+    Path p1 = new Path("file1");
+    Path p2 = new Path("file2");
+
+    p1 = fs.makeQualified(p1);
+    // do not qualify p2
+
+    TokenCache.obtainTokensForNamenodes(new Path [] {p1, p2}, jConf);
+
+    // this token is keyed by hostname:port key.
+    String fs_addr = TokenCache.buildDTServiceName(p1.toUri()); 
+    Token<DelegationTokenIdentifier> nnt = TokenCache.getDelegationToken(fs_addr);
+    System.out.println("dt for " + p1 + "(" + fs_addr + ")" + " = " +  nnt);
+
+    assertNotNull("Token for nn is null", nnt);
+
+    // verify the size
+    Collection<Token<? extends TokenIdentifier>> tns = TokenCache.getAllTokens();
+    assertEquals("number of tokens is not 1", 1, tns.size());
+
+    boolean found = false;
+    for(Token<? extends TokenIdentifier> t: tns) {
+      System.out.println("kind="+t.getKind() + ";servic=" + t.getService() + ";str=" + t.toString());
+
+      if(t.getKind().equals(new Text("HDFS_DELEGATION_TOKEN")) &&
+          t.getService().equals(new Text(fs_addr))) {
+        found = true;
+      }
+      assertTrue("didn't find token for " + p1 ,found);
+    }
+  }
 }

src/tools/org/apache/hadoop/tools/DistCp.java

@@ -63,6 +63,7 @@ import org.apache.hadoop.mapred.RecordReader;
 import org.apache.hadoop.mapred.Reporter;
 import org.apache.hadoop.mapred.SequenceFileRecordReader;
 import org.apache.hadoop.mapreduce.JobSubmissionFiles;
+import org.apache.hadoop.mapreduce.security.TokenCache;
 import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.Tool;
@@ -623,6 +624,12 @@ public class DistCp implements Tool {
   private static void checkSrcPath(Configuration conf, List<Path> srcPaths
       ) throws IOException {
     List<IOException> rslt = new ArrayList<IOException>();
+    
+    // get tokens for all the required FileSystems..
+    Path[] ps = new Path[srcPaths.size()];
+    ps = srcPaths.toArray(ps);
+    TokenCache.obtainTokensForNamenodes(ps, conf);
+
     for (Path p : srcPaths) {
       FileSystem fs = p.getFileSystem(conf);
       if (!fs.exists(p)) {
@@ -1018,6 +1025,10 @@ public class DistCp implements Tool {
     jobConf.set(JOB_DIR_LABEL, jobDirectory.toString());
 
     FileSystem dstfs = args.dst.getFileSystem(conf);
+    
+    // get tokens for all the required FileSystems..
+    TokenCache.obtainTokensForNamenodes(new Path[] {args.dst}, conf);
+    
     boolean dstExists = dstfs.exists(args.dst);
     boolean dstIsDir = false;
     if (dstExists) {