### Question:
DatasetSink extends AbstractSink implements Configurable { @VisibleForTesting void createWriter() throws EventDeliveryException { committedBatch = false; try { View<GenericRecord> view; view = privilegedExecutor.execute( new PrivilegedAction<Dataset<GenericRecord>>() { @Override public Dataset<GenericRecord> run() { return Datasets.load(datasetUri); } }); DatasetDescriptor descriptor = view.getDataset().getDescriptor(); Format format = descriptor.getFormat(); Preconditions.checkArgument(allowedFormats().contains(format.getName()), "Unsupported format: " + format.getName()); Schema newSchema = descriptor.getSchema(); if (datasetSchema == null || !newSchema.equals(datasetSchema)) { this.datasetSchema = descriptor.getSchema(); parser = ENTITY_PARSER_FACTORY.newParser(datasetSchema, context); } this.reuseEntity = !(Formats.PARQUET.equals(format)); this.commitOnBatch = context.getBoolean(CONFIG_FLUSHABLE_COMMIT_ON_BATCH, DEFAULT_FLUSHABLE_COMMIT_ON_BATCH) && (Formats.AVRO.equals(format)); this.syncOnBatch = context.getBoolean(CONFIG_SYNCABLE_SYNC_ON_BATCH, DEFAULT_SYNCABLE_SYNC_ON_BATCH) && (Formats.AVRO.equals(format)); this.datasetName = view.getDataset().getName(); this.writer = view.newWriter(); this.lastRolledMillis = System.currentTimeMillis(); this.bytesParsed = 0L; } catch (DatasetNotFoundException ex) { throw new EventDeliveryException("Dataset " + datasetUri + " not found." + " The dataset must be created before Flume can write to it.", ex); } catch (RuntimeException ex) { throw new EventDeliveryException("Error trying to open a new" + " writer for dataset " + datasetUri, ex); } }  @Override void configure(Context context); @Override synchronized void start(); @Override synchronized void stop(); @Override Status process();  }### Answer:
@Test public void testCreateWriter() 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.commitTransaction(); sink.createWriter(); Assert.assertNotNull("Writer should not be null", sink.getWriter()); Assert.assertEquals("Should have committed", 0, remaining(in)); sink.stop(); Assert.assertEquals(0, read(Datasets.load(FILE_DATASET_URI)).size()); }