HADOOP-322. Add a job control utility. Contributed by Runping.

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@431715 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -46,6 +46,11 @@ Trunk (unreleased changes)
     critical percentage of the datanodes are unavailable.
     (Konstantin Shvachko via cutting)
 
+11. HADOOP-322.  Add a job control utility.  This permits one to
+    specify job interdependencies.  Each job is submitted only after
+    the jobs it depends on have successfully completed.
+    (Runping Qi via cutting)
+
 
 Release 0.5.0 - 2006-08-04
 

src/java/org/apache/hadoop/mapred/jobcontrol/Job.java

@@ -0,0 +1,331 @@
+/**
+ * Copyright 2005 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mapred.jobcontrol;
+
+
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.util.StringUtils;
+
+import java.util.ArrayList;
+import java.io.IOException;
+
+/** This class encapsulates a MapReduce job and its dependency. It monitors 
+ *  the states of the depending jobs and updates the state of this job.
+ *  A job stats in the WAITING state. If it does not have any deoending jobs, or
+ *  all of the depending jobs are in SUCCESS state, then the job state will become
+ *  READY. If any depending jobs fail, the job will fail too. 
+ *  When in READY state, the job can be submitted to Hadoop for execution, with
+ *  the state changing into RUNNING state. From RUNNING state, the job can get into 
+ *  SUCCESS or FAILED state, depending the status of the jon execution.
+ *  
+ */
+
+public class Job {
+
+	// A job will be in one of the following states
+	final public static int SUCCESS = 0;
+	final public static int WAITING = 1;
+	final public static int RUNNING = 2;
+	final public static int READY = 3;
+	final public static int FAILED = 4;
+	final public static int DEPENDENT_FAILED = 5;
+	
+	
+	private JobConf theJobConf;
+	private int state;
+	private String jobID; 		// assigned and used by JobControl class
+	private String mapredJobID; // the job ID assigned by map/reduce
+	private String jobName;		// external name, assigned/used by client app
+	private String message;		// some info for human consumption, 
+								// e.g. the reason why the job failed
+	private ArrayList dependingJobs;	// the jobs the current job depends on
+	
+	private JobClient jc = null;		// the map reduce job client
+	
+    /** 
+     * Construct a job.
+     * @param jobConf a mapred job configuration representing a job to be executed.
+     * @param dependingJobs an array of jobs the current job depends on
+     */
+    public Job(JobConf jobConf, ArrayList dependingJobs) throws IOException {
+       	this.theJobConf = jobConf;
+		this.dependingJobs = dependingJobs;
+		this.state = Job.WAITING;
+		this.jobID = "unassigned";
+		this.mapredJobID = "unassigned";
+		this.jobName = "unassigned";
+		this.message = "just initialized";
+		this.jc = new JobClient(jobConf);
+	}
+	
+	public String toString() {
+		StringBuffer sb = new StringBuffer();
+		sb.append("job name:\t").append(this.jobName).append("\n");
+		sb.append("job id:\t").append(this.jobID).append("\n");
+		sb.append("job state:\t").append(this.state).append("\n");
+		sb.append("job mapred id:\t").append(this.mapredJobID).append("\n");
+		sb.append("job message:\t").append(this.message).append("\n");
+		
+		if (this.dependingJobs == null) {
+			sb.append("job has no depending job:\t").append("\n");
+		} else {
+			sb.append("job has ").append(this.dependingJobs.size()).append(" dependeng jobs:\n");
+			for (int i = 0; i < this.dependingJobs.size(); i++) {
+				sb.append("\t depending job ").append(i).append(":\t");
+				sb.append(((Job) this.dependingJobs.get(i)).getJobName()).append("\n");
+			}
+		}
+		return sb.toString();
+	}
+	
+	/**
+	 * @return the job name of this job
+	 */
+	public String getJobName() {
+		return this.jobName;
+	}
+	
+	/**
+	 * Set the job name for  this job.
+	 * @param jobName the job name
+	 */
+	public void setJobName(String jobName) {
+		this.jobName = jobName;
+	}
+	
+	/**
+	 * @return the job ID of this job
+	 */
+	public String getJobID() {
+		return this.jobID;
+	}
+	
+	/**
+	 * Set the job ID for  this job.
+	 * @param id the job ID
+	 */
+	public void setJobID(String id) {
+		this.jobID = id;
+	}
+	
+	/**
+	 * @return the mapred ID of this job
+	 */
+	public String getMapredJobID() {
+		return this.mapredJobID;
+	}
+	
+	/**
+	 * Set the mapred ID for this job.
+	 * @param mapredJobID the mapred job ID for this job.
+	 */
+	public void setMapredJobID(String mapredJobID) {
+		this.jobID = mapredJobID;
+	}
+	
+	/**
+	 * @return the mapred job conf of this job
+	 */
+	public JobConf getJobConf() {
+		return this.theJobConf;
+	}
+	
+
+	/**
+	 * Set the mapred job conf for this job.
+	 * @param jobConf the mapred job conf for this job.
+	 */
+	public void setJobConf(JobConf jobConf) {
+		this.theJobConf = jobConf;
+	}
+	
+	/**
+	 * @return the state of this job
+	 */
+	public int getState() {
+		return this.state;
+	}
+	
+	/**
+	 * Set the state for this job.
+	 * @param state the new state for this job.
+	 */
+	public void setState(int state) {
+		this.state = state;
+	}
+	
+	/**
+	 * @return the message of this job
+	 */
+	public String getMessage() {
+		return this.message;
+	}
+	
+	/**
+	 * Set the message for this job.
+	 * @param message the message for this job.
+	 */
+	public void setMessage(String message) {
+		this.message = message;
+	}
+	
+	/**
+	 * @return the depending jobs of this job
+	 */
+	public ArrayList getDependingJobs() {
+		return this.dependingJobs;
+	}
+	
+	/**
+	 * @return true if this job is in a complete state
+	 */
+	public boolean isCompleted() {
+		return this.state == Job.FAILED || 
+		       this.state == Job.DEPENDENT_FAILED ||
+		       this.state == Job.SUCCESS;
+	}
+	
+	/**
+	 * @return true if this job is in READY state
+	 */
+	public boolean isReady() {
+		return this.state == Job.READY;
+	}
+	
+	/**
+	 * Check the state of this running job. The state may 
+	 * remain the same, become SUCCESS or FAILED.
+	 */
+	private void checkRunningState() {
+		RunningJob running = null;
+		try {
+			running = jc.getJob(this.mapredJobID);
+			if (running.isComplete()) {
+				if (running.isSuccessful()) {
+					this.state = Job.SUCCESS;
+				} else {
+					this.state = Job.FAILED;
+					this.message = "Job failed!";
+					try {
+						running.killJob();
+					} catch (IOException e1) {
+
+					}
+					try {
+						this.jc.close();
+					} catch (IOException e2) {
+
+					}
+				}
+			}
+
+		} catch (IOException ioe) {
+			this.state = Job.FAILED;
+			this.message = StringUtils.stringifyException(ioe);
+			try {
+				running.killJob();
+			} catch (IOException e1) {
+
+			}
+			try {
+				this.jc.close();
+			} catch (IOException e1) {
+
+			}
+		}
+	}
+	
+	/**
+	 * Check and update the state of this job. The state changes  
+	 * depending on its current state and the states of the depending jobs.
+	 */
+	public int checkState() {
+		if (this.state == Job.RUNNING) {
+			checkRunningState();
+		}
+		if (this.state != Job.WAITING) {
+			return this.state;
+		}
+		if (this.dependingJobs == null || this.dependingJobs.size() == 0) {
+			this.state = Job.READY;
+			return this.state;
+		}
+		Job pred = null;
+		int n = this.dependingJobs.size();
+		for (int i = 0; i < n; i++) {
+			pred = (Job) this.dependingJobs.get(i);
+			int s = pred.checkState();
+			if (s == Job.WAITING || s == Job.READY || s == Job.RUNNING) {
+				break; // a pred is still not completed, continue in WAITING
+						// state
+			}
+			if (s == Job.FAILED || s == Job.DEPENDENT_FAILED) {
+				this.state = Job.DEPENDENT_FAILED;
+				this.message = "depending job " + i + " with jobID "
+						+ pred.getJobID() + " failed. " + pred.getMessage();
+				break;
+			}
+			// pred must be in success state
+			if (i == n - 1) {
+				this.state = Job.READY;
+			}
+		}
+
+		return this.state;
+	}
+	
+    /**
+     * Submit this job to mapred. The state becomes RUNNING if submission 
+     * is successful, FAILED otherwise.  
+     */
+    public void submit() {
+        try {
+            if (theJobConf.getBoolean("create.empty.dir.if.nonexist", false)) {
+                FileSystem fs = FileSystem.get(theJobConf);
+                Path inputPaths[] = theJobConf.getInputPaths();
+                for (int i = 0; i < inputPaths.length; i++) {
+                    if (!fs.exists(inputPaths[i])) {
+                        try {
+                            fs.mkdirs(inputPaths[i]);
+                        } catch (IOException e) {
+
+                        }
+                    }
+                }
+            }
+            RunningJob running = jc.submitJob(theJobConf);
+            this.mapredJobID = running.getJobID();
+            this.state = Job.RUNNING;
+        } catch (IOException ioe) {
+            this.state = Job.FAILED;
+            this.message = StringUtils.stringifyException(ioe);
+        }
+    }
+	
+	/**
+	 * @param args
+	 */
+	public static void main(String[] args) {
+		// TODO Auto-generated method stub
+
+	}
+
+}

