@Test public void testCloseWriter() throws EventDeliveryException { config.put(DatasetSinkConstants.CONFIG_FLUSHABLE_COMMIT_ON_BATCH, Boolean.toString(false)); config.put(DatasetSinkConstants.CONFIG_SYNCABLE_SYNC_ON_BATCH, Boolean.toString(false)); DatasetSink sink = sink(in, config); sink.start(); sink.process(); sink.closeWriter(); sink.commitTransaction(); Assert.assertNull("Writer should be null", sink.getWriter()); Assert.assertEquals("Should have committed", 0, remaining(in)); sink.stop(); Assert.assertEquals( Sets.newHashSet(expected), read(Datasets.load(FILE_DATASET_URI))); }
@VisibleForTesting void closeWriter() throws EventDeliveryException { if (writer != null) { try { writer.close(); long elapsedTimeSeconds = TimeUnit.MILLISECONDS.toSeconds( System.currentTimeMillis() - lastRolledMillis); LOG.info("Closed writer for {} after {} seconds and {} bytes parsed", new Object[]{datasetUri, elapsedTimeSeconds, bytesParsed}); } catch (DatasetIOException ex) { throw new EventDeliveryException("Check HDFS permissions/health. IO" + " error trying to close the writer for dataset " + datasetUri, ex); } catch (RuntimeException ex) { throw new EventDeliveryException("Error trying to close the writer for" + " dataset " + datasetUri, ex); } finally { this.writer = null; failurePolicy.close(); } } }
DatasetSink extends AbstractSink implements Configurable, BatchSizeSupported { @VisibleForTesting void closeWriter() throws EventDeliveryException { if (writer != null) { try { writer.close(); long elapsedTimeSeconds = TimeUnit.MILLISECONDS.toSeconds( System.currentTimeMillis() - lastRolledMillis); LOG.info("Closed writer for {} after {} seconds and {} bytes parsed", new Object[]{datasetUri, elapsedTimeSeconds, bytesParsed}); } catch (DatasetIOException ex) { throw new EventDeliveryException("Check HDFS permissions/health. IO" + " error trying to close the writer for dataset " + datasetUri, ex); } catch (RuntimeException ex) { throw new EventDeliveryException("Error trying to close the writer for" + " dataset " + datasetUri, ex); } finally { this.writer = null; failurePolicy.close(); } } } }
DatasetSink extends AbstractSink implements Configurable, BatchSizeSupported { @VisibleForTesting void closeWriter() throws EventDeliveryException { if (writer != null) { try { writer.close(); long elapsedTimeSeconds = TimeUnit.MILLISECONDS.toSeconds( System.currentTimeMillis() - lastRolledMillis); LOG.info("Closed writer for {} after {} seconds and {} bytes parsed", new Object[]{datasetUri, elapsedTimeSeconds, bytesParsed}); } catch (DatasetIOException ex) { throw new EventDeliveryException("Check HDFS permissions/health. IO" + " error trying to close the writer for dataset " + datasetUri, ex); } catch (RuntimeException ex) { throw new EventDeliveryException("Error trying to close the writer for" + " dataset " + datasetUri, ex); } finally { this.writer = null; failurePolicy.close(); } } }  }
DatasetSink extends AbstractSink implements Configurable, BatchSizeSupported { @VisibleForTesting void closeWriter() throws EventDeliveryException { if (writer != null) { try { writer.close(); long elapsedTimeSeconds = TimeUnit.MILLISECONDS.toSeconds( System.currentTimeMillis() - lastRolledMillis); LOG.info("Closed writer for {} after {} seconds and {} bytes parsed", new Object[]{datasetUri, elapsedTimeSeconds, bytesParsed}); } catch (DatasetIOException ex) { throw new EventDeliveryException("Check HDFS permissions/health. IO" + " error trying to close the writer for dataset " + datasetUri, ex); } catch (RuntimeException ex) { throw new EventDeliveryException("Error trying to close the writer for" + " dataset " + datasetUri, ex); } finally { this.writer = null; failurePolicy.close(); } } }  @Override void configure(Context context); @Override synchronized void start(); @Override synchronized void stop(); @Override Status process(); @Override long getBatchSize(); }
DatasetSink extends AbstractSink implements Configurable, BatchSizeSupported { @VisibleForTesting void closeWriter() throws EventDeliveryException { if (writer != null) { try { writer.close(); long elapsedTimeSeconds = TimeUnit.MILLISECONDS.toSeconds( System.currentTimeMillis() - lastRolledMillis); LOG.info("Closed writer for {} after {} seconds and {} bytes parsed", new Object[]{datasetUri, elapsedTimeSeconds, bytesParsed}); } catch (DatasetIOException ex) { throw new EventDeliveryException("Check HDFS permissions/health. IO" + " error trying to close the writer for dataset " + datasetUri, ex); } catch (RuntimeException ex) { throw new EventDeliveryException("Error trying to close the writer for" + " dataset " + datasetUri, ex); } finally { this.writer = null; failurePolicy.close(); } } }  @Override void configure(Context context); @Override synchronized void start(); @Override synchronized void stop(); @Override Status process(); @Override long getBatchSize();  }