code: @Override
    public void write(Batch batch) throws StageException {
      Iterator<Record> it = batch.getRecords();
      if (it.hasNext()) {
        while (it.hasNext()) {
          Record record = it.next();
          try {
            write(record);
          } catch (OnRecordErrorException ex) {
            switch (getContext().getOnErrorRecord()) {
              case DISCARD:
                break;
              case TO_ERROR:
                getContext().toError(record, ex);
                break;
              case STOP_PIPELINE:
                throw ex;
              default:
                throw new IllegalStateException(Utils.format("It should never happen. OnError '{}'",
                                                             getContext().getOnErrorRecord(), ex));
            }
          }
        }
      } else {
        emptyBatch();
      }
    }
docs: Writes the batch by calling the {@link #write(Record)} method for each record in the batch.
  <p/>
  If the calls to the {@link #write(Record)} throws an {@link OnRecordErrorException}, the error
  handling is done based on the stage 'on record error' configuration, discarded, sent to error, or stopping the
  pipeline.
  
  @param batch the batch of records to write.
  for them to be available to the rest of the pipeline.
  @throws StageException if the <code>Target</code> had an error while writing records.