@Test
public void testDataSourceWithRuntimeContext() {

	try {
		TestRichInputFormat in = new TestRichInputFormat();
		GenericDataSourceBase<String, TestRichInputFormat> source =
				new GenericDataSourceBase<String, TestRichInputFormat>(
						in, new OperatorInformation<String>(BasicTypeInfo.STRING_TYPE_INFO), "testSource");
		final HashMap<String, Accumulator<?, ?>> accumulatorMap = new HashMap<String, Accumulator<?, ?>>();
		final HashMap<String, Future<Path>> cpTasks = new HashMap<>();
		final TaskInfo taskInfo = new TaskInfo("test_source", 0, 1, 0);
		ExecutionConfig executionConfig = new ExecutionConfig();
		executionConfig.disableObjectReuse();
		List<String> resultMutableSafe = source.executeOnCollections(new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap), executionConfig);
		in.reset();
		executionConfig.enableObjectReuse();
		List<String> resultRegular = source.executeOnCollections(new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap), executionConfig);
		assertEquals(asList(TestIOData.RICH_NAMES), resultMutableSafe);
		assertEquals(asList(TestIOData.RICH_NAMES), resultRegular);
	} catch(Exception e){
		e.printStackTrace();
		fail(e.getMessage());
	}
}