public synchronized void release(RecordWriter writer, boolean roll) throws StageException, IOException {
    writer.closeLock();
    try {
      if (roll || writer.isIdleClosed() || manager.isOverThresholds(writer)) {
        if (IS_TRACE_ENABLED) {
          LOG.trace("Release '{}'", writer.getPath());
        }
        writers.remove(writer.getPath().toString());
        manager.commitWriter(writer);
      }
    } finally {
      writer.closeUnlock();
    }
    purge();
  }
public synchronized void release(RecordWriter writer, boolean roll) throws StageException, IOException {
    writer.closeLock();
    try {
      if (roll || writer.isIdleClosed() || manager.isOverThresholds(writer)) {
        if (IS_TRACE_ENABLED) {
          LOG.trace("Release '{}'", writer.getPath());
        }
        writers.remove(writer.getPath().toString());
        manager.commitWriter(writer);
      }
    } finally {
      writer.closeUnlock();
    }
    purge();
  }
public synchronized void release ( RecordWriter writer , boolean roll ) throws StageException , IOException { writer . closeLock ( ) ; try { if ( roll || writer . isIdleClosed ( ) || manager . isOverThresholds ( writer ) ) { if ( IS_TRACE_ENABLED ) { LOG . trace ( "Release '{}'" , writer . getPath ( ) ) ; } writers . remove ( writer . getPath ( ) . toString ( ) ) ; manager . commitWriter ( writer ) ; } } finally { writer . closeUnlock ( ) ; } purge ( ) ; }
and the hdfsTarget (in the pipeline runnable thread), calls flushAll
and the hdfsTarget ( in the pipeline runnable thread ) calls flushAll