public int fetchAndMergeRecords() {
		ArrayList<MapperInfo> mappers = new ArrayList<MapperInfo>();
		int i = 0;
		for (TaskAttemptID id : task.getMapTaskToTaskTrackerIDMap().keySet()) {
			TaskTrackerID tid = task.getMapTaskToTaskTrackerIDMap().get(id);
			mappers.add(new MapperInfo(tid.getHost(), 3579, i++, id));
		}
		Log.write("Preparing to fetch intermediate records from " + mappers.size() + " mappers...");
		int size = mappers.size();
		
		for (MapperInfo mapper : mappers)
			new Thread(new RecordFileFetcher(mapper)).start();
		
		// Wait until all records are fetched.
		while (!failed) {
			synchronized(nFinished) {
				if (nFinished >= size)
					break;
			}
		}
		
		if (nFinished == size) {
			Log.write("All partitions fetched.");
			try {
				return mergeRecords(size);
			} catch (Exception e) {
				e.printStackTrace();
				return -1;
			}
		} else
			return -1;
	}