public static void submitJobAndWaitForCompletion(Job job) {
		ClusterConfig clusterConfig = new ClusterConfig(job.getConfigFile());
		//Ask Job Tracker for a new job Id.
		//Check output specification for the job. Error if output dir already exists.
		
		//Compute input splits for the job.
		
		//Copy the resources needed to run the job. (Jar file, config file, computed splits)
		//Might not be needed if we use AFS.
		
		//Make an RPC call to JobTracker to start the job.

        String host = clusterConfig.getJobTrackerHost();
        int port = clusterConfig.getJobTrackerClientCommPort();

        Socket sock = null;

        try {
        	List<Split> splits = computeSplits(job);
        	job.setSplits(splits);
        	
        	JobProgress progress;
        	//Indicates that requesting new Id
        	job.setId(-1);
        	
            sock = new Socket(host,port);

            ObjectOutputStream oos = new ObjectOutputStream(sock.getOutputStream());
            oos.writeObject(job);
            oos.flush();
            
            ObjectInputStream ois = new ObjectInputStream(sock.getInputStream());
            progress = (JobProgress) ois.readObject();
            
            job.setId(progress.getJobId());
            
            oos.close();
            ois.close();
            sock.close();
            
            //Poll for job status periodically.
            boolean inProgress = true;
            do {
            	Thread.sleep(1000);
            	sock = new Socket(host,port);
            	oos = new ObjectOutputStream(sock.getOutputStream());
                oos.writeObject(job);
                oos.flush();
                
                ois = new ObjectInputStream(sock.getInputStream());
                progress = (JobProgress) ois.readObject();
                
                oos.close();
                ois.close();
                sock.close();
            	inProgress = (progress.getState() != JobStatus.JobState.SUCCESS) &&
            			(progress.getState() != JobStatus.JobState.FAILED);
            	System.out.println(progress.getMessage());
            } while (inProgress);

            //Clean up the tmp files in the shared file system.
            File dir = new File(job.getTmpMapOpDir());
            if(dir.exists()){
                File[] files = dir.listFiles();
                for(File f : files) {
                	f.delete();
                }
            }
            dir.delete();
            System.out.println("Job with job Id: " + progress.getJobId() + " finished " +
            		"with state " + progress.getState());
            
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (sock != null) {
                try {
                    sock.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }