Merge -r 752833:752834 from trunk to branch 0.20 to fix HADOOP-5327.

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

CHANGES.txt

@@ -693,6 +693,10 @@ Release 0.20.0 - Unreleased
     HADOOP-5416. Correct the shell command "fs -test" forrest doc description.
     (Ravi Phulari via szetszwo) 
 
+    HADOOP-5327. Fixed job tracker to remove files from system directory on
+    ACL check failures and also check ACLs on restart.
+    (Amar Kamat via yhemanth)
+
 Release 0.19.2 - Unreleased
 
   BUG FIXES

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

@@ -235,9 +235,9 @@ class JobInProgress {
                                                       +"/"+jobid + ".xml");
     this.localJarFile = default_job_conf.getLocalPath(JobTracker.SUBDIR
                                                       +"/"+ jobid + ".jar");
-    Path sysDir = new Path(this.jobtracker.getSystemDir());
-    FileSystem fs = sysDir.getFileSystem(default_conf);
-    jobFile = new Path(sysDir, jobid + "/job.xml");
+    Path jobDir = jobtracker.getSystemDirectoryForJob(jobId);
+    FileSystem fs = jobDir.getFileSystem(default_conf);
+    jobFile = new Path(jobDir, "job.xml");
     fs.copyToLocalFile(jobFile, localJobFile);
     conf = new JobConf(localJobFile);
     this.priority = conf.getJobPriority();
@@ -2468,7 +2468,7 @@ class JobInProgress {
       // so we remove that directory to cleanup
       // Delete temp dfs dirs created if any, like in case of 
       // speculative exn of reduces.  
-      Path tempDir = new Path(jobtracker.getSystemDir(), jobId.toString());
+      Path tempDir = jobtracker.getSystemDirectoryForJob(getJobID());
       new CleanupQueue().addToQueue(conf,tempDir); 
     } catch (IOException e) {
       LOG.warn("Error cleaning up "+profile.getJobID()+": "+e);

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

@@ -1098,31 +1098,45 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         try {
           // 1. Create the job object
           JobInProgress job = new JobInProgress(id, JobTracker.this, conf);
-          String logFileName;
-          Path jobHistoryFilePath;
 
-          // 2. Get the log file and the file path
-          logFileName = 
+          // 2. Check if the user has appropriate access
+          // Get the user group info for the job's owner
+          UserGroupInformation ugi =
+            UserGroupInformation.readFrom(job.getJobConf());
+          LOG.info("Submitting job " + id + " on behalf of user "
+                   + ugi.getUserName() + " in groups : "
+                   + StringUtils.arrayToString(ugi.getGroupNames()));
+
+          // check the access
+          try {
+            checkAccess(job, QueueManager.QueueOperation.SUBMIT_JOB, ugi);
+          } catch (Throwable t) {
+            LOG.warn("Access denied for user " + ugi.getUserName() 
+                     + " in groups : [" 
+                     + StringUtils.arrayToString(ugi.getGroupNames()) + "]");
+            throw t;
+          }
+
+          // 3. Get the log file and the file path
+          String logFileName = 
             JobHistory.JobInfo.getJobHistoryFileName(job.getJobConf(), id);
-          jobHistoryFilePath = 
+          Path jobHistoryFilePath = 
             JobHistory.JobInfo.getJobHistoryLogLocation(logFileName);
 
-          // 3. Recover the history file. This involved
+          // 4. Recover the history file. This involved
           //     - deleting file.recover if file exists
           //     - renaming file.recover to file if file doesnt exist
           // This makes sure that the (master) file exists
           JobHistory.JobInfo.recoverJobHistoryFile(job.getJobConf(), 
                                                    jobHistoryFilePath);
           
-          // 4. Cache the history file name as it costs one dfs access
+          // 5. Cache the history file name as it costs one dfs access
           jobHistoryFilenameMap.put(job.getJobID(), jobHistoryFilePath);
 
-          // 5. Sumbit the job to the jobtracker
+          // 6. Sumbit the job to the jobtracker
           addJob(id, job);
         } catch (Throwable t) {
-          LOG.warn("Failed to recover job " + id + " history details." 
-                   + " Ignoring.", t);
-          // TODO : remove job details from the system directory
+          LOG.warn("Failed to recover job " + id + " Ignoring the job.", t);
           idIter.remove();
           continue;
         }
@@ -1157,8 +1171,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
           JobHistory.parseHistoryFromFS(jobHistoryFilePath.toString(), 
                                         listener, fs);
         } catch (Throwable t) {
-          LOG.info("JobTracker failed to recover job " + pJob.getJobID() 
-                   + " from history. Ignoring it.", t);
+          LOG.info("Error reading history file of job " + pJob.getJobID() 
+                   + ". Ignoring the error and continuing.", t);
         }
 
         // 3. Close the listener
@@ -1179,7 +1193,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
           }
         } catch (Throwable t) {
           LOG.warn("Failed to delete log file (" + logFileName + ") for job " 
-                   + id + ". Ignoring it.", t);
+                   + id + ". Continuing.", t);
         }
 
         if (pJob.isComplete()) {
@@ -2763,7 +2777,14 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
     JobInProgress job = new JobInProgress(jobId, this, this.conf);
 
     // check for access
-    checkAccess(job, QueueManager.QueueOperation.SUBMIT_JOB);
+    try {
+      checkAccess(job, QueueManager.QueueOperation.SUBMIT_JOB);
+    } catch (IOException ioe) {
+       LOG.warn("Access denied for user " + job.getJobConf().getUser() 
+                + ". Ignoring job " + jobId, ioe);
+      new CleanupQueue().addToQueue(conf, getSystemDirectoryForJob(jobId));
+      throw ioe;
+    }
 
    return addJob(jobId, job); 
   }
@@ -2794,15 +2815,18 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
   }
 
   // Check whether the specified operation can be performed
-  // related to the job. If ownerAllowed is true, then an owner
-  // of the job can perform the operation irrespective of
-  // access control.
+  // related to the job.
   private void checkAccess(JobInProgress job, 
                                 QueueManager.QueueOperation oper) 
                                   throws IOException {
     // get the user group info
     UserGroupInformation ugi = UserGroupInformation.getCurrentUGI();
+    checkAccess(job, oper, ugi);
+  }
 
+  // use the passed ugi for checking the access
+  private void checkAccess(JobInProgress job, QueueManager.QueueOperation oper,
+                           UserGroupInformation ugi) throws IOException {
     // get the queue
     String queue = job.getProfile().getQueueName();
     if (!queueManager.hasAccess(queue, job, oper, ugi)) {
@@ -3148,6 +3172,11 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
     return jobs.get(jobid);
   }
 
+  // Get the job directory in system directory
+  Path getSystemDirectoryForJob(JobID id) {
+    return new Path(getSystemDir(), id.toString());
+  }
+
   /**
    * Change the run-time priority of the given job.
    * @param jobId job id

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

@@ -31,6 +31,7 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.examples.SleepJob;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.UnixUserGroupInformation;
@@ -217,6 +218,16 @@ public class TestQueueManager extends TestCase {
         assertTrue(ioe.getMessage().
             contains("cannot perform operation " +
             "SUBMIT_JOB on queue " + queue));
+        // check if the system directory gets cleaned up or not
+        JobTracker jobtracker = miniMRCluster.getJobTrackerRunner().getJobTracker();
+        Path sysDir = new Path(jobtracker.getSystemDir());
+        FileSystem fs = sysDir.getFileSystem(conf);
+        int size = fs.listStatus(sysDir).length;
+        while (size > 0) {
+          System.out.println("Waiting for the job files in sys directory to be cleaned up");
+          UtilsForTests.waitFor(100);
+          size = fs.listStatus(sysDir).length;
+        }
       }
     } finally {
       tearDownCluster();

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

@@ -29,6 +29,7 @@ import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapred.JobTracker.RecoveryManager;
+import org.apache.hadoop.security.UserGroupInformation;
 
 /**
  * Test whether the {@link RecoveryManager} is able to tolerate job-recovery 
@@ -203,6 +204,25 @@ public class TestRecoveryManager extends TestCase {
       UtilsForTests.waitFor(100);
     }
     
+    // now submit job3 with inappropriate acls
+    JobConf job3 = mr.createJobConf();
+    job3.set("hadoop.job.ugi","abc,users");
+
+    UtilsForTests.configureWaitingJobConf(job3, 
+        new Path(TEST_DIR, "input"), new Path(TEST_DIR, "output2"), 1, 0, 
+        "test-recovery-manager", signalFile, signalFile);
+    
+    // submit the job
+    RunningJob rJob3 = (new JobClient(job3)).submitJob(job3);
+    LOG.info("Submitted job " + rJob3.getID() + " with different user");
+    
+    jip = jobtracker.getJob(rJob3.getID());
+
+    while (!jip.inited()) {
+      LOG.info("Waiting for job " + jip.getJobID() + " to be inited");
+      UtilsForTests.waitFor(100);
+    }
+
     // kill the jobtracker
     LOG.info("Stopping jobtracker");
     mr.stopJobTracker();
@@ -212,9 +232,15 @@ public class TestRecoveryManager extends TestCase {
                                       true);
     mr.getJobTrackerConf().setInt("mapred.jobtracker.maxtasks.per.job", 25);
     
+    mr.getJobTrackerConf().setBoolean("mapred.acls.enabled" , true);
+    UserGroupInformation ugi = UserGroupInformation.readFrom(job1);
+    mr.getJobTrackerConf().set("mapred.queue.default.acl-submit-job", 
+                               ugi.getUserName());
+
     // start the jobtracker
     LOG.info("Starting jobtracker");
     mr.startJobTracker();
+    UtilsForTests.waitForJobTracker(jc);
     
     jobtracker = mr.getJobTrackerRunner().getJobTracker();
     
@@ -222,6 +248,17 @@ public class TestRecoveryManager extends TestCase {
     assertEquals("Recovery manager failed to tolerate job failures",
                  2, jobtracker.getAllJobs().length);
     
+    // check if the job#1 has failed
+    JobStatus status = jobtracker.getJobStatus(rJob1.getID());
+    assertEquals("Faulty job not failed", 
+                 JobStatus.FAILED, status.getRunState());
+    
+    jip = jobtracker.getJob(rJob2.getID());
+    assertFalse("Job should be running", jip.isComplete());
+    
+    status = jobtracker.getJobStatus(rJob3.getID());
+    assertNull("Job should be missing", status);
+    
     mr.shutdown();
   }
 }