test_class.declarator: MEMORY_MANAGER_SIZE = 1024 * 1024
test_class.modifier: private static final
test_class.original_string: private static final int MEMORY_MANAGER_SIZE = 1024 * 1024;
test_class.type: int
test_class.var_name: MEMORY_MANAGER_SIZE
test_class.declarator: NETWORK_BUFFER_SIZE = 1024
test_class.modifier: private static final
test_class.original_string: private static final int NETWORK_BUFFER_SIZE = 1024;
test_class.type: int
test_class.var_name: NETWORK_BUFFER_SIZE
test_class.declarator: outList
test_class.modifier: private
test_class.original_string: private List<Record> outList;
test_class.type: List<Record>
test_class.var_name: outList
test_class.declarator: tempTestPath = DataSinkTaskTest.constructTestPath(DataSourceTaskTest.class, "dst_test")
test_class.modifier: private
test_class.original_string: private String tempTestPath = DataSinkTaskTest.constructTestPath(DataSourceTaskTest.class, "dst_test");
test_class.type: String
test_class.var_name: tempTestPath
test_class.file: flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSourceTaskTest.java
test_class.identifier: DataSourceTaskTest
test_class.superclass: extends TaskTestBase
test_case.body: @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());
  	}
test_case.class_method_signature: DataSourceTaskTest.testCancelDataSourceTask()
test_case.full_signature: @Test public void testCancelDataSourceTask()
test_case.identifier: testCancelDataSourceTask
test_case.invocations: initEnvironment addOutput prepareInputFile fail registerFileInputTask toString toURI invoke printStackTrace fail start start join join fail assertTrue exists
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testCancelDataSourceTask()
focal_class.declarator: LOG = LoggerFactory.getLogger(DataSourceTask.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOG = LoggerFactory.getLogger(DataSourceTask.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: eventualOutputs
focal_class.modifier: private
focal_class.original_string: private List<RecordWriter<?>> eventualOutputs;
focal_class.type: List<RecordWriter<?>>
focal_class.var_name: eventualOutputs
focal_class.declarator: output
focal_class.modifier: private
focal_class.original_string: private Collector<OT> output;
focal_class.type: Collector<OT>
focal_class.var_name: output
focal_class.declarator: format
focal_class.modifier: private
focal_class.original_string: private InputFormat<OT, InputSplit> format;
focal_class.type: InputFormat<OT, InputSplit>
focal_class.var_name: format
focal_class.declarator: serializerFactory
focal_class.modifier: private
focal_class.original_string: private TypeSerializerFactory<OT> serializerFactory;
focal_class.type: TypeSerializerFactory<OT>
focal_class.var_name: serializerFactory
focal_class.declarator: config
focal_class.modifier: private
focal_class.original_string: private TaskConfig config;
focal_class.type: TaskConfig
focal_class.var_name: config
focal_class.declarator: chainedTasks
focal_class.modifier: private
focal_class.original_string: private ArrayList<ChainedDriver<?, ?>> chainedTasks;
focal_class.type: ArrayList<ChainedDriver<?, ?>>
focal_class.var_name: chainedTasks
focal_class.declarator: taskCanceled = false
focal_class.modifier: private volatile
focal_class.original_string: private volatile boolean taskCanceled = false;
focal_class.type: boolean
focal_class.var_name: taskCanceled
focal_class.file: flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
focal_class.identifier: DataSourceTask
focal_class.class_method_signature: DataSourceTask.invoke()
focal_class.full_signature: @Override public void invoke()
focal_class.identifier: invoke
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void invoke()
focal_class.class_method_signature: DataSourceTask.cancel()
focal_class.full_signature: @Override public void cancel()
focal_class.identifier: cancel
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void cancel()
focal_class.class_method_signature: DataSourceTask.initInputFormat()
focal_class.full_signature: private void initInputFormat()
focal_class.identifier: initInputFormat
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void initInputFormat()
focal_class.class_method_signature: DataSourceTask.initOutputs(ClassLoader cl)
focal_class.full_signature: private void initOutputs(ClassLoader cl)
focal_class.identifier: initOutputs
focal_class.modifiers: private
focal_class.parameters: (ClassLoader cl)
focal_class.return: void
focal_class.signature: void initOutputs(ClassLoader cl)
focal_class.class_method_signature: DataSourceTask.getLogString(String message)
focal_class.full_signature: private String getLogString(String message)
focal_class.identifier: getLogString
focal_class.modifiers: private
focal_class.parameters: (String message)
focal_class.return: String
focal_class.signature: String getLogString(String message)
focal_class.class_method_signature: DataSourceTask.getLogString(String message, String taskName)
focal_class.full_signature: private String getLogString(String message, String taskName)
focal_class.identifier: getLogString
focal_class.modifiers: private
focal_class.parameters: (String message, String taskName)
focal_class.return: String
focal_class.signature: String getLogString(String message, String taskName)
focal_class.class_method_signature: DataSourceTask.getInputSplits()
focal_class.full_signature: private Iterator<InputSplit> getInputSplits()
focal_class.identifier: getInputSplits
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: Iterator<InputSplit>
focal_class.signature: Iterator<InputSplit> getInputSplits()
focal_class.class_method_signature: DataSourceTask.createRuntimeContext()
focal_class.full_signature: public DistributedRuntimeUDFContext createRuntimeContext()
focal_class.identifier: createRuntimeContext
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: DistributedRuntimeUDFContext
focal_class.signature: DistributedRuntimeUDFContext createRuntimeContext()
focal_class.superclass: extends AbstractInvokable
focal_method.body: @Override
  	public void invoke() throws Exception {
  		// --------------------------------------------------------------------
  		// Initialize
  		// --------------------------------------------------------------------
  		initInputFormat();
  
  		LOG.debug(getLogString("Start registering input and output"));
  
  		try {
  			initOutputs(getUserCodeClassLoader());
  		} catch (Exception ex) {
  			throw new RuntimeException("The initialization of the DataSource's outputs caused an error: " +
  					ex.getMessage(), ex);
  		}
  
  		LOG.debug(getLogString("Finished registering input and output"));
  
  		// --------------------------------------------------------------------
  		// Invoke
  		// --------------------------------------------------------------------
  		LOG.debug(getLogString("Starting data source operator"));
  
  		if(RichInputFormat.class.isAssignableFrom(this.format.getClass())){
  			((RichInputFormat) this.format).setRuntimeContext(createRuntimeContext());
  			LOG.debug(getLogString("Rich Source detected. Initializing runtime context."));
  		}
  
  		ExecutionConfig executionConfig;
  		try {
  			ExecutionConfig c = (ExecutionConfig) InstantiationUtil.readObjectFromConfig(
  					getJobConfiguration(),
  					ExecutionConfig.CONFIG_KEY,
  					getUserCodeClassLoader());
  			if (c != null) {
  				executionConfig = c;
  			} else {
  				LOG.warn("ExecutionConfig from job configuration is null. Creating empty config");
  				executionConfig = new ExecutionConfig();
  			}
  		} catch (IOException | ClassNotFoundException e) {
  			throw new RuntimeException("Could not load ExecutionConfig from Job Configuration: ", e);
  		}
  
  		boolean objectReuseEnabled = executionConfig.isObjectReuseEnabled();
  
  		LOG.debug("DataSourceTask object reuse: " + (objectReuseEnabled ? "ENABLED" : "DISABLED") + ".");
  		
  		final TypeSerializer<OT> serializer = this.serializerFactory.getSerializer();
  		
  		try {
  			// start all chained tasks
  			BatchTask.openChainedTasks(this.chainedTasks, this);
  			
  			// get input splits to read
  			final Iterator<InputSplit> splitIterator = getInputSplits();
  			
  			// for each assigned input split
  			while (!this.taskCanceled && splitIterator.hasNext())
  			{
  				// get start and end
  				final InputSplit split = splitIterator.next();
  
  				LOG.debug(getLogString("Opening input split " + split.toString()));
  				
  				final InputFormat<OT, InputSplit> format = this.format;
  			
  				// open input format
  				format.open(split);
  	
  				LOG.debug(getLogString("Starting to read input from split " + split.toString()));
  				
  				try {
  					final Collector<OT> output = this.output;
  
  					if (objectReuseEnabled) {
  						OT reuse1 = serializer.createInstance();
  						OT reuse2 = serializer.createInstance();
  						OT reuse3 = serializer.createInstance();
  
  						// as long as there is data to read
  						while (!this.taskCanceled && !format.reachedEnd()) {
  
  							OT returned;
  							if ((returned = format.nextRecord(reuse1)) != null) {
  								output.collect(returned);
  
  								reuse1 = reuse2;
  								reuse2 = reuse3;
  								reuse3 = returned;
  							}
  						}
  					} else {
  						// as long as there is data to read
  						while (!this.taskCanceled && !format.reachedEnd()) {
  
  							OT returned;
  							if ((returned = format.nextRecord(serializer.createInstance())) != null) {
  								output.collect(returned);
  							}
  						}
  					}
  
  					if (LOG.isDebugEnabled() && !this.taskCanceled) {
  						LOG.debug(getLogString("Closing input split " + split.toString()));
  					}
  				} finally {
  					// close. We close here such that a regular close throwing an exception marks a task as failed.
  					format.close();
  				}
  			} // end for all input splits
  
  			// close the collector. if it is a chaining task collector, it will close its chained tasks
  			this.output.close();
  
  			// close all chained tasks letting them report failure
  			BatchTask.closeChainedTasks(this.chainedTasks, this);
  
  		}
  		catch (Exception ex) {
  			// close the input, but do not report any exceptions, since we already have another root cause
  			try {
  				this.format.close();
  			} catch (Throwable ignored) {}
  
  			BatchTask.cancelChainedTasks(this.chainedTasks);
  
  			ex = ExceptionInChainedStubException.exceptionUnwrap(ex);
  
  			if (ex instanceof CancelTaskException) {
  				// forward canceling exception
  				throw ex;
  			}
  			else if (!this.taskCanceled) {
  				// drop exception, if the task was canceled
  				BatchTask.logAndThrowException(ex, this);
  			}
  		} finally {
  			BatchTask.clearWriters(eventualOutputs);
  		}
  
  		if (!this.taskCanceled) {
  			LOG.debug(getLogString("Finished data source operator"));
  		}
  		else {
  			LOG.debug(getLogString("Data source operator cancelled"));
  		}
  	}
focal_method.class_method_signature: DataSourceTask.invoke()
focal_method.full_signature: @Override public void invoke()
focal_method.identifier: invoke
focal_method.invocations: initInputFormat debug getLogString initOutputs getUserCodeClassLoader getMessage debug getLogString debug getLogString isAssignableFrom getClass setRuntimeContext createRuntimeContext debug getLogString readObjectFromConfig getJobConfiguration getUserCodeClassLoader warn isObjectReuseEnabled debug getSerializer openChainedTasks getInputSplits hasNext next debug getLogString toString open debug getLogString toString createInstance createInstance createInstance reachedEnd nextRecord collect reachedEnd nextRecord createInstance collect isDebugEnabled debug getLogString toString close close closeChainedTasks close cancelChainedTasks exceptionUnwrap logAndThrowException clearWriters debug getLogString debug getLogString
focal_method.modifiers: @Override public
focal_method.parameters: ()
focal_method.return: void
focal_method.signature: void invoke()