@Test
public void testFailingDataSourceTask() {

	int keyCnt = 20;
	int valCnt = 10;
	
	this.outList = new NirvanaOutputList();
	
	try {
		InputFilePreparator.prepareInputFile(new UniformRecordGenerator(keyCnt, valCnt, false), 
			this.tempTestPath, false);
	} catch (IOException e1) {
		Assert.fail("Unable to set-up test input file");
	}
	super.initEnvironment(MEMORY_MANAGER_SIZE, NETWORK_BUFFER_SIZE);
	super.addOutput(this.outList);
	
	DataSourceTask<Record> testTask = new DataSourceTask<>();
	super.registerFileInputTask(testTask, MockFailingInputFormat.class, new File(tempTestPath).toURI().toString(), "\n");
	
	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());
	
}