test_class.file: flink-core/src/test/java/org/apache/flink/api/common/operators/GenericDataSourceBaseTest.java
test_class.identifier: GenericDataSourceBaseTest
test_class.interfaces: implements java.io.Serializable
test_case.body: @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());
  		}
  	}
test_case.class_method_signature: GenericDataSourceBaseTest.testDataSourceWithRuntimeContext()
test_case.full_signature: @Test public void testDataSourceWithRuntimeContext()
test_case.identifier: testDataSourceWithRuntimeContext
test_case.invocations: disableObjectReuse executeOnCollections reset enableObjectReuse executeOnCollections assertEquals asList assertEquals asList printStackTrace fail getMessage
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testDataSourceWithRuntimeContext()
focal_class.declarator: DEFAULT_NAME = "<Unnamed Generic Data Source>"
focal_class.modifier: private static final
focal_class.original_string: private static final String DEFAULT_NAME = "<Unnamed Generic Data Source>";
focal_class.type: String
focal_class.var_name: DEFAULT_NAME
focal_class.declarator: formatWrapper
focal_class.modifier: protected final
focal_class.original_string: protected final UserCodeWrapper<? extends T> formatWrapper;
focal_class.type: UserCodeWrapper<? extends T>
focal_class.var_name: formatWrapper
focal_class.declarator: statisticsKey
focal_class.modifier: protected
focal_class.original_string: protected String statisticsKey;
focal_class.type: String
focal_class.var_name: statisticsKey
focal_class.declarator: splitProperties
focal_class.modifier: private
focal_class.original_string: private SplitDataProperties splitProperties;
focal_class.type: SplitDataProperties
focal_class.var_name: splitProperties
focal_class.file: flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSourceBase.java
focal_class.identifier: GenericDataSourceBase
focal_class.class_method_signature: GenericDataSourceBase.GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo, String name)
focal_class.full_signature: public  GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo, String name)
focal_class.identifier: GenericDataSourceBase
focal_class.modifiers: public
focal_class.parameters: (T format, OperatorInformation<OUT> operatorInfo, String name)
focal_class.signature:  GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo, String name)
focal_class.class_method_signature: GenericDataSourceBase.GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo)
focal_class.full_signature: public  GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo)
focal_class.identifier: GenericDataSourceBase
focal_class.modifiers: public
focal_class.parameters: (T format, OperatorInformation<OUT> operatorInfo)
focal_class.signature:  GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo)
focal_class.class_method_signature: GenericDataSourceBase.GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo, String name)
focal_class.full_signature: public  GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo, String name)
focal_class.identifier: GenericDataSourceBase
focal_class.modifiers: public
focal_class.parameters: (Class<? extends T> format, OperatorInformation<OUT> operatorInfo, String name)
focal_class.signature:  GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo, String name)
focal_class.class_method_signature: GenericDataSourceBase.GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo)
focal_class.full_signature: public  GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo)
focal_class.identifier: GenericDataSourceBase
focal_class.modifiers: public
focal_class.parameters: (Class<? extends T> format, OperatorInformation<OUT> operatorInfo)
focal_class.signature:  GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo)
focal_class.class_method_signature: GenericDataSourceBase.getFormatWrapper()
focal_class.full_signature: public UserCodeWrapper<? extends T> getFormatWrapper()
focal_class.identifier: getFormatWrapper
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: UserCodeWrapper<? extends T>
focal_class.signature: UserCodeWrapper<? extends T> getFormatWrapper()
focal_class.class_method_signature: GenericDataSourceBase.getUserCodeWrapper()
focal_class.full_signature: @Override public UserCodeWrapper<? extends T> getUserCodeWrapper()
focal_class.identifier: getUserCodeWrapper
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: UserCodeWrapper<? extends T>
focal_class.signature: UserCodeWrapper<? extends T> getUserCodeWrapper()
focal_class.class_method_signature: GenericDataSourceBase.getStatisticsKey()
focal_class.full_signature: public String getStatisticsKey()
focal_class.identifier: getStatisticsKey
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: String
focal_class.signature: String getStatisticsKey()
focal_class.class_method_signature: GenericDataSourceBase.setStatisticsKey(String statisticsKey)
focal_class.full_signature: public void setStatisticsKey(String statisticsKey)
focal_class.identifier: setStatisticsKey
focal_class.modifiers: public
focal_class.parameters: (String statisticsKey)
focal_class.return: void
focal_class.signature: void setStatisticsKey(String statisticsKey)
focal_class.class_method_signature: GenericDataSourceBase.setSplitDataProperties(SplitDataProperties<OUT> splitDataProperties)
focal_class.full_signature: public void setSplitDataProperties(SplitDataProperties<OUT> splitDataProperties)
focal_class.identifier: setSplitDataProperties
focal_class.modifiers: public
focal_class.parameters: (SplitDataProperties<OUT> splitDataProperties)
focal_class.return: void
focal_class.signature: void setSplitDataProperties(SplitDataProperties<OUT> splitDataProperties)
focal_class.class_method_signature: GenericDataSourceBase.getSplitDataProperties()
focal_class.full_signature: public SplitDataProperties<OUT> getSplitDataProperties()
focal_class.identifier: getSplitDataProperties
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: SplitDataProperties<OUT>
focal_class.signature: SplitDataProperties<OUT> getSplitDataProperties()
focal_class.class_method_signature: GenericDataSourceBase.accept(Visitor<Operator<?>> visitor)
focal_class.full_signature: @Override public void accept(Visitor<Operator<?>> visitor)
focal_class.identifier: accept
focal_class.modifiers: @Override public
focal_class.parameters: (Visitor<Operator<?>> visitor)
focal_class.return: void
focal_class.signature: void accept(Visitor<Operator<?>> visitor)
focal_class.class_method_signature: GenericDataSourceBase.executeOnCollections(RuntimeContext ctx, ExecutionConfig executionConfig)
focal_class.full_signature: protected List<OUT> executeOnCollections(RuntimeContext ctx, ExecutionConfig executionConfig)
focal_class.identifier: executeOnCollections
focal_class.modifiers: protected
focal_class.parameters: (RuntimeContext ctx, ExecutionConfig executionConfig)
focal_class.return: List<OUT>
focal_class.signature: List<OUT> executeOnCollections(RuntimeContext ctx, ExecutionConfig executionConfig)
focal_class.class_method_signature: GenericDataSourceBase.toString()
focal_class.full_signature: public String toString()
focal_class.identifier: toString
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: String
focal_class.signature: String toString()
focal_class.superclass: extends Operator<OUT>
focal_method.body: protected List<OUT> executeOnCollections(RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
  		@SuppressWarnings("unchecked")
  		InputFormat<OUT, InputSplit> inputFormat = (InputFormat<OUT, InputSplit>) this.formatWrapper.getUserCodeObject();
  		inputFormat.configure(this.parameters);
  
  		if(inputFormat instanceof RichInputFormat){
  			((RichInputFormat) inputFormat).setRuntimeContext(ctx);
  		}
  
  		List<OUT> result = new ArrayList<OUT>();
  		
  		// splits
  		InputSplit[] splits = inputFormat.createInputSplits(1);
  		TypeSerializer<OUT> serializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
  		
  		for (InputSplit split : splits) {
  			inputFormat.open(split);
  			
  			while (!inputFormat.reachedEnd()) {
  				OUT next = inputFormat.nextRecord(serializer.createInstance());
  				if (next != null) {
  					result.add(serializer.copy(next));
  				}
  			}
  			
  			inputFormat.close();
  		}
  		
  		return result;
  	}
focal_method.class_method_signature: GenericDataSourceBase.executeOnCollections(RuntimeContext ctx, ExecutionConfig executionConfig)
focal_method.full_signature: protected List<OUT> executeOnCollections(RuntimeContext ctx, ExecutionConfig executionConfig)
focal_method.identifier: executeOnCollections
focal_method.invocations: getUserCodeObject configure setRuntimeContext createInputSplits createSerializer getOutputType getOperatorInfo open reachedEnd nextRecord createInstance add copy close
focal_method.modifiers: protected
focal_method.parameters: (RuntimeContext ctx, ExecutionConfig executionConfig)
focal_method.return: List<OUT>
focal_method.signature: List<OUT> executeOnCollections(RuntimeContext ctx, ExecutionConfig executionConfig)