@Override
  public int read(long position, byte[] buffer, int offset, int length)
      throws IOException {
    // When bufferedPreadDisabled = true, this API does not use any shared buffer,
    // cursor position etc. So this is implemented as NOT synchronized. HBase
    // kind of random reads on a shared file input stream will greatly get
    // benefited by such implementation.
    // Strict close check at the begin of the API only not for the entire flow.
    synchronized (this) {
      if (closed) {
        throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
      }
    }
    LOG.debug("pread requested offset = {} len = {} bufferedPreadDisabled = {}",
        offset, length, bufferedPreadDisabled);
    if (!bufferedPreadDisabled) {
      return super.read(position, buffer, offset, length);
    }
    validatePositionedReadArgs(position, buffer, offset, length);
    if (length == 0) {
      return 0;
    }
    if (streamStatistics != null) {
      streamStatistics.readOperationStarted();
    }
    int bytesRead = readRemote(position, buffer, offset, length, tracingContext);
    if (statistics != null) {
      statistics.incrementBytesRead(bytesRead);
    }
    if (streamStatistics != null) {
      streamStatistics.bytesRead(bytesRead);
    }
    return bytesRead;
  }
@Test
  public void testFailedReadAheadEviction() throws Exception {
    AbfsClient client = getMockAbfsClient();
    AbfsRestOperation successOp = getMockRestOp();
    ReadBufferManager.setThresholdAgeMilliseconds(INCREASED_READ_BUFFER_AGE_THRESHOLD);
    // Stub :
    // Read request leads to 3 readahead calls: Fail all 3 readahead-client.read()
    // Actual read request fails with the failure in readahead thread
    doThrow(new TimeoutException("Internal Server error"))
        .when(client)
        .read(any(String.class), any(Long.class), any(byte[].class),
            any(Integer.class), any(Integer.class), any(String.class),
            any(String.class), any(), any(TracingContext.class));

    AbfsInputStream inputStream = getAbfsInputStream(client, "testFailedReadAheadEviction.txt");

    // Add a failed buffer to completed queue and set to no free buffers to read ahead.
    ReadBuffer buff = new ReadBuffer();
    buff.setStatus(ReadBufferStatus.READ_FAILED);
    ReadBufferManager.getBufferManager().testMimicFullUseAndAddFailedBuffer(buff);

    // if read failed buffer eviction is tagged as a valid eviction, it will lead to
    // wrong assumption of queue logic that a buffer is freed up and can lead to :
    // java.util.EmptyStackException
    // at java.util.Stack.peek(Stack.java:102)
    // at java.util.Stack.pop(Stack.java:84)
    // at org.apache.hadoop.fs.azurebfs.services.ReadBufferManager.queueReadAhead
    ReadBufferManager.getBufferManager().queueReadAhead(inputStream, 0, ONE_KB,
        getTestTracingContext(getFileSystem(), true));
  }