src/java/org/apache/hadoop/mapred/jobcontrol/JobControl.java

@@ -0,0 +1,299 @@
+/**
+ * Copyright 2005 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mapred.jobcontrol;
+
+import java.util.ArrayList;
+import java.util.Hashtable;
+import java.util.Iterator;
+
+/** This class encapsulates a set of MapReduce jobs and its dependency. It tracks 
+ *  the states of the jobs by placing them into different tables according to their 
+ *  states. 
+ *  
+ *  This class provides APIs for the client app to add a job to the group and to get 
+ *  the jobs in the group in different states. When a 
+ *  job is added, an ID unique to the group is assigned to the job. 
+ *  
+ *  This class has a thread that submits jobs when they become ready, monitors the
+ *  states of the running jobs, and updates the states of jobs based on the state changes 
+ *  of their depending jobs states. The class provides APIs for suspending/resuming
+ *  the thread,and for stopping the thread.
+ *  
+ */
+public class JobControl implements Runnable{
+
+	// The thread can be in one of the following state
+	private static final int RUNNING = 0;
+	private static final int SUSPENDED = 1;
+	private static final int STOPPED = 2;
+	private static final int STOPPING = 3;
+	private static final int READY = 4;
+	
+	private int runnerState;			// the thread state
+	
+	private Hashtable waitingJobs;
+	private Hashtable readyJobs;
+	private Hashtable runningJobs;
+	private Hashtable successfulJobs;
+	private Hashtable failedJobs;
+	
+	private long nextJobID;
+	private String groupName;
+	
+	/** 
+	 * Construct a job control for a group of jobs.
+	 * @param groupName a name identifying this group
+	 */
+	public JobControl(String groupName) {
+		this.waitingJobs = new Hashtable();
+		this.readyJobs = new Hashtable();
+		this.runningJobs = new Hashtable();
+		this.successfulJobs = new Hashtable();
+		this.failedJobs = new Hashtable();
+		this.nextJobID = -1;
+		this.groupName = groupName;
+		this.runnerState = JobControl.READY;
+		
+	}
+	
+	private static ArrayList toArrayList(Hashtable jobs) {
+		ArrayList retv = new ArrayList();
+		Iterator iter = jobs.values().iterator();
+		while (iter.hasNext()) {
+			retv.add(iter.next());
+		}
+		return retv;
+	}
+	
+	/**
+	 * @return the jobs in the waiting state
+	 */
+	public ArrayList getWaitingJobs() {
+		return JobControl.toArrayList(this.waitingJobs);
+	}
+	
+	/**
+	 * @return the jobs in the running state
+	 */
+	public ArrayList getRunningJobs() {
+		return JobControl.toArrayList(this.runningJobs);
+	}
+	
+	/**
+	 * @return the jobs in the ready state
+	 */
+	public ArrayList getReadyJobs() {
+		return JobControl.toArrayList(this.readyJobs);
+	}
+	
+	/**
+	 * @return the jobs in the success state
+	 */
+	public ArrayList getSuccessfulJobs() {
+		return JobControl.toArrayList(this.successfulJobs);
+	}
+	
+	public ArrayList getFailedJobs() {
+		return JobControl.toArrayList(this.failedJobs);
+	}
+	
+	private String getNextJobID() {
+		nextJobID += 1;
+		return this.groupName + this.nextJobID;
+	}
+	
+	private static void addToQueue(Job aJob, Hashtable queue) {
+		synchronized(queue) {
+			queue.put(aJob.getJobID(), aJob);
+		}		
+	}
+	
+	private void addToQueue(Job aJob) {
+		Hashtable queue = getQueue(aJob.getState());
+		addToQueue(aJob, queue);	
+	}
+	
+	private Hashtable getQueue(int state) {
+		Hashtable retv = null;
+		if (state == Job.WAITING) {
+			retv = this.waitingJobs;
+		} else if (state == Job.READY) {
+			retv = this.readyJobs;
+		} else if (state == Job.RUNNING) {
+			retv = this.runningJobs;
+		} else if (state == Job.SUCCESS) {
+			retv = this.successfulJobs;
+		} else if (state == Job.FAILED || state == Job.DEPENDENT_FAILED) {
+			retv = this.failedJobs;
+		} 
+		return retv;
+			
+	}
+
+	/**
+	 * Add a new job.
+	 * @param aJob the the new job
+	 */
+	synchronized public String addJob(Job aJob) {
+		String id = this.getNextJobID();
+		aJob.setJobID(id);
+		aJob.setState(Job.WAITING);
+		this.addToQueue(aJob);
+		return id;	
+	}
+	
+	/**
+	 * @param args
+	 */
+	public static void main(String[] args) {
+		// TODO Auto-generated method stub
+
+	}
+
+	/**
+	 * @return the thread state
+	 */
+	public int getState() {
+		return this.runnerState;
+	}
+	
+	/**
+	 * set the thread state to STOPPING so that the 
+	 * thread will stop when it wakes up.
+	 */
+	public void stop() {
+		this.runnerState = JobControl.STOPPING;
+	}
+	
+	/**
+	 * suspend the running thread
+	 */
+	public void suspend () {
+		if (this.runnerState == JobControl.RUNNING) {
+			this.runnerState = JobControl.SUSPENDED;
+		}
+	}
+	
+	/**
+	 * resume the suspended thread
+	 */
+	public void resume () {
+		if (this.runnerState == JobControl.SUSPENDED) {
+			this.runnerState = JobControl.RUNNING;
+		}
+	}
+	
+	synchronized private void checkRunningJobs() {
+		
+		Hashtable oldJobs = null;
+		oldJobs = this.runningJobs;
+		this.runningJobs = new Hashtable();
+		
+		Iterator jobs = oldJobs.values().iterator();
+		while (jobs.hasNext()) {
+			Job nextJob = (Job)jobs.next();
+			int state = nextJob.checkState();
+			/*
+			if (state != Job.RUNNING) {
+				System.out.println("The state of the running job " +
+					nextJob.getJobName() + " has changed to: " + nextJob.getState());
+			}
+			*/
+			this.addToQueue(nextJob);
+		}
+	}
+	
+	synchronized private void checkWaitingJobs() {
+		Hashtable oldJobs = null;
+		oldJobs = this.waitingJobs;
+		this.waitingJobs = new Hashtable();
+		
+		Iterator jobs = oldJobs.values().iterator();
+		while (jobs.hasNext()) {
+			Job nextJob = (Job)jobs.next();
+			int state = nextJob.checkState();
+			/*
+			if (state != Job.WAITING) {
+				System.out.println("The state of the waiting job " +
+					nextJob.getJobName() + " has changed to: " + nextJob.getState());
+			}
+			*/
+			this.addToQueue(nextJob);
+		}
+	}
+	
+	synchronized private void startReadyJobs() {
+		Hashtable oldJobs = null;
+		oldJobs = this.readyJobs;
+		this.readyJobs = new Hashtable();
+		
+		Iterator jobs = oldJobs.values().iterator();
+		while (jobs.hasNext()) {
+			Job nextJob = (Job)jobs.next();
+			//System.out.println("Job to submit to Hadoop: " + nextJob.getJobName());
+			nextJob.submit();
+			//System.out.println("Hadoop ID: " + nextJob.getMapredJobID());
+			this.addToQueue(nextJob);
+		}	
+	}
+	
+	synchronized public boolean allFinished() {
+		return this.waitingJobs.size() == 0 &&
+			this.readyJobs.size() == 0 &&
+			this.runningJobs.size() == 0;
+	}
+	
+	/**
+	 *  The main loop for the thread.
+	 *  The loop does the following:
+	 *  	Check the states of the running jobs
+	 *  	Update the states of waiting jobs
+	 *  	Submit the jobs in ready state
+	 */
+	public void run() {
+		this.runnerState = JobControl.RUNNING;
+		while (true) {
+			while (this.runnerState == JobControl.SUSPENDED) {
+				try {
+					Thread.sleep(5000);
+				}
+				catch (Exception e) {
+					
+				}
+			}
+			checkRunningJobs();	
+			checkWaitingJobs();		
+			startReadyJobs();		
+			if (this.runnerState != JobControl.RUNNING && 
+					this.runnerState != JobControl.SUSPENDED) {
+				break;
+			}
+			try {
+				Thread.sleep(5000);
+			}
+			catch (Exception e) {
+				
+			}
+			if (this.runnerState != JobControl.RUNNING && 
+					this.runnerState != JobControl.SUSPENDED) {
+				break;
+			}
+		}
+		this.runnerState = JobControl.STOPPED;
+	}
+
+}

src/java/org/apache/hadoop/mapred/jobcontrol/package.html

@@ -0,0 +1,7 @@
+<html>
+<body>
+
+<p>Utilities for managing dependent jobs.</p>
+
+</body>
+</html>

src/test/org/apache/hadoop/mapred/jobcontrol/TestJobControl.java

@@ -0,0 +1,297 @@
+/**
+ * Copyright 2005 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mapred.jobcontrol;
+
+import java.io.IOException;
+import java.text.NumberFormat;
+import java.util.Iterator;
+import java.util.ArrayList;
+import java.util.Random;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+
+/**
+ * This class performs unit test for Job/JobControl classes.
+ *  
+ * @author runping
+ *
+ */
+public class TestJobControl extends junit.framework.TestCase {
+
+    private static NumberFormat idFormat = NumberFormat.getInstance();
+    static {
+        idFormat.setMinimumIntegerDigits(4);
+        idFormat.setGroupingUsed(false);
+    }
+
+    static private Random rand = new Random();
+
+    private static void cleanData(FileSystem fs, Path dirPath)
+            throws IOException {
+        fs.delete(dirPath);
+    }
+
+    private static String generateRandomWord() {
+        return idFormat.format(rand.nextLong());
+    }
+
+    private static String generateRandomLine() {
+        long r = rand.nextLong() % 7;
+        long n = r + 20;
+        StringBuffer sb = new StringBuffer();
+        for (int i = 0; i < n; i++) {
+            sb.append(generateRandomWord()).append(" ");
+        }
+        sb.append("\n");
+        return sb.toString();
+    }
+
+    private static void generateData(FileSystem fs, Path dirPath)
+            throws IOException {
+        FSDataOutputStream out = fs.create(new Path(dirPath, "data.txt"));
+        for (int i = 0; i < 100000; i++) {
+            String line = TestJobControl.generateRandomLine();
+            out.write(line.getBytes("UTF-8"));
+        }
+        out.close();
+    }
+
+    public static class DataCopy extends MapReduceBase implements Mapper,
+            Reducer {
+        public void map(WritableComparable key, Writable value,
+                OutputCollector output, Reporter reporter) throws IOException {
+            output.collect(new Text(key.toString()), value);
+        }
+
+        public void reduce(WritableComparable key, Iterator values,
+                OutputCollector output, Reporter reporter) throws IOException {
+            Text dumbKey = new Text("");
+            while (values.hasNext()) {
+                Text data = (Text) values.next();
+                output.collect(dumbKey, data);
+            }
+        }
+    }
+
+    private static JobConf createCopyJob(ArrayList indirs, Path outdir)
+            throws Exception {
+
+        Configuration defaults = new Configuration();
+        JobConf theJob = new JobConf(defaults, TestJobControl.class);
+        theJob.setJobName("DataMoveJob");
+
+        theJob.setInputPath((Path) indirs.get(0));
+        if (indirs.size() > 1) {
+            for (int i = 1; i < indirs.size(); i++) {
+                theJob.addInputPath((Path) indirs.get(i));
+            }
+        }
+        theJob.setMapperClass(DataCopy.class);
+        theJob.setOutputPath(outdir);
+        theJob.setOutputKeyClass(Text.class);
+        theJob.setOutputValueClass(Text.class);
+        theJob.setReducerClass(DataCopy.class);
+        theJob.setNumMapTasks(12);
+        theJob.setNumReduceTasks(4);
+        return theJob;
+    }
+
+    /**
+     * This is a main function for testing JobControl class.
+     * It first cleans all the dirs it will use. Then it generates some random text
+     * data in TestJobControlData/indir. Then it creates 4 jobs: 
+     *      Job 1: copy data from indir to outdir_1
+     *      Job 2: copy data from indir to outdir_2
+     *      Job 3: copy data from outdir_1 and outdir_2 to outdir_3
+     *      Job 4: copy data from outdir to outdir_4
+     * The jobs 1 and 2 have no dependency. The job 3 depends on jobs 1 and 2.
+     * The job 4 depends on job 3.
+     * 
+     * Then it creates a JobControl object and add the 4 jobs to the JobControl object.
+     * Finally, it creates a thread to run the JobControl object and monitors/reports
+     * the job states.
+     * 
+     * @param args
+     */
+    public static void doJobControlTest() throws Exception {
+        
+        Configuration defaults = new Configuration();
+        FileSystem fs = FileSystem.get(defaults);
+        Path rootDataDir = new Path(System.getProperty("test.build.data", "."), "TestJobControlData");
+        Path indir = new Path(rootDataDir, "indir");
+        Path outdir_1 = new Path(rootDataDir, "outdir_1");
+        Path outdir_2 = new Path(rootDataDir, "outdir_2");
+        Path outdir_3 = new Path(rootDataDir, "outdir_3");
+        Path outdir_4 = new Path(rootDataDir, "outdir_4");
+
+        cleanData(fs, indir);
+        generateData(fs, indir);
+
+        cleanData(fs, outdir_1);
+        cleanData(fs, outdir_2);
+        cleanData(fs, outdir_3);
+        cleanData(fs, outdir_4);
+
+        ArrayList dependingJobs = null;
+
+        ArrayList inPaths_1 = new ArrayList();
+        inPaths_1.add(indir);
+        JobConf jobConf_1 = createCopyJob(inPaths_1, outdir_1);
+        Job job_1 = new Job(jobConf_1, dependingJobs);
+        ArrayList inPaths_2 = new ArrayList();
+        inPaths_2.add(indir);
+        JobConf jobConf_2 = createCopyJob(inPaths_2, outdir_2);
+        Job job_2 = new Job(jobConf_2, dependingJobs);
+
+        ArrayList inPaths_3 = new ArrayList();
+        inPaths_3.add(outdir_1);
+        inPaths_3.add(outdir_2);
+        JobConf jobConf_3 = createCopyJob(inPaths_3, outdir_3);
+        dependingJobs = new ArrayList();
+        dependingJobs.add(job_1);
+        dependingJobs.add(job_2);
+        Job job_3 = new Job(jobConf_3, dependingJobs);
+
+        ArrayList inPaths_4 = new ArrayList();
+        inPaths_4.add(outdir_3);
+        JobConf jobConf_4 = createCopyJob(inPaths_4, outdir_4);
+        dependingJobs = new ArrayList();
+        dependingJobs.add(job_3);
+        Job job_4 = new Job(jobConf_4, dependingJobs);
+
+        JobControl theControl = new JobControl("Test");
+        theControl.addJob(job_1);
+        theControl.addJob(job_2);
+        theControl.addJob(job_3);
+        theControl.addJob(job_4);
+
+        Thread theController = new Thread(theControl);
+        theController.start();
+        while (!theControl.allFinished()) {
+
+            System.out.println("Jobs in waiting state: "
+                    + theControl.getWaitingJobs().size());
+            System.out.println("Jobs in ready state: "
+                    + theControl.getReadyJobs().size());
+            System.out.println("Jobs in running state: "
+                    + theControl.getRunningJobs().size());
+            System.out.println("Jobs in success state: "
+                    + theControl.getSuccessfulJobs().size());
+            System.out.println("Jobs in failed state: "
+                    + theControl.getFailedJobs().size());
+            System.out.println("\n");
+
+            try {
+                Thread.sleep(5000);
+            } catch (Exception e) {
+
+            }
+        }
+        System.out.println("Jobs are all done???");
+        System.out.println("Jobs in waiting state: "
+                + theControl.getWaitingJobs().size());
+        System.out.println("Jobs in ready state: "
+                + theControl.getReadyJobs().size());
+        System.out.println("Jobs in running state: "
+                + theControl.getRunningJobs().size());
+        System.out.println("Jobs in success state: "
+                + theControl.getSuccessfulJobs().size());
+        System.out.println("Jobs in failed state: "
+                + theControl.getFailedJobs().size());
+        System.out.println("\n");
+        
+        if (job_1.getState() != Job.FAILED && 
+                job_1.getState() != Job.DEPENDENT_FAILED && 
+                job_1.getState() != Job.SUCCESS) {
+           
+                String states = "job_1:  " + job_1.getState() + "\n";
+                throw new Exception("The state of job_1 is not in a complete state\n" + states);
+        }
+        
+        if (job_2.getState() != Job.FAILED &&
+                job_2.getState() != Job.DEPENDENT_FAILED && 
+                job_2.getState() != Job.SUCCESS) {
+           
+                String states = "job_2:  " + job_2.getState() + "\n";
+                throw new Exception("The state of job_2 is not in a complete state\n" + states);
+        }
+        
+        if (job_3.getState() != Job.FAILED && 
+                job_3.getState() != Job.DEPENDENT_FAILED && 
+                job_3.getState() != Job.SUCCESS) {
+           
+                String states = "job_3:  " + job_3.getState() + "\n";
+                throw new Exception("The state of job_3 is not in a complete state\n" + states);
+        }
+        if (job_4.getState() != Job.FAILED && 
+                job_4.getState() != Job.DEPENDENT_FAILED && 
+                job_4.getState() != Job.SUCCESS) {
+           
+                String states = "job_4:  " + job_4.getState() + "\n";
+                throw new Exception("The state of job_4 is not in a complete state\n" + states);
+        }
+        
+        if (job_1.getState() == Job.FAILED || 
+                job_2.getState() == Job.FAILED ||
+                job_1.getState() == Job.DEPENDENT_FAILED || 
+                job_2.getState() == Job.DEPENDENT_FAILED) {
+            if (job_3.getState() != Job.DEPENDENT_FAILED) {
+                String states = "job_1:  " + job_1.getState() + "\n";
+                states = "job_2:  " + job_2.getState() + "\n";
+                states = "job_3:  " + job_3.getState() + "\n";
+                states = "job_4:  " + job_4.getState() + "\n";
+                throw new Exception("The states of jobs 1, 2, 3, 4 are not consistent\n" + states);
+            }
+        }
+        if (job_3.getState() == Job.FAILED || 
+                job_3.getState() == Job.DEPENDENT_FAILED) {
+            if (job_4.getState() != Job.DEPENDENT_FAILED) {
+                String states = "job_3:  " + job_3.getState() + "\n";
+                states = "job_4:  " + job_4.getState() + "\n";
+                throw new Exception("The states of jobs 3, 4 are not consistent\n" + states);
+            }
+        }
+        
+        theControl.stop();
+    }
+
+    public void testJobControl() throws Exception {
+        doJobControlTest();
+    }
+    
+    public static void main(String[] args) {
+        TestJobControl test = new TestJobControl();
+        try {
+            test.testJobControl();
+        }
+        catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+}