@Test
public void testCancelDataSourceTask() {

	int keyCnt = 20;
	int valCnt = 4;
	super.initEnvironment(MEMORY_MANAGER_SIZE, NETWORK_BUFFER_SIZE);
	super.addOutput(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");
	}
	
	final DataSourceTask<Record> testTask = new DataSourceTask<>();
	super.registerFileInputTask(testTask, MockDelayingInputFormat.class,  new File(tempTestPath).toURI().toString(), "\n");
	
	Thread taskRunner = new Thread() {
		@Override
		public void run() {
			try {
				testTask.invoke();
			} catch (Exception ie) {
				ie.printStackTrace();
				Assert.fail("Task threw exception although it was properly canceled");
			}
		}
	};
	taskRunner.start();
	
	TaskCancelThread tct = new TaskCancelThread(1, taskRunner, testTask);
	tct.start();
	
	try {
		tct.join();
		taskRunner.join();
	} catch(InterruptedException ie) {
		Assert.fail("Joining threads failed");
	}
	
	// assert that temp file was created
	File tempTestFile = new File(this.tempTestPath);
	Assert.assertTrue("Temp output file does not exist",tempTestFile.exists());
}