@Test
public void testDataSinkTask() {

	FileReader fr = null;
	BufferedReader br = null;
	try {
		int keyCnt = 100;
		int valCnt = 20;
		super.initEnvironment(MEMORY_MANAGER_SIZE, NETWORK_BUFFER_SIZE);
		super.addInput(new UniformRecordGenerator(keyCnt, valCnt, false), 0);
		DataSinkTask<Record> testTask = new DataSinkTask<>();
		super.registerFileOutputTask(testTask, MockOutputFormat.class, new File(tempTestPath).toURI().toString());
		testTask.invoke();
		File tempTestFile = new File(this.tempTestPath);
		Assert.assertTrue("Temp output file does not exist", tempTestFile.exists());
		fr = new FileReader(tempTestFile);
		br = new BufferedReader(fr);
		HashMap<Integer, HashSet<Integer>> keyValueCountMap = new HashMap<>(keyCnt);
		while (br.ready()) {
			String line = br.readLine();
			Integer key = Integer.parseInt(line.substring(0, line.indexOf("_")));
			Integer val = Integer.parseInt(line.substring(line.indexOf("_") + 1, line.length()));
			if (!keyValueCountMap.containsKey(key)) {
				keyValueCountMap.put(key, new HashSet<Integer>());
			}
			keyValueCountMap.get(key).add(val);
		}
		Assert.assertTrue("Invalid key count in out file. Expected: " + keyCnt + " Actual: " + keyValueCountMap.keySet().size(),
				keyValueCountMap.keySet().size() == keyCnt);
		for (Integer key : keyValueCountMap.keySet()) {
			Assert.assertTrue("Invalid value count for key: " + key + ". Expected: " + valCnt + " Actual: " + keyValueCountMap.get(key).size(),
					keyValueCountMap.get(key).size() == valCnt);
		}
	}
	catch (Exception e) {
		e.printStackTrace();
		Assert.fail(e.getMessage());
	}
	finally {
		if (br != null) {
			try { br.close(); } catch (Throwable t) {}
		}
		if (fr != null) {
			try { fr.close(); } catch (Throwable t) {}
		}
	}
}