### Question:
ProtocolBatchProcessor implements StateTransferBatchProcessor { @VisibleForTesting CompletableFuture<ReadBatch> readRecords(TransferBatchRequest transferBatchRequest) { return CompletableFuture.supplyAsync(() -> { for (int i = 0; i < maxReadRetries; i++) { try { Map<Long, ILogData> records = addressSpaceView.simpleProtocolRead( transferBatchRequest.getAddresses(), readOptions); ReadBatch batch = checkReadRecords( transferBatchRequest.getAddresses(), records, Optional.empty()); if (batch.getStatus() == ReadBatch.ReadStatus.FAILED) { throw new IllegalStateException("Some addresses failed to transfer: " + batch.getFailedAddresses()); } return batch; } catch (WrongEpochException e) { log.warn("readRecords: encountered a wrong epoch exception on try {}: {}.", i, e); throw e; } catch (RuntimeException e) { log.warn("readRecords: encountered an exception on try {}: {}.", i, e); Sleep.sleepUninterruptibly(readSleepDuration); } } throw new ReadBatchException("readRecords: read retries were exhausted"); }); }  @Override CompletableFuture<TransferBatchResponse> transfer(
            TransferBatchRequest transferBatchRequest);  }### Answer:
@Test public void testReadRecordsSuccess() { List<Long> addresses = LongStream.range(0L, 10L).boxed().collect(Collectors.toList()); List<LogData> stubList = createStubList(addresses); Map<Long, ILogData> stubMap = createStubMap(stubList); ReadOptions readOptions = ReadOptions.builder() .waitForHole(true) .clientCacheable(false) .serverCacheable(false) .build(); LogUnitClient logUnitClient = mock(LogUnitClient.class); doReturn(CompletableFuture.completedFuture(true)).when(logUnitClient).writeRange(stubList); AddressSpaceView addressSpaceView = mock(AddressSpaceView.class); doReturn(stubMap).when(addressSpaceView).simpleProtocolRead(addresses, readOptions); ProtocolBatchProcessor batchProcessor = ProtocolBatchProcessor .builder() .logUnitClient(logUnitClient) .addressSpaceView(addressSpaceView) .build(); CompletableFuture<ReadBatch> f = batchProcessor.readRecords(new TransferBatchRequest(addresses, Optional.empty())); ReadBatch join = f.join(); List<LogData> expected = getRecordsFromStubMap(stubMap); assertThat(join.getStatus() == ReadBatch.ReadStatus.SUCCEEDED).isTrue(); assertThat(join.getData()).isEqualTo(expected); }
@Test public void handleWrongEpochException() { List<Long> addresses = LongStream.range(0L, 10L).boxed().collect(Collectors.toList()); List<LogData> secondReadList = createStubList(addresses); ReadOptions readOptions = ReadOptions.builder() .waitForHole(true) .clientCacheable(false) .serverCacheable(false) .build(); LogUnitClient logUnitClient = mock(LogUnitClient.class); doReturn(CompletableFuture.completedFuture(true)).when(logUnitClient).writeRange(secondReadList); AddressSpaceView addressSpaceView = mock(AddressSpaceView.class); doAnswer(answer -> { throw new WrongEpochException(0L); }).when(addressSpaceView).simpleProtocolRead(addresses, readOptions); ProtocolBatchProcessor batchProcessor = ProtocolBatchProcessor .builder() .logUnitClient(logUnitClient) .addressSpaceView(addressSpaceView) .build(); ProtocolBatchProcessor spy = spy(batchProcessor); CompletableFuture<ReadBatch> res = spy.readRecords(TransferBatchRequest.builder().addresses(addresses).build()); assertThatThrownBy(res::join) .isInstanceOf(CompletionException.class) .hasRootCauseInstanceOf(WrongEpochException.class); }