class DatasetSink extends AbstractSink implements Configurable, BatchSizeSupported {

  @VisibleForTesting
  void createWriter() throws EventDeliveryException {
    // reset the commited flag whenever a new writer is created
    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();
        // dataset schema has changed, create a new parser
        parser = ENTITY_PARSER_FACTORY.newParser(datasetSchema, context);
      }

      this.reuseEntity = !(Formats.PARQUET.equals(format));

      // TODO: Check that the format implements Flushable after CDK-863
      // goes in. For now, just check that the Dataset is Avro format
      this.commitOnBatch = context.getBoolean(CONFIG_FLUSHABLE_COMMIT_ON_BATCH,
          DEFAULT_FLUSHABLE_COMMIT_ON_BATCH) && (Formats.AVRO.equals(format));

      // TODO: Check that the format implements Syncable after CDK-863
      // goes in. For now, just check that the Dataset is Avro 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();

      // Reset the last rolled time and the metrics
      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);
    }
  }

}