@Test
@SuppressWarnings("unchecked")
public void testSortingDataSinkTask() {

	int keyCnt = 100;
	int valCnt = 20;
	super.initEnvironment(MEMORY_MANAGER_SIZE * 4, NETWORK_BUFFER_SIZE);
	super.addInput(new UniformRecordGenerator(keyCnt, valCnt, true), 0);
	
	DataSinkTask<Record> testTask = new DataSinkTask<Record>();
	
	// set sorting
	super.getTaskConfig().setInputLocalStrategy(0, LocalStrategy.SORT);
	super.getTaskConfig().setInputComparator(
			new RecordComparatorFactory(new int[]{1},((Class<? extends Key<?>>[])new Class[]{IntValue.class})), 0);
	super.getTaskConfig().setMemoryInput(0, 4 * 1024 * 1024);
	super.getTaskConfig().setFilehandlesInput(0, 8);
	super.getTaskConfig().setSpillingThresholdInput(0, 0.8f);
	super.registerFileOutputTask(testTask, MockOutputFormat.class, new File(tempTestPath).toURI().toString());
	
	try {
		testTask.invoke();
	} catch (Exception e) {
		LOG.debug(e);
		Assert.fail("Invoke method caused exception.");
	}
	
	File tempTestFile = new File(this.tempTestPath);
	
	Assert.assertTrue("Temp output file does not exist",tempTestFile.exists());
	
	FileReader fr = null;
	BufferedReader br = null;
	try {
		fr = new FileReader(tempTestFile);
		br = new BufferedReader(fr);
		
		Set<Integer> keys = new HashSet<Integer>();
		
		int curVal = -1;
		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()));
			
			// check that values are in correct order
			Assert.assertTrue("Values not in ascending order", val >= curVal);
			// next value hit
			if(val > curVal) {
				if(curVal != -1) {
					// check that we saw 100 distinct keys for this values
					Assert.assertTrue("Keys missing for value", keys.size() == 100);
				}
				// empty keys set
				keys.clear();
				// update current value
				curVal = val;
			}
			
			Assert.assertTrue("Duplicate key for value", keys.add(key));
		}
		
	} catch (FileNotFoundException e) {
		Assert.fail("Out file got lost...");
	} catch (IOException ioe) {
		Assert.fail("Caught IOE while reading out file");
	} finally {
		if (br != null) {
			try { br.close(); } catch (Throwable t) {}
		}
		if (fr != null) {
			try { fr.close(); } catch (Throwable t) {}
		}
	}
}