@Override
	public void run() {

			

			try
			{
				
				pushStatusToTaskTracker();

				//called once
				reducer.setup();
				
				
				// Shuffle
				int count = 0;
				if(lookUpJobTracker()){
					MapJobsStatus status;
					status = getJobTrackerServiceProvider().reportMapStatus(taskID);
					while(!status.equals(MapJobsStatus.SUCCEEDED)){
						status = getJobTrackerServiceProvider().reportMapStatus(taskID);
						Thread.sleep(1000);
					}	
				} else {
					System.out.println("ERROR : Cannot look up the Job tracker.");
					this.outputCollector.close();
					System.exit(0);
				}
				
				
				System.out.println("All map task have been completed.");
				System.out.println("Starting the Shuffle process");
				List<String> data = shuffleData();
				if(data == null){
					System.out.println("Error has been occured while accessing the Data from the Mappers.");
					// let the job tracker know??????
					this.outputCollector.close();
					System.exit(0);
				} else {
					System.out.println("Got all data from the Mapper.");
				}
				
				
				
				// Packaging the Data
				System.out.println("Combining the data.");
				HashMap<String, List<String>> packagedData = packageData(data);
				if(packagedData.isEmpty()){
					System.out.println("There are errors in packaging the Data.");
					this.outputCollector.close();
					System.exit(0);
				} else {
					System.out.println("Data packaged.");
					totalEntryNumber = packagedData.size();
				}

				
				
				// Reduce
				try {
					Iterator iter = packagedData.entrySet().iterator();
					while (iter.hasNext()) {

						Entry<String, List<String>> entry = (Entry<String, List<String>>) iter.next();
						reducer.reduce(entry.getKey(), entry.getValue().iterator(), this.outputCollector);
						
						// This is for tracking the progress
						processedEntryNumber++;
					}
					/* close the files */
					this.outputCollector.close();
					
					
					String name = this.outputCollector.getOutputFileName().substring(SystemConstants.getConfig(SystemConstants.ADFS_DIRECTORY).length() + 1);
					this.nameNodeSlaveReference.registerToLocalDataNode(name);
					getCreatedFiles().add(name);
				}/* if runtime exception happens in user's code, exit jvm */
				catch (RuntimeException e) {
					e.printStackTrace();
					System.exit(0);
				}

				// Send the file to the Job Client location
				sendResultToJobClient();
				
				//clean-up at end
				reducer.cleanUp();
				
				//Shows task is Done
				this.updateStatusSucceeded();
				
				Thread.sleep(4000);
			}
			catch(IOException | InterruptedException e)
			{
				System.exit(0);
			}

			
			
			System.exit(0);
		
	}