test_class.declarator: LOG = LoggerFactory.getLogger(DataSinkTaskTest.class)
test_class.modifier: private static final
test_class.original_string: private static final Logger LOG = LoggerFactory.getLogger(DataSinkTaskTest.class);
test_class.type: Logger
test_class.var_name: LOG
test_class.declarator: MEMORY_MANAGER_SIZE = 3 * 1024 * 1024
test_class.modifier: private static final
test_class.original_string: private static final int MEMORY_MANAGER_SIZE = 3 * 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: tempTestPath = constructTestPath(DataSinkTaskTest.class, "dst_test")
test_class.modifier: private final
test_class.original_string: private final String tempTestPath = constructTestPath(DataSinkTaskTest.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/DataSinkTaskTest.java
test_class.identifier: DataSinkTaskTest
test_class.superclass: extends TaskTestBase
test_case.body: @Test
  	public void testUnionDataSinkTask() {
  
  		int keyCnt = 10;
  		int valCnt = 20;
  
  		super.initEnvironment(MEMORY_MANAGER_SIZE, NETWORK_BUFFER_SIZE);
  
  		final IteratorWrappingTestSingleInputGate<?>[] readers = new IteratorWrappingTestSingleInputGate[4];
  		readers[0] = super.addInput(new UniformRecordGenerator(keyCnt, valCnt, 0, 0, false), 0, false);
  		readers[1] = super.addInput(new UniformRecordGenerator(keyCnt, valCnt, keyCnt, 0, false), 0, false);
  		readers[2] = super.addInput(new UniformRecordGenerator(keyCnt, valCnt, keyCnt * 2, 0, false), 0, false);
  		readers[3] = super.addInput(new UniformRecordGenerator(keyCnt, valCnt, keyCnt * 3, 0, false), 0, false);
  
  		DataSinkTask<Record> testTask = new DataSinkTask<>();
  
  		super.registerFileOutputTask(testTask, MockOutputFormat.class, new File(tempTestPath).toURI().toString());
  
  		try {
  			// For the union reader to work, we need to start notifications *after* the union reader
  			// has been initialized.
  			for (IteratorWrappingTestSingleInputGate<?> reader : readers) {
  				reader.read();
  			}
  
  			testTask.invoke();
  		} catch (Exception e) {
  			LOG.debug("Exception while invoking the test task.", e);
  			Assert.fail("Invoke method caused exception.");
  		}
  
  		File tempTestFile = new File(this.tempTestPath);
  
  		Assert.assertTrue("Temp output file does not exist",tempTestFile.exists());
  
  		FileReader fr = null;
  		BufferedReader br = null;
  		try {
  			fr = new FileReader(tempTestFile);
  			br = new BufferedReader(fr);
  
  			HashMap<Integer,HashSet<Integer>> keyValueCountMap = new HashMap<>(keyCnt);
  
  			while(br.ready()) {
  				String line = br.readLine();
  
  				Integer key = Integer.parseInt(line.substring(0,line.indexOf("_")));
  				Integer val = Integer.parseInt(line.substring(line.indexOf("_")+1,line.length()));
  
  				if(!keyValueCountMap.containsKey(key)) {
  					keyValueCountMap.put(key,new HashSet<Integer>());
  				}
  				keyValueCountMap.get(key).add(val);
  			}
  
  			Assert.assertTrue("Invalid key count in out file. Expected: "+keyCnt+" Actual: "+keyValueCountMap.keySet().size(),
  				keyValueCountMap.keySet().size() == keyCnt * 4);
  
  			for(Integer key : keyValueCountMap.keySet()) {
  				Assert.assertTrue("Invalid value count for key: "+key+". Expected: "+valCnt+" Actual: "+keyValueCountMap.get(key).size(),
  					keyValueCountMap.get(key).size() == valCnt);
  			}
  
  		} catch (FileNotFoundException e) {
  			Assert.fail("Out file got lost...");
  		} catch (IOException ioe) {
  			Assert.fail("Caught IOE while reading out file");
  		} finally {
  			if (br != null) {
  				try { br.close(); } catch (Throwable t) {}
  			}
  			if (fr != null) {
  				try { fr.close(); } catch (Throwable t) {}
  			}
  		}
  	}
test_case.class_method_signature: DataSinkTaskTest.testUnionDataSinkTask()
test_case.full_signature: @Test public void testUnionDataSinkTask()
test_case.identifier: testUnionDataSinkTask
test_case.invocations: initEnvironment addInput addInput addInput addInput registerFileOutputTask toString toURI read invoke debug fail assertTrue exists ready readLine parseInt substring indexOf parseInt substring indexOf length containsKey put add get assertTrue size keySet size keySet keySet assertTrue size get size get fail fail close close
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testUnionDataSinkTask()
focal_class.declarator: LOG = LoggerFactory.getLogger(DataSinkTask.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOG = LoggerFactory.getLogger(DataSinkTask.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: format
focal_class.modifier: private volatile
focal_class.original_string: private volatile OutputFormat<IT> format;
focal_class.type: OutputFormat<IT>
focal_class.var_name: format
focal_class.declarator: inputReader
focal_class.modifier: private
focal_class.original_string: private MutableReader<?> inputReader;
focal_class.type: MutableReader<?>
focal_class.var_name: inputReader
focal_class.declarator: reader
focal_class.modifier: private
focal_class.original_string: private MutableObjectIterator<IT> reader;
focal_class.type: MutableObjectIterator<IT>
focal_class.var_name: reader
focal_class.declarator: inputTypeSerializerFactory
focal_class.modifier: private
focal_class.original_string: private TypeSerializerFactory<IT> inputTypeSerializerFactory;
focal_class.type: TypeSerializerFactory<IT>
focal_class.var_name: inputTypeSerializerFactory
focal_class.declarator: localStrategy
focal_class.modifier: private
focal_class.original_string: private CloseableInputProvider<IT> localStrategy;
focal_class.type: CloseableInputProvider<IT>
focal_class.var_name: localStrategy
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: taskCanceled
focal_class.modifier: private volatile
focal_class.original_string: private volatile boolean taskCanceled;
focal_class.type: boolean
focal_class.var_name: taskCanceled
focal_class.declarator: cleanupCalled
focal_class.modifier: private volatile
focal_class.original_string: private volatile boolean cleanupCalled;
focal_class.type: boolean
focal_class.var_name: cleanupCalled
focal_class.file: flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
focal_class.identifier: DataSinkTask
focal_class.class_method_signature: DataSinkTask.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: DataSinkTask.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: DataSinkTask.initOutputFormat()
focal_class.full_signature: private void initOutputFormat()
focal_class.identifier: initOutputFormat
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void initOutputFormat()
focal_class.class_method_signature: DataSinkTask.initInputReaders()
focal_class.full_signature: @SuppressWarnings("unchecked") private void initInputReaders()
focal_class.identifier: initInputReaders
focal_class.modifiers: @SuppressWarnings("unchecked") private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void initInputReaders()
focal_class.class_method_signature: DataSinkTask.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: DataSinkTask.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
  		// --------------------------------------------------------------------
  		LOG.debug(getLogString("Start registering input and output"));
  
  		// initialize OutputFormat
  		initOutputFormat();
  
  		// initialize input readers
  		try {
  			initInputReaders();
  		} catch (Exception e) {
  			throw new RuntimeException("Initializing the input streams failed" +
  					(e.getMessage() == null ? "." : ": " + e.getMessage()), e);
  		}
  
  		LOG.debug(getLogString("Finished registering input and output"));
  
  		// --------------------------------------------------------------------
  		// Invoke
  		// --------------------------------------------------------------------
  		LOG.debug(getLogString("Starting data sink operator"));
  
  		if(RichOutputFormat.class.isAssignableFrom(this.format.getClass())){
  			((RichOutputFormat) this.format).setRuntimeContext(createRuntimeContext());
  			LOG.debug(getLogString("Rich Sink detected. Initializing runtime context."));
  		}
  
  		ExecutionConfig executionConfig;
  		try {
  			ExecutionConfig c = InstantiationUtil.readObjectFromConfig(
  					getJobConfiguration(),
  					ExecutionConfig.CONFIG_KEY,
  					getUserCodeClassLoader());
  			if (c != null) {
  				executionConfig = c;
  			} else {
  				LOG.warn("The execution config returned by the configuration was null");
  				executionConfig = new ExecutionConfig();
  			}
  		} catch (IOException | ClassNotFoundException e) {
  			throw new RuntimeException("Could not load ExecutionConfig from Job Configuration: " + e);
  		}
  		boolean objectReuseEnabled = executionConfig.isObjectReuseEnabled();
  		
  		try {
  			// initialize local strategies
  			MutableObjectIterator<IT> input1;
  			switch (this.config.getInputLocalStrategy(0)) {
  			case NONE:
  				// nothing to do
  				localStrategy = null;
  				input1 = reader;
  				break;
  			case SORT:
  				// initialize sort local strategy
  				try {
  					// get type comparator
  					TypeComparatorFactory<IT> compFact = this.config.getInputComparator(0,
  							getUserCodeClassLoader());
  					if (compFact == null) {
  						throw new Exception("Missing comparator factory for local strategy on input " + 0);
  					}
  					
  					// initialize sorter
  					UnilateralSortMerger<IT> sorter = new UnilateralSortMerger<IT>(
  							getEnvironment().getMemoryManager(), 
  							getEnvironment().getIOManager(),
  							this.reader, this, this.inputTypeSerializerFactory, compFact.createComparator(),
  							this.config.getRelativeMemoryInput(0), this.config.getFilehandlesInput(0),
  							this.config.getSpillingThresholdInput(0),
  							this.getExecutionConfig().isObjectReuseEnabled());
  					
  					this.localStrategy = sorter;
  					input1 = sorter.getIterator();
  				} catch (Exception e) {
  					throw new RuntimeException("Initializing the input processing failed" +
  							(e.getMessage() == null ? "." : ": " + e.getMessage()), e);
  				}
  				break;
  			default:
  				throw new RuntimeException("Invalid local strategy for DataSinkTask");
  			}
  			
  			// read the reader and write it to the output
  			
  			final TypeSerializer<IT> serializer = this.inputTypeSerializerFactory.getSerializer();
  			final MutableObjectIterator<IT> input = input1;
  			final OutputFormat<IT> format = this.format;
  
  
  			// check if task has been canceled
  			if (this.taskCanceled) {
  				return;
  			}
  
  			LOG.debug(getLogString("Starting to produce output"));
  
  			// open
  			format.open(this.getEnvironment().getTaskInfo().getIndexOfThisSubtask(), this.getEnvironment().getTaskInfo().getNumberOfParallelSubtasks());
  
  			if (objectReuseEnabled) {
  				IT record = serializer.createInstance();
  
  				// work!
  				while (!this.taskCanceled && ((record = input.next(record)) != null)) {
  					format.writeRecord(record);
  				}
  			} else {
  				IT record;
  
  				// work!
  				while (!this.taskCanceled && ((record = input.next()) != null)) {
  					format.writeRecord(record);
  				}
  			}
  			
  			// close. We close here such that a regular close throwing an exception marks a task as failed.
  			if (!this.taskCanceled) {
  				this.format.close();
  				this.format = null;
  			}
  		}
  		catch (Exception ex) {
  			
  			// make a best effort to clean up
  			try {
  				if (!cleanupCalled && format instanceof CleanupWhenUnsuccessful) {
  					cleanupCalled = true;
  					((CleanupWhenUnsuccessful) format).tryCleanupOnError();
  				}
  			}
  			catch (Throwable t) {
  				LOG.error("Cleanup on error failed.", t);
  			}
  			
  			ex = ExceptionInChainedStubException.exceptionUnwrap(ex);
  
  			if (ex instanceof CancelTaskException) {
  				// forward canceling exception
  				throw ex;
  			}
  			// drop, if the task was canceled
  			else if (!this.taskCanceled) {
  				if (LOG.isErrorEnabled()) {
  					LOG.error(getLogString("Error in user code: " + ex.getMessage()), ex);
  				}
  				throw ex;
  			}
  		}
  		finally {
  			if (this.format != null) {
  				// close format, if it has not been closed, yet.
  				// This should only be the case if we had a previous error, or were canceled.
  				try {
  					this.format.close();
  				}
  				catch (Throwable t) {
  					if (LOG.isWarnEnabled()) {
  						LOG.warn(getLogString("Error closing the output format"), t);
  					}
  				}
  			}
  			// close local strategy if necessary
  			if (localStrategy != null) {
  				try {
  					this.localStrategy.close();
  				} catch (Throwable t) {
  					LOG.error("Error closing local strategy", t);
  				}
  			}
  
  			BatchTask.clearReaders(new MutableReader<?>[]{inputReader});
  		}
  
  		if (!this.taskCanceled) {
  			LOG.debug(getLogString("Finished data sink operator"));
  		}
  		else {
  			LOG.debug(getLogString("Data sink operator cancelled"));
  		}
  	}
focal_method.class_method_signature: DataSinkTask.invoke()
focal_method.full_signature: @Override public void invoke()
focal_method.identifier: invoke
focal_method.invocations: debug getLogString initOutputFormat initInputReaders getMessage getMessage debug getLogString debug getLogString isAssignableFrom getClass setRuntimeContext createRuntimeContext debug getLogString readObjectFromConfig getJobConfiguration getUserCodeClassLoader warn isObjectReuseEnabled getInputLocalStrategy getInputComparator getUserCodeClassLoader getMemoryManager getEnvironment getIOManager getEnvironment createComparator getRelativeMemoryInput getFilehandlesInput getSpillingThresholdInput isObjectReuseEnabled getExecutionConfig getIterator getMessage getMessage getSerializer debug getLogString open getIndexOfThisSubtask getTaskInfo getEnvironment getNumberOfParallelSubtasks getTaskInfo getEnvironment createInstance next writeRecord next writeRecord close tryCleanupOnError error exceptionUnwrap isErrorEnabled error getLogString getMessage close isWarnEnabled warn getLogString close error clearReaders debug getLogString debug getLogString
focal_method.modifiers: @Override public
focal_method.parameters: ()
focal_method.return: void
focal_method.signature: void invoke()