hdfs-protolib/src/main/java/com/streamsets/pipeline/stage/destination/hdfs/HdfsTarget.java
protected void emptyBatch() throws StageException {
    setBatchTime();
    try {
      hdfsTargetConfigBean.getUGI().doAs(new PrivilegedExceptionAction<Void>() {
        @Override
        public Void run() throws Exception {
          hdfsTargetConfigBean.getCurrentWriters().purge();
          if (hdfsTargetConfigBean.getLateWriters() != null) {
            hdfsTargetConfigBean.getLateWriters().purge();
          }
          return null;
        }
      });
    } catch (Exception ex) {
      throw throwStageException(ex);
    }
  }
protected void emptyBatch() throws StageException {
    setBatchTime();
    try {
      hdfsTargetConfigBean.getUGI().doAs(new PrivilegedExceptionAction<Void>() {
        @Override
        public Void run() throws Exception {
          hdfsTargetConfigBean.getCurrentWriters().purge();
          if (hdfsTargetConfigBean.getLateWriters() != null) {
            hdfsTargetConfigBean.getLateWriters().purge();
          }
          return null;
        }
      });
    } catch (Exception ex) {
      throw throwStageException(ex);
    }
  }
new data.