### Question:
LongPollReadEntryProcessorV3 extends ReadEntryProcessorV3 implements Watcher<LastAddConfirmedUpdateNotification> { @Override protected ReadResponse readEntry(ReadResponse.Builder readResponseBuilder, long entryId, Stopwatch startTimeSw) throws IOException { if (RequestUtils.shouldPiggybackEntry(readRequest)) { if (!readRequest.hasPreviousLAC() || (BookieProtocol.LAST_ADD_CONFIRMED != entryId)) { logger.error("Incorrect read request, entry piggyback requested incorrectly for ledgerId {} entryId {}", ledgerId, entryId); return buildResponse(readResponseBuilder, StatusCode.EBADREQ, startTimeSw); } else { long knownLAC = requestProcessor.bookie.readLastAddConfirmed(ledgerId); readResponseBuilder.setMaxLAC(knownLAC); if (knownLAC > previousLAC) { entryId = previousLAC + 1; readResponseBuilder.setMaxLAC(knownLAC); if (lastAddConfirmedUpdateTime.isPresent()) { readResponseBuilder.setLacUpdateTimestamp(lastAddConfirmedUpdateTime.get()); } if (logger.isDebugEnabled()) { logger.debug("ReadLAC Piggy Back reading entry:{} from ledger: {}", entryId, ledgerId); } try { return super.readEntry(readResponseBuilder, entryId, true, startTimeSw); } catch (Bookie.NoEntryException e) { requestProcessor.getRequestStats().getReadLastEntryNoEntryErrorCounter().inc(); logger.info( "No entry found while piggyback reading entry {} from ledger {} : previous lac = {}", entryId, ledgerId, previousLAC); return buildResponse(readResponseBuilder, StatusCode.EOK, startTimeSw); } } else { if (knownLAC < previousLAC) { if (logger.isDebugEnabled()) { logger.debug("Found smaller lac when piggy back reading lac and entry from ledger {} :" + " previous lac = {}, known lac = {}", ledgerId, previousLAC, knownLAC); } } return buildResponse(readResponseBuilder, StatusCode.EOK, startTimeSw); } } } else { return super.readEntry(readResponseBuilder, entryId, false, startTimeSw); } } LongPollReadEntryProcessorV3(Request request,
                                 Channel channel,
                                 BookieRequestProcessor requestProcessor,
                                 ExecutorService fenceThreadPool,
                                 ExecutorService longPollThreadPool,
                                 HashedWheelTimer requestTimer); @Override void update(LastAddConfirmedUpdateNotification newLACNotification);  }### Answer:
@Test public void testWatchIsCancelledOnTimeout() throws Exception { Request request = Request.newBuilder() .setHeader(BKPacketHeader.newBuilder() .setTxnId(System.currentTimeMillis()) .setVersion(ProtocolVersion.VERSION_THREE) .setOperation(OperationType.READ_ENTRY) .build()) .setReadRequest(ReadRequest.newBuilder() .setLedgerId(10) .setEntryId(1) .setMasterKey(ByteString.copyFrom(new byte[0])) .setPreviousLAC(0) .setTimeOut(1) .build()) .build(); Channel channel = mock(Channel.class); Bookie bookie = mock(Bookie.class); BookieRequestProcessor requestProcessor = mock(BookieRequestProcessor.class); when(requestProcessor.getBookie()).thenReturn(bookie); when(requestProcessor.getRequestStats()).thenReturn(new RequestStats(NullStatsLogger.INSTANCE)); when(bookie.waitForLastAddConfirmedUpdate(anyLong(), anyLong(), any())) .thenReturn(true); when(bookie.readEntry(anyLong(), anyLong())).thenReturn(Unpooled.buffer()); when(bookie.readLastAddConfirmed(anyLong())).thenReturn(Long.valueOf(1)); CompletableFuture<Void> cancelFuture = new CompletableFuture<>(); doAnswer(invocationOnMock -> { cancelFuture.complete(null); return null; }).when(bookie).cancelWaitForLastAddConfirmedUpdate(anyLong(), any()); LongPollReadEntryProcessorV3 processor = new LongPollReadEntryProcessorV3( request, channel, requestProcessor, executor, executor, timer); processor.run(); cancelFuture.get(10, TimeUnit.SECONDS); }