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

	super.initEnvironment(MEMORY_MANAGER_SIZE * 4, NETWORK_BUFFER_SIZE);
	super.addInput(new InfiniteInputIterator(), 0);
	
	final 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, MockOutputFormat.class,  new File(tempTestPath).toURI().toString());
	
	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(2, taskRunner, testTask);
	tct.start();
	
	try {
		tct.join();
		taskRunner.join();
	} catch(InterruptedException ie) {
		Assert.fail("Joining threads failed");
	}
			
}