@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<>(this.mockEnv); File tempTestFile = new File(tempFolder.getRoot(), UUID.randomUUID().toString()); super.registerFileOutputTask(MockOutputFormat.class, tempTestFile.toURI().toString(), new Configuration()); try { for (IteratorWrappingTestSingleInputGate<?> reader : readers) { reader.notifyNonEmpty(); } testTask.invoke(); } catch (Exception e) { LOG.debug("Exception while invoking the test task.", e); Assert.fail("Invoke method caused exception."); } 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) {} } } }
@Override public void invoke() throws Exception { LOG.debug(getLogString("Start registering input and output")); initOutputFormat(); 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")); LOG.debug(getLogString("Starting data sink operator")); RuntimeContext ctx = createRuntimeContext(); final Counter numRecordsIn; { Counter tmpNumRecordsIn; try { OperatorIOMetricGroup ioMetricGroup = ((OperatorMetricGroup) ctx.getMetricGroup()).getIOMetricGroup(); ioMetricGroup.reuseInputMetricsForTask(); ioMetricGroup.reuseOutputMetricsForTask(); tmpNumRecordsIn = ioMetricGroup.getNumRecordsInCounter(); } catch (Exception e) { LOG.warn("An exception occurred during the metrics setup.", e); tmpNumRecordsIn = new SimpleCounter(); } numRecordsIn = tmpNumRecordsIn; } if(RichOutputFormat.class.isAssignableFrom(this.format.getClass())){ ((RichOutputFormat) this.format).setRuntimeContext(ctx); LOG.debug(getLogString("Rich Sink detected. Initializing runtime context.")); } ExecutionConfig executionConfig = getExecutionConfig(); boolean objectReuseEnabled = executionConfig.isObjectReuseEnabled(); try { MutableObjectIterator<IT> input1; switch (this.config.getInputLocalStrategy(0)) { case NONE: localStrategy = null; input1 = reader; break; case SORT: try { TypeComparatorFactory<IT> compFact = this.config.getInputComparator(0, getUserCodeClassLoader()); if (compFact == null) { throw new Exception("Missing comparator factory for local strategy on input " + 0); } 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.config.getUseLargeRecordHandler(), 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"); } final TypeSerializer<IT> serializer = this.inputTypeSerializerFactory.getSerializer(); final MutableObjectIterator<IT> input = input1; final OutputFormat<IT> format = this.format; if (this.taskCanceled) { return; } LOG.debug(getLogString("Starting to produce output")); format.open(this.getEnvironment().getTaskInfo().getIndexOfThisSubtask(), this.getEnvironment().getTaskInfo().getNumberOfParallelSubtasks()); if (objectReuseEnabled) { IT record = serializer.createInstance(); while (!this.taskCanceled && ((record = input.next(record)) != null)) { numRecordsIn.inc(); format.writeRecord(record); } } else { IT record; while (!this.taskCanceled && ((record = input.next()) != null)) { numRecordsIn.inc(); format.writeRecord(record); } } if (!this.taskCanceled) { this.format.close(); this.format = null; } } catch (Exception ex) { 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) { throw ex; } else if (!this.taskCanceled) { if (LOG.isErrorEnabled()) { LOG.error(getLogString("Error in user code: " + ex.getMessage()), ex); } throw ex; } } finally { if (this.format != null) { try { this.format.close(); } catch (Throwable t) { if (LOG.isWarnEnabled()) { LOG.warn(getLogString("Error closing the output format"), t); } } } 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")); } }
DataSinkTask extends AbstractInvokable { @Override public void invoke() throws Exception { LOG.debug(getLogString("Start registering input and output")); initOutputFormat(); 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")); LOG.debug(getLogString("Starting data sink operator")); RuntimeContext ctx = createRuntimeContext(); final Counter numRecordsIn; { Counter tmpNumRecordsIn; try { OperatorIOMetricGroup ioMetricGroup = ((OperatorMetricGroup) ctx.getMetricGroup()).getIOMetricGroup(); ioMetricGroup.reuseInputMetricsForTask(); ioMetricGroup.reuseOutputMetricsForTask(); tmpNumRecordsIn = ioMetricGroup.getNumRecordsInCounter(); } catch (Exception e) { LOG.warn("An exception occurred during the metrics setup.", e); tmpNumRecordsIn = new SimpleCounter(); } numRecordsIn = tmpNumRecordsIn; } if(RichOutputFormat.class.isAssignableFrom(this.format.getClass())){ ((RichOutputFormat) this.format).setRuntimeContext(ctx); LOG.debug(getLogString("Rich Sink detected. Initializing runtime context.")); } ExecutionConfig executionConfig = getExecutionConfig(); boolean objectReuseEnabled = executionConfig.isObjectReuseEnabled(); try { MutableObjectIterator<IT> input1; switch (this.config.getInputLocalStrategy(0)) { case NONE: localStrategy = null; input1 = reader; break; case SORT: try { TypeComparatorFactory<IT> compFact = this.config.getInputComparator(0, getUserCodeClassLoader()); if (compFact == null) { throw new Exception("Missing comparator factory for local strategy on input " + 0); } 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.config.getUseLargeRecordHandler(), 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"); } final TypeSerializer<IT> serializer = this.inputTypeSerializerFactory.getSerializer(); final MutableObjectIterator<IT> input = input1; final OutputFormat<IT> format = this.format; if (this.taskCanceled) { return; } LOG.debug(getLogString("Starting to produce output")); format.open(this.getEnvironment().getTaskInfo().getIndexOfThisSubtask(), this.getEnvironment().getTaskInfo().getNumberOfParallelSubtasks()); if (objectReuseEnabled) { IT record = serializer.createInstance(); while (!this.taskCanceled && ((record = input.next(record)) != null)) { numRecordsIn.inc(); format.writeRecord(record); } } else { IT record; while (!this.taskCanceled && ((record = input.next()) != null)) { numRecordsIn.inc(); format.writeRecord(record); } } if (!this.taskCanceled) { this.format.close(); this.format = null; } } catch (Exception ex) { 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) { throw ex; } else if (!this.taskCanceled) { if (LOG.isErrorEnabled()) { LOG.error(getLogString("Error in user code: " + ex.getMessage()), ex); } throw ex; } } finally { if (this.format != null) { try { this.format.close(); } catch (Throwable t) { if (LOG.isWarnEnabled()) { LOG.warn(getLogString("Error closing the output format"), t); } } } 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")); } } }
DataSinkTask extends AbstractInvokable { @Override public void invoke() throws Exception { LOG.debug(getLogString("Start registering input and output")); initOutputFormat(); 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")); LOG.debug(getLogString("Starting data sink operator")); RuntimeContext ctx = createRuntimeContext(); final Counter numRecordsIn; { Counter tmpNumRecordsIn; try { OperatorIOMetricGroup ioMetricGroup = ((OperatorMetricGroup) ctx.getMetricGroup()).getIOMetricGroup(); ioMetricGroup.reuseInputMetricsForTask(); ioMetricGroup.reuseOutputMetricsForTask(); tmpNumRecordsIn = ioMetricGroup.getNumRecordsInCounter(); } catch (Exception e) { LOG.warn("An exception occurred during the metrics setup.", e); tmpNumRecordsIn = new SimpleCounter(); } numRecordsIn = tmpNumRecordsIn; } if(RichOutputFormat.class.isAssignableFrom(this.format.getClass())){ ((RichOutputFormat) this.format).setRuntimeContext(ctx); LOG.debug(getLogString("Rich Sink detected. Initializing runtime context.")); } ExecutionConfig executionConfig = getExecutionConfig(); boolean objectReuseEnabled = executionConfig.isObjectReuseEnabled(); try { MutableObjectIterator<IT> input1; switch (this.config.getInputLocalStrategy(0)) { case NONE: localStrategy = null; input1 = reader; break; case SORT: try { TypeComparatorFactory<IT> compFact = this.config.getInputComparator(0, getUserCodeClassLoader()); if (compFact == null) { throw new Exception("Missing comparator factory for local strategy on input " + 0); } 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.config.getUseLargeRecordHandler(), 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"); } final TypeSerializer<IT> serializer = this.inputTypeSerializerFactory.getSerializer(); final MutableObjectIterator<IT> input = input1; final OutputFormat<IT> format = this.format; if (this.taskCanceled) { return; } LOG.debug(getLogString("Starting to produce output")); format.open(this.getEnvironment().getTaskInfo().getIndexOfThisSubtask(), this.getEnvironment().getTaskInfo().getNumberOfParallelSubtasks()); if (objectReuseEnabled) { IT record = serializer.createInstance(); while (!this.taskCanceled && ((record = input.next(record)) != null)) { numRecordsIn.inc(); format.writeRecord(record); } } else { IT record; while (!this.taskCanceled && ((record = input.next()) != null)) { numRecordsIn.inc(); format.writeRecord(record); } } if (!this.taskCanceled) { this.format.close(); this.format = null; } } catch (Exception ex) { 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) { throw ex; } else if (!this.taskCanceled) { if (LOG.isErrorEnabled()) { LOG.error(getLogString("Error in user code: " + ex.getMessage()), ex); } throw ex; } } finally { if (this.format != null) { try { this.format.close(); } catch (Throwable t) { if (LOG.isWarnEnabled()) { LOG.warn(getLogString("Error closing the output format"), t); } } } 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")); } } DataSinkTask(Environment environment); }
DataSinkTask extends AbstractInvokable { @Override public void invoke() throws Exception { LOG.debug(getLogString("Start registering input and output")); initOutputFormat(); 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")); LOG.debug(getLogString("Starting data sink operator")); RuntimeContext ctx = createRuntimeContext(); final Counter numRecordsIn; { Counter tmpNumRecordsIn; try { OperatorIOMetricGroup ioMetricGroup = ((OperatorMetricGroup) ctx.getMetricGroup()).getIOMetricGroup(); ioMetricGroup.reuseInputMetricsForTask(); ioMetricGroup.reuseOutputMetricsForTask(); tmpNumRecordsIn = ioMetricGroup.getNumRecordsInCounter(); } catch (Exception e) { LOG.warn("An exception occurred during the metrics setup.", e); tmpNumRecordsIn = new SimpleCounter(); } numRecordsIn = tmpNumRecordsIn; } if(RichOutputFormat.class.isAssignableFrom(this.format.getClass())){ ((RichOutputFormat) this.format).setRuntimeContext(ctx); LOG.debug(getLogString("Rich Sink detected. Initializing runtime context.")); } ExecutionConfig executionConfig = getExecutionConfig(); boolean objectReuseEnabled = executionConfig.isObjectReuseEnabled(); try { MutableObjectIterator<IT> input1; switch (this.config.getInputLocalStrategy(0)) { case NONE: localStrategy = null; input1 = reader; break; case SORT: try { TypeComparatorFactory<IT> compFact = this.config.getInputComparator(0, getUserCodeClassLoader()); if (compFact == null) { throw new Exception("Missing comparator factory for local strategy on input " + 0); } 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.config.getUseLargeRecordHandler(), 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"); } final TypeSerializer<IT> serializer = this.inputTypeSerializerFactory.getSerializer(); final MutableObjectIterator<IT> input = input1; final OutputFormat<IT> format = this.format; if (this.taskCanceled) { return; } LOG.debug(getLogString("Starting to produce output")); format.open(this.getEnvironment().getTaskInfo().getIndexOfThisSubtask(), this.getEnvironment().getTaskInfo().getNumberOfParallelSubtasks()); if (objectReuseEnabled) { IT record = serializer.createInstance(); while (!this.taskCanceled && ((record = input.next(record)) != null)) { numRecordsIn.inc(); format.writeRecord(record); } } else { IT record; while (!this.taskCanceled && ((record = input.next()) != null)) { numRecordsIn.inc(); format.writeRecord(record); } } if (!this.taskCanceled) { this.format.close(); this.format = null; } } catch (Exception ex) { 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) { throw ex; } else if (!this.taskCanceled) { if (LOG.isErrorEnabled()) { LOG.error(getLogString("Error in user code: " + ex.getMessage()), ex); } throw ex; } } finally { if (this.format != null) { try { this.format.close(); } catch (Throwable t) { if (LOG.isWarnEnabled()) { LOG.warn(getLogString("Error closing the output format"), t); } } } 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")); } } DataSinkTask(Environment environment); @Override void invoke(); @Override void cancel(); DistributedRuntimeUDFContext createRuntimeContext(); }
DataSinkTask extends AbstractInvokable { @Override public void invoke() throws Exception { LOG.debug(getLogString("Start registering input and output")); initOutputFormat(); 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")); LOG.debug(getLogString("Starting data sink operator")); RuntimeContext ctx = createRuntimeContext(); final Counter numRecordsIn; { Counter tmpNumRecordsIn; try { OperatorIOMetricGroup ioMetricGroup = ((OperatorMetricGroup) ctx.getMetricGroup()).getIOMetricGroup(); ioMetricGroup.reuseInputMetricsForTask(); ioMetricGroup.reuseOutputMetricsForTask(); tmpNumRecordsIn = ioMetricGroup.getNumRecordsInCounter(); } catch (Exception e) { LOG.warn("An exception occurred during the metrics setup.", e); tmpNumRecordsIn = new SimpleCounter(); } numRecordsIn = tmpNumRecordsIn; } if(RichOutputFormat.class.isAssignableFrom(this.format.getClass())){ ((RichOutputFormat) this.format).setRuntimeContext(ctx); LOG.debug(getLogString("Rich Sink detected. Initializing runtime context.")); } ExecutionConfig executionConfig = getExecutionConfig(); boolean objectReuseEnabled = executionConfig.isObjectReuseEnabled(); try { MutableObjectIterator<IT> input1; switch (this.config.getInputLocalStrategy(0)) { case NONE: localStrategy = null; input1 = reader; break; case SORT: try { TypeComparatorFactory<IT> compFact = this.config.getInputComparator(0, getUserCodeClassLoader()); if (compFact == null) { throw new Exception("Missing comparator factory for local strategy on input " + 0); } 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.config.getUseLargeRecordHandler(), 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"); } final TypeSerializer<IT> serializer = this.inputTypeSerializerFactory.getSerializer(); final MutableObjectIterator<IT> input = input1; final OutputFormat<IT> format = this.format; if (this.taskCanceled) { return; } LOG.debug(getLogString("Starting to produce output")); format.open(this.getEnvironment().getTaskInfo().getIndexOfThisSubtask(), this.getEnvironment().getTaskInfo().getNumberOfParallelSubtasks()); if (objectReuseEnabled) { IT record = serializer.createInstance(); while (!this.taskCanceled && ((record = input.next(record)) != null)) { numRecordsIn.inc(); format.writeRecord(record); } } else { IT record; while (!this.taskCanceled && ((record = input.next()) != null)) { numRecordsIn.inc(); format.writeRecord(record); } } if (!this.taskCanceled) { this.format.close(); this.format = null; } } catch (Exception ex) { 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) { throw ex; } else if (!this.taskCanceled) { if (LOG.isErrorEnabled()) { LOG.error(getLogString("Error in user code: " + ex.getMessage()), ex); } throw ex; } } finally { if (this.format != null) { try { this.format.close(); } catch (Throwable t) { if (LOG.isWarnEnabled()) { LOG.warn(getLogString("Error closing the output format"), t); } } } 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")); } } DataSinkTask(Environment environment); @Override void invoke(); @Override void cancel(); DistributedRuntimeUDFContext createRuntimeContext();  }