commit 8aebad0a6c44d2241e71f589a8ff673dbf0f6373

Author: Balaji Rajagopalan <balajirg@yahoo-inc.com>
Date:   Tue Jul 27 16:51:58 2010 -0700

    MAPREDUCE-1971 from https://issues.apache.org/jira/secure/attachment/12450589/concurrent_exp_y20_1.patch


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

src/test/system/aop/org/apache/hadoop/mapred/JTProtocolAspect.aj

@@ -113,4 +113,8 @@ public aspect JTProtocolAspect {
       throws Exception {
     return -1;
   }
+  
+  public void JTProtocol.accessHistoryData(JobID jobId) throws Exception{
+    
+  }
 }

src/test/system/aop/org/apache/hadoop/mapred/JobTrackerAspect.aj

@@ -25,8 +25,11 @@ import java.util.LinkedList;
 import java.util.Set;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.JobHistory.Keys;
 import org.apache.hadoop.mapred.JobTracker.RetireJobInfo;
 import org.apache.hadoop.mapred.Counters;
 import org.apache.hadoop.mapreduce.JobID;
@@ -517,4 +520,23 @@ public privileged aspect JobTrackerAspect {
       throws Exception {
     return (getNextHeartbeatInterval());
   }
+  
+  //access the job data the method only does a get on read-only data
+  //it does not return anything purposely, since the test case
+  //does not require this but this can be extended in future
+  public void JobTracker.accessHistoryData(JobID id) throws Exception {
+    String location = getJobHistoryLocationForRetiredJob(id);
+    Path logFile = new Path(location);
+    FileSystem fs = logFile.getFileSystem(getConf());
+    JobHistory.JobInfo jobInfo  = new JobHistory.JobInfo(id.toString());
+    DefaultJobHistoryParser.parseJobTasks(location,
+        jobInfo, fs);
+    //Now read the info so two threads can access the info at the
+    //same time from client side
+    LOG.info("user " +jobInfo.get(Keys.USER));
+    LOG.info("jobname "+jobInfo.get(Keys.JOBNAME));
+    jobInfo.get(Keys.JOBCONF);
+    jobInfo.getJobACLs();
+  }
+  
 }

src/test/system/java/org/apache/hadoop/mapred/TestCMExceptionDuringRunJob.java

@@ -0,0 +1,114 @@
+package org.apache.hadoop.mapred;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.examples.SleepJob;
+import org.apache.hadoop.mapreduce.test.system.FinishTaskControlAction;
+import org.apache.hadoop.mapreduce.test.system.JTProtocol;
+import org.apache.hadoop.mapreduce.test.system.JobInfo;
+import org.apache.hadoop.mapreduce.test.system.MRCluster;
+import org.apache.hadoop.mapreduce.test.system.MRCluster.Role;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.AfterClass;
+import org.junit.Test;
+import org.apache.commons.logging.LogFactory;
+import org.apache.commons.logging.Log;
+
+public class TestCMExceptionDuringRunJob {
+  
+  private static MRCluster cluster = null;
+  static final Log LOG = LogFactory.getLog(TestCMExceptionDuringRunJob.class);
+  private JTProtocol remoteJTClient=null;
+  private Configuration conf =null;
+  @BeforeClass
+  public static void setUp() throws java.lang.Exception {
+    String [] expExcludeList = new String[2];
+    expExcludeList[0] = "java.net.ConnectException";
+    expExcludeList[1] = "java.io.IOException";
+    cluster = MRCluster.createCluster(new Configuration());
+    cluster.setExcludeExpList(expExcludeList);
+    cluster.setUp();
+  }
+  
+  /**
+   * The objective of the test is, when the user accesses the retired job data
+   * or the running job data simultaneously in different threads there
+   * is no concurrent modification exceptions gets thrown
+   */
+  @Test
+  public void testNoCMExcepRunningJob() throws Exception {
+    
+    remoteJTClient = cluster.getJTClient().getProxy();
+    
+    conf = remoteJTClient.getDaemonConf();
+    //set interval to 30 secs, check to 10 secs, check determines the frequency
+    //of the jobtracker thread that will check for retired jobs, and interval
+    //will determine how long it will take before a job retires.
+    conf.setInt("mapred.jobtracker.retirejob.interval",30*1000);
+    conf.setInt("mapred.jobtracker.retirejob.check",10*1000);
+    cluster.restartDaemonWithNewConfig(cluster.getJTClient(), "mapred-site.xml",
+        conf, Role.JT);
+    JobID jobid1 = runSleepJob(true);
+    JobID jobid2 = runSleepJob(true);
+    JobID jobid3 = runSleepJob(false);
+    //Waiting for a minute for the job to retire
+    UtilsForTests.waitFor(60*1000);
+    RunAccessHistoryData access1 = new RunAccessHistoryData(jobid1);
+    RunAccessHistoryData access2 = new RunAccessHistoryData(jobid2);
+    new Thread(access1).start();
+    new Thread(access2).start();
+    remoteJTClient.getJobSummaryInfo(jobid3);
+    cluster.signalAllTasks(jobid3);
+    cluster.getJTClient().isJobStopped(jobid3);
+    cluster.restart(cluster.getJTClient(), Role.JT);
+  }
+  
+  
+  public class RunAccessHistoryData implements Runnable {
+    private  JobID jobId = null;
+    
+    public RunAccessHistoryData (JobID jobId) {
+      this.jobId =jobId; 
+    }
+    
+     public void run () {
+       try {
+         remoteJTClient.accessHistoryData(jobId);
+       }
+       catch (Exception ex) {
+         ex.printStackTrace();
+       }
+     }
+  }
+
+  
+  public JobID runSleepJob(boolean signalJob) throws Exception{
+    SleepJob job = new SleepJob();
+    job.setConf(conf);
+    conf = job.setupJobConf(5, 1, 100, 5, 100, 5);
+    JobConf jconf = new JobConf(conf);
+    //Controls the job till all verification is done 
+    FinishTaskControlAction.configureControlActionForJob(conf);
+    //Submitting the job
+    RunningJob rJob = cluster.getJTClient().getClient().submitJob(jconf);
+    JobID jobId = rJob.getID();
+    JobInfo jInfo = remoteJTClient.getJobInfo(jobId);
+    LOG.info("jInfo is :" + jInfo);
+    boolean jobStarted = cluster.getJTClient().isJobStarted(jobId);
+    Assert.assertTrue("Job has not started even after a minute", 
+        jobStarted );
+      
+    if(signalJob) {
+      cluster.signalAllTasks(jobId);
+      Assert.assertTrue("Job has not stopped yet",
+          cluster.getJTClient().isJobStopped(jobId));
+    }
+    return jobId;
+  }
+  
+  @AfterClass
+  public static void tearDown() throws Exception {    
+    cluster.tearDown();
+  }
+
+}

src/test/system/java/org/apache/hadoop/mapreduce/test/system/JTProtocol.java

@@ -185,5 +185,16 @@ public interface JTProtocol extends DaemonProtocol {
     * This gets the value of all task trackers windows in the tasktracker page.
     */
    public int getTaskTrackerHeartbeatInterval() throws Exception;
+   
+
+   
+   /**
+    * The method with access the history data in JobTracker, it will only do a 
+    * read on the data structure none is returned, this is used to verify a
+    * bug with simultaneously accessing history data
+    * @param jobId
+    * @throws Exception
+    */
+   public void accessHistoryData(JobID jobId) throws Exception;
 
 }