commit 0d2feca8e07e5003e1f133df55355ff59c49c05c

Author: Devaraj Das <ddas@yahoo-inc.com>
Date:   Wed Oct 6 22:02:35 2010 -0700

    Fixed a problem to do with DefaultTaskController trying to authorize for the distributed cache update


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

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

@@ -163,13 +163,14 @@ public class DefaultTaskController extends TaskController {
    * downloaded
    * @param taskTracker the connection to the task tracker
    * @throws IOException
+   * @throws InterruptedException
    */
   @Override
   public void initializeJob(String user, String jobid, 
                             Path credentials, Path jobConf, 
                             TaskUmbilicalProtocol taskTracker,
                             InetSocketAddress ttAddr
-                            ) throws IOException {
+                            ) throws IOException, InterruptedException {
     final LocalDirAllocator lDirAlloc = allocator;
     FileSystem localFs = FileSystem.getLocal(getConf());
     JobLocalizer localizer = new JobLocalizer((JobConf)getConf(), user, jobid);

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

@@ -358,15 +358,16 @@ public class JobLocalizer {
 
   public void localizeJobFiles(JobID jobid, JobConf jConf,
       Path localJobTokenFile, TaskUmbilicalProtocol taskTracker)
-      throws IOException {
+      throws IOException, InterruptedException {
     localizeJobFiles(jobid, jConf,
         lDirAlloc.getLocalPathForWrite(JOBCONF, ttConf), localJobTokenFile,
         taskTracker);
   }
 
-  public void localizeJobFiles(JobID jobid, JobConf jConf,
+  public void localizeJobFiles(final JobID jobid, JobConf jConf,
       Path localJobFile, Path localJobTokenFile,
-      TaskUmbilicalProtocol taskTracker) throws IOException {
+      final TaskUmbilicalProtocol taskTracker) 
+  throws IOException, InterruptedException {
     // Download the job.jar for this job from the system FS
     localizeJobJarFile(jConf);
 
@@ -379,9 +380,17 @@ public class JobLocalizer {
     TaskTracker.resetNumTasksPerJvm(jConf);
 
     //setup the distributed cache
-    long[] sizes = downloadPrivateCache(jConf);
+    final long[] sizes = downloadPrivateCache(jConf);
     if (sizes != null) {
-      taskTracker.updatePrivateDistributedCacheSizes(jobid, sizes);
+      UserGroupInformation ugi = 
+        UserGroupInformation.createRemoteUser(jobid.toString());
+      ugi.doAs(new PrivilegedExceptionAction<Object>() { 
+        public Object run() throws IOException {
+          taskTracker.updatePrivateDistributedCacheSizes(jobid, sizes);
+          return null;
+        }
+      });
+      
     }
 
     // Create job-acls.xml file in job userlog dir and write the needed
@@ -449,7 +458,8 @@ public class JobLocalizer {
   }
 
   public int runSetup(String user, String jobid, Path localJobTokenFile,
-                      TaskUmbilicalProtocol taskTracker) throws IOException {
+                      TaskUmbilicalProtocol taskTracker) throws IOException, 
+                      InterruptedException {
     // load user credentials, configuration
     // ASSUME
     // let $x = $mapred.local.dir

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

@@ -92,12 +92,13 @@ public abstract class TaskController implements Configurable {
    * @param taskTracker the connection the task tracker
    * @param ttAddr the tasktracker's RPC address
    * @throws IOException
+   * @throws InterruptedException
    */
   public abstract void initializeJob(String user, String jobid, 
                                      Path credentials, Path jobConf,
                                      TaskUmbilicalProtocol taskTracker,
                                      InetSocketAddress ttAddr) 
-  throws IOException;
+  throws IOException, InterruptedException;
   
   /**
    * Create all of the directories for the task and launches the child jvm.

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

@@ -1027,7 +1027,7 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
       * To support potential authenticated HDFS accesses, we need the tokens
       */
     rjob.ugi.doAs(new PrivilegedExceptionAction<Object>() {
-      public Object run() throws IOException {
+      public Object run() throws IOException, InterruptedException {
         try {
           final JobConf localJobConf = new JobConf(localJobFile);
           // Setup the public distributed cache
@@ -1057,6 +1057,10 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
           LOG.warn("Exception while localization " + 
               StringUtils.stringifyException(e));
           throw e;
+        } catch (InterruptedException ie) {
+          LOG.warn("Exception while localization " + 
+              StringUtils.stringifyException(ie));
+          throw ie;
         }
         return null;
       }