@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);
  // run the sink
  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)));
}