@Test public void readsWithFailure() throws IOException { List<Integer> failurePositions = new ArrayList<>(); int lastFailurePosition = 1; while (true) { lastFailurePosition += (2 + randomGenerator.nextInt(READ_COUNT - lastFailurePosition - 1)); if (lastFailurePosition >= (READ_COUNT - 1)) { break; } failurePositions.add(lastFailurePosition); } Stubber stubber = doReturn(generateRandomByte()); for (int i = 1, j = 0; i < READ_COUNT; i++) { if (j < failurePositions.size() && i == failurePositions.get(j)) { stubber = stubber.doThrow(new IOException("test exception")); j++; } else { stubber = stubber.doReturn(generateRandomByte()); } } stubber.when(inputStream).read(); ArgumentCaptor<GetObjectRequest> argumentCaptor = ArgumentCaptor.forClass(GetObjectRequest.class); when(objectStorage.getObject(argumentCaptor.capture())) .thenAnswer( new Answer<GetObjectResponse>() { @Override public GetObjectResponse answer(InvocationOnMock invocationOnMock) { final GetObjectRequest getObjectRequest = invocationOnMock.getArgumentAt(0, GetObjectRequest.class); final Range contentRange = getObjectRequest.getRange() == null ? new Range(0L, null) : getObjectRequest.getRange(); return GetObjectResponse.builder() .inputStream(inputStream) .contentRange(contentRange) .eTag("dummy etag") .contentLength(0L) .build(); } }); final BmcDirectFSInputStream directFSInputStream = new BmcDirectFSInputStream( objectStorage, fileStatus, getObjectRequestBuilderSupplier, statistics); for (int i = 0; i < READ_COUNT; i++) { directFSInputStream.read(); } verify(objectStorage, times(1 + failurePositions.size())) .getObject(any(GetObjectRequest.class)); verify(inputStream, times(READ_COUNT + failurePositions.size())).read(); assertEquals(1 + failurePositions.size(), argumentCaptor.getAllValues().size()); for (int i = 0, counter = 0; i < argumentCaptor.getAllValues().size(); i++) { final Range range = argumentCaptor.getAllValues().get(i).getRange(); if (i == 0) { assertEquals(null, range); } else { assertEquals( failurePositions.get(i - 1) - counter++, range.getStartByte().intValue()); assertEquals(null, range.getEndByte()); } } }
@Override public int read() throws IOException { try { return super.read(); } catch (IOException e) { LOG.warn("Read failed, possibly a stale connection. Will re-attempt.", e); FSStreamUtils.closeQuietly(super.getSourceInputStream()); super.setSourceInputStream(null); return super.read(); } }
BmcDirectFSInputStream extends BmcFSInputStream { @Override public int read() throws IOException { try { return super.read(); } catch (IOException e) { LOG.warn("Read failed, possibly a stale connection. Will re-attempt.", e); FSStreamUtils.closeQuietly(super.getSourceInputStream()); super.setSourceInputStream(null); return super.read(); } } }
BmcDirectFSInputStream extends BmcFSInputStream { @Override public int read() throws IOException { try { return super.read(); } catch (IOException e) { LOG.warn("Read failed, possibly a stale connection. Will re-attempt.", e); FSStreamUtils.closeQuietly(super.getSourceInputStream()); super.setSourceInputStream(null); return super.read(); } } BmcDirectFSInputStream(
            final ObjectStorage objectStorage,
            final FileStatus status,
            final Supplier<GetObjectRequest.Builder> requestBuilder,
            final Statistics statistics); }
BmcDirectFSInputStream extends BmcFSInputStream { @Override public int read() throws IOException { try { return super.read(); } catch (IOException e) { LOG.warn("Read failed, possibly a stale connection. Will re-attempt.", e); FSStreamUtils.closeQuietly(super.getSourceInputStream()); super.setSourceInputStream(null); return super.read(); } } BmcDirectFSInputStream(
            final ObjectStorage objectStorage,
            final FileStatus status,
            final Supplier<GetObjectRequest.Builder> requestBuilder,
            final Statistics statistics); @Override int read(); @Override int read(final byte[] b, final int off, final int len); }
BmcDirectFSInputStream extends BmcFSInputStream { @Override public int read() throws IOException { try { return super.read(); } catch (IOException e) { LOG.warn("Read failed, possibly a stale connection. Will re-attempt.", e); FSStreamUtils.closeQuietly(super.getSourceInputStream()); super.setSourceInputStream(null); return super.read(); } } BmcDirectFSInputStream(
            final ObjectStorage objectStorage,
            final FileStatus status,
            final Supplier<GetObjectRequest.Builder> requestBuilder,
            final Statistics statistics); @Override int read(); @Override int read(final byte[] b, final int off, final int len);  }