/**
 * This will free sockets and file descriptors held by the stream only when the stream implements
 * org.apache.hadoop.fs.CanUnbuffer. NOT THREAD SAFE. Must be called only when all the clients
 * using this stream to read the blocks have finished reading. If by chance the stream is
 * unbuffered and there are clients still holding this stream for read then on next client read
 * request a new socket will be opened by Datanode without client knowing about it and will serve
 * its read request. Note: If this socket is idle for some time then the DataNode will close the
 * socket and the socket will move into CLOSE_WAIT state and on the next client request on this
 * stream, the current socket will be closed and a new socket will be opened to serve the
 * requests.
 */
@SuppressWarnings({ "rawtypes" })
public void unbuffer() {
  FSDataInputStream stream = this.getStream(this.shouldUseHBaseChecksum());
  if (stream != null) {
    InputStream wrappedStream = stream.getWrappedStream();
    // CanUnbuffer interface was added as part of HDFS-7694 and the fix is available in Hadoop
    // 2.6.4+ and 2.7.1+ versions only so check whether the stream object implements the
    // CanUnbuffer interface or not and based on that call the unbuffer api.
    final Class<? extends InputStream> streamClass = wrappedStream.getClass();
    if (this.instanceOfCanUnbuffer == null) {
      // To ensure we compute whether the stream is instance of CanUnbuffer only once.
      this.instanceOfCanUnbuffer = false;
      if (wrappedStream instanceof CanUnbuffer) {
        this.unbuffer = (CanUnbuffer) wrappedStream;
        this.instanceOfCanUnbuffer = true;
      }
    }
    if (this.instanceOfCanUnbuffer) {
      try {
        this.unbuffer.unbuffer();
      } catch (UnsupportedOperationException e){
        if (isLogTraceEnabled) {
          LOG.trace("Failed to invoke 'unbuffer' method in class " + streamClass
              + " . So there may be the stream does not support unbuffering.", e);
        }
      }
    } else {
      if (isLogTraceEnabled) {
        LOG.trace("Failed to find 'unbuffer' method in class " + streamClass);
      }
    }
  }
}