@Test public void testFailingDataSinkTask() { int keyCnt = 100; int valCnt = 20; super.initEnvironment(MEMORY_MANAGER_SIZE, NETWORK_BUFFER_SIZE); super.addInput(new UniformRecordGenerator(keyCnt, valCnt, false), 0); DataSinkTask<Record> testTask = new DataSinkTask<>(this.mockEnv); Configuration stubParams = new Configuration(); File tempTestFile = new File(tempFolder.getRoot(), UUID.randomUUID().toString()); super.registerFileOutputTask(MockFailingOutputFormat.class, tempTestFile.toURI().toString(), stubParams); boolean stubFailed = false; try { testTask.invoke(); } catch (Exception e) { stubFailed = true; } Assert.assertTrue("Function exception was not forwarded.", stubFailed); Assert.assertFalse("Temp output file has not been removed", tempTestFile.exists()); }
@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();  }