hdfs-protolib/src/main/java/com/streamsets/pipeline/stage/destination/hdfs/writer/RecordWriterManager.java
public Path commitWriter(RecordWriter writer) throws IOException, StageException {
    Path path = null;
    if ((!writer.isClosed() || writer.isIdleClosed()) && !writer.isRenamed()) {
      // Unset the interrupt flag before close(). InterruptedIOException makes close() fail
      // resulting that the tmp file never gets renamed when stopping the pipeline.
      boolean interrupted = Thread.interrupted();
      try {
        // Since this method is always called from exactly one thread, and
        // we checked to make sure that it was not closed or it was idle closed, this method either closes
        // the file or pushes us into the catch block.
        writer.close();
      } catch (IdleClosedException e) {
        LOG.info("Writer for {} was idle closed, renaming.." , writer.getPath());
      }

      LOG.debug("Path[{}] - Committing Writer", writer.getPath());
      path = renameToFinalName(fs, writer.getPath());
      writer.setRenamed(true);
      LOG.debug("Path[{}] - Committed Writer to '{}'", writer.getPath(), path);
      // Reset the interrupt flag back.
      if (interrupted) {
        Thread.currentThread().interrupt();
      }
    }
    return path;
  }
public Path commitWriter(RecordWriter writer) throws IOException, StageException {
    Path path = null;
    if ((!writer.isClosed() || writer.isIdleClosed()) && !writer.isRenamed()) {
      // Unset the interrupt flag before close(). InterruptedIOException makes close() fail
      // resulting that the tmp file never gets renamed when stopping the pipeline.
      boolean interrupted = Thread.interrupted();
      try {
        // Since this method is always called from exactly one thread, and
        // we checked to make sure that it was not closed or it was idle closed, this method either closes
        // the file or pushes us into the catch block.
        writer.close();
      } catch (IdleClosedException e) {
        LOG.info("Writer for {} was idle closed, renaming.." , writer.getPath());
      }

      LOG.debug("Path[{}] - Committing Writer", writer.getPath());
      path = renameToFinalName(fs, writer.getPath());
      writer.setRenamed(true);
      LOG.debug("Path[{}] - Committed Writer to '{}'", writer.getPath(), path);
      // Reset the interrupt flag back.
      if (interrupted) {
        Thread.currentThread().interrupt();
      }
    }
    return path;
  }
public Path commitWriter ( RecordWriter writer ) throws IOException , StageException { Path path = null ; if ( ( ! writer . isClosed ( ) || writer . isIdleClosed ( ) ) && ! writer . isRenamed ( ) ) { // Unset the interrupt flag before close(). InterruptedIOException makes close() fail // resulting that the tmp file never gets renamed when stopping the pipeline. boolean interrupted = Thread . interrupted ( ) ; try { // Since this method is always called from exactly one thread, and // we checked to make sure that it was not closed or it was idle closed, this method either closes // the file or pushes us into the catch block. writer . close ( ) ; } catch ( IdleClosedException e ) { LOG . info ( "Writer for {} was idle closed, renaming.." , writer . getPath ( ) ) ; } LOG . debug ( "Path[{}] - Committing Writer" , writer . getPath ( ) ) ; path = renameToFinalName ( fs , writer . getPath ( ) ) ; writer . setRenamed ( true ) ; LOG . debug ( "Path[{}] - Committed Writer to '{}'" , writer . getPath ( ) , path ) ; // Reset the interrupt flag back. if ( interrupted ) { Thread . currentThread ( ) . interrupt ( ) ; } } return path ; }
This method must always be called after the closeLock() method on the writer has been called.
This method must always be called after the closeLock () method on the writer has been called .