AbfsInputStream extends FSInputStream implements CanUnbuffer,
        StreamCapabilities { @Override public int read() throws IOException { byte[] b = new byte[1]; int numberOfBytesRead = read(b, 0, 1); if (numberOfBytesRead < 0) { return -1; } else { return (b[0] & 0xFF); } } AbfsInputStream(
          final AbfsClient client,
          final Statistics statistics,
          final String path,
          final long contentLength,
          final AbfsInputStreamContext abfsInputStreamContext,
          final String eTag); String getPath(); @Override int read(); @Override synchronized int read(final byte[] b, final int off, final int len); @Override synchronized void seek(long n); @Override synchronized long skip(long n); @Override synchronized int available(); long length(); @Override synchronized long getPos(); @Override boolean seekToNewSource(long l); @Override synchronized void close(); @Override synchronized void mark(int readlimit); @Override synchronized void reset(); @Override boolean markSupported(); @Override synchronized void unbuffer(); @Override boolean hasCapability(String capability); @VisibleForTesting AbfsInputStreamStatistics getStreamStatistics(); @VisibleForTesting long getBytesFromReadAhead(); @VisibleForTesting long getBytesFromRemoteRead(); @Override String toString();  }
@Test public void testFailedReadAhead() throws Exception { AbfsClient client = getMockAbfsClient(); AbfsRestOperation successOp = getMockRestOp(); 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) .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, "testFailedReadAhead.txt"); intercept(IOException.class, () -> inputStream.read(new byte[ONE_KB])); verifyReadCallCount(client, 3); checkEvictedStatus(inputStream, 0, false); }
@Test public void testOlderReadAheadFailure() throws Exception { AbfsClient client = getMockAbfsClient(); AbfsRestOperation successOp = getMockRestOp(); doThrow(new TimeoutException("Internal Server error for RAH-X")) .doThrow(new TimeoutException("Internal Server error for RAH-Y")) .doThrow(new TimeoutException("Internal Server error for RAH-Z")) .doReturn(successOp) .doReturn(successOp) .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, "testOlderReadAheadFailure.txt"); intercept(IOException.class, () -> inputStream.read(new byte[ONE_KB])); verifyReadCallCount(client, 3); Thread.sleep(ReadBufferManager.getBufferManager().getThresholdAgeMilliseconds()); inputStream.read(ONE_KB, new byte[ONE_KB], 0, ONE_KB); verifyReadCallCount(client, 4); checkEvictedStatus(inputStream, 0, false); }
@Test public void testSuccessfulReadAhead() throws Exception { AbfsClient client = getMockAbfsClient(); AbfsRestOperation op = getMockRestOp(); doReturn(op) .doReturn(op) .doReturn(op) .doThrow(new TimeoutException("Internal Server error for RAH-X")) .doThrow(new TimeoutException("Internal Server error for RAH-Y")) .doThrow(new TimeoutException("Internal Server error for RAH-Z")) .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, "testSuccessfulReadAhead.txt"); inputStream.read(new byte[ONE_KB]); verifyReadCallCount(client, 3); inputStream.read(ONE_KB, new byte[ONE_KB], 0, ONE_KB); verifyReadCallCount(client, 3); checkEvictedStatus(inputStream, 0, true); }
@Test public void testReadAheadManagerForFailedReadAhead() throws Exception { AbfsClient client = getMockAbfsClient(); AbfsRestOperation successOp = getMockRestOp(); 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) .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); Thread.sleep(1000); intercept(IOException.class, () -> ReadBufferManager.getBufferManager().getBlock( inputStream, 0, ONE_KB, new byte[ONE_KB])); verifyReadCallCount(client, 3); checkEvictedStatus(inputStream, 0, false); }
@Test public void testReadAheadManagerForOlderReadAheadFailure() throws Exception { AbfsClient client = getMockAbfsClient(); AbfsRestOperation successOp = getMockRestOp(); doThrow(new TimeoutException("Internal Server error for RAH-X")) .doThrow(new TimeoutException("Internal Server error for RAH-X")) .doThrow(new TimeoutException("Internal Server error for RAH-X")) .doReturn(successOp) .doReturn(successOp) .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, "testReadAheadManagerForOlderReadAheadFailure.txt"); queueReadAheads(inputStream); Thread.sleep(ReadBufferManager.getBufferManager().getThresholdAgeMilliseconds()); verifyReadCallCount(client, 3); int bytesRead = ReadBufferManager.getBufferManager().getBlock( inputStream, ONE_KB, ONE_KB, new byte[ONE_KB]); Assert.assertEquals("bytesRead should be zero when previously read " + "ahead buffer had failed", 0, bytesRead); checkEvictedStatus(inputStream, 0, false); }
@Test public void testReadAheadManagerForSuccessfulReadAhead() throws Exception { AbfsClient client = getMockAbfsClient(); AbfsRestOperation op = getMockRestOp(); doReturn(op) .doReturn(op) .doReturn(op) .doThrow(new TimeoutException("Internal Server error for RAH-X")) .doThrow(new TimeoutException("Internal Server error for RAH-Y")) .doThrow(new TimeoutException("Internal Server error for RAH-Z")) .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, "testSuccessfulReadAhead.txt"); queueReadAheads(inputStream); Thread.sleep(1000); verifyReadCallCount(client, 3); int bytesRead = ReadBufferManager.getBufferManager().getBlock( inputStream, ONE_KB, ONE_KB, new byte[ONE_KB]); Assert.assertTrue("bytesRead should be non-zero from the " + "buffer that was read-ahead", bytesRead > 0); verifyReadCallCount(client, 3); checkEvictedStatus(inputStream, 0, true); }