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

	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>();
	Configuration stubParams = new Configuration();
	super.getTaskConfig().setStubParameters(stubParams);
	
	// 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, MockFailingOutputFormat.class, new File(tempTestPath).toURI().toString());
	
	boolean stubFailed = false;
	
	try {
		testTask.invoke();
	} catch (Exception e) {
		stubFailed = true;
	}
	
	Assert.assertTrue("Function exception was not forwarded.", stubFailed);
	
	// assert that temp file was created
	File tempTestFile = new File(this.tempTestPath);
	Assert.assertTrue("Temp output file does not exist",tempTestFile.exists());
	
}