@Test
public void testReadAheadManagerForFailedReadAhead() throws Exception {

  AbfsClient client = getMockAbfsClient();
  AbfsRestOperation successOp = getMockRestOp();
  // 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 for RAH-Thread-X"))
      .doThrow(new TimeoutException("Internal Server error for RAH-Thread-Y"))
      .doThrow(new TimeoutException("Internal Server error RAH-Thread-Z"))
      .doReturn(successOp) // Any extra calls to read, pass it.
      .when(client)
      .read(any(String.class), any(Long.class), any(byte[].class),
          any(Integer.class), any(Integer.class), any(String.class),
          any(String.class));
  AbfsInputStream inputStream = getAbfsInputStream(client, "testReadAheadManagerForFailedReadAhead.txt");
  queueReadAheads(inputStream);
  // AbfsInputStream Read would have waited for the read-ahead for the requested offset
  // as we are testing from ReadAheadManager directly, sleep for a sec to
  // get the read ahead threads to complete
  Thread.sleep(1000);
  // if readAhead failed for specific offset, getBlock should
  // throw exception from the ReadBuffer that failed within last thresholdAgeMilliseconds sec
  intercept(IOException.class,
      () -> ReadBufferManager.getBufferManager().getBlock(
          inputStream,
          0,
          ONE_KB,
          new byte[ONE_KB]));
  // Only the 3 readAhead threads should have triggered client.read
  verifyReadCallCount(client, 3);
  // Stub returns success for the 4th read request, if ReadBuffers still
  // persisted, ReadAheadManager getBlock would have returned exception.
  checkEvictedStatus(inputStream, 0, false);
}