void readEntries(ReadHandle lh, long firstEntry, long lastEntry, boolean shouldCacheEntry,
                     final AsyncCallbacks.ReadEntriesCallback callback, Object ctx) {
        final PendingReadKey key = new PendingReadKey(firstEntry, lastEntry);

        Map<PendingReadKey, PendingRead> pendingReadsForLedger =
                cachedPendingReads.computeIfAbsent(lh.getId(), (l) -> new ConcurrentHashMap<>());

        boolean listenerAdded = false;
        while (!listenerAdded) {
            AtomicBoolean createdByThisThread = new AtomicBoolean();
            FindPendingReadOutcome findBestCandidateOutcome = findPendingRead(key,
                    pendingReadsForLedger, createdByThisThread);
            PendingRead pendingRead = findBestCandidateOutcome.pendingRead;
            if (findBestCandidateOutcome.needsAdditionalReads()) {
                AsyncCallbacks.ReadEntriesCallback wrappedCallback = new AsyncCallbacks.ReadEntriesCallback() {
                    @Override
                    public void readEntriesComplete(List<Entry> entries, Object ctx) {
                        PendingReadKey missingOnLeft = findBestCandidateOutcome.missingOnLeft;
                        PendingReadKey missingOnRight = findBestCandidateOutcome.missingOnRight;
                        if (missingOnRight != null && missingOnLeft != null) {
                            AsyncCallbacks.ReadEntriesCallback readFromLeftCallback =
                                    new AsyncCallbacks.ReadEntriesCallback() {
                                @Override
                                public void readEntriesComplete(List<Entry> entriesFromLeft, Object dummyCtx1) {
                                    AsyncCallbacks.ReadEntriesCallback readFromRightCallback =
                                            new AsyncCallbacks.ReadEntriesCallback() {
                                        @Override
                                        public void readEntriesComplete(List<Entry> entriesFromRight,
                                                                        Object dummyCtx2) {
                                            List<Entry> finalResult =
                                                    new ArrayList<>(entriesFromLeft.size()
                                                            + entries.size() + entriesFromRight.size());
                                            finalResult.addAll(entriesFromLeft);
                                            finalResult.addAll(entries);
                                            finalResult.addAll(entriesFromRight);
                                            callback.readEntriesComplete(finalResult, ctx);
                                        }

                                        @Override
                                        public void readEntriesFailed(ManagedLedgerException exception,
                                                                      Object dummyCtx3) {
                                            entries.forEach(Entry::release);
                                            entriesFromLeft.forEach(Entry::release);
                                            callback.readEntriesFailed(exception, ctx);
                                        }
                                    };
                                    rangeEntryCache.asyncReadEntry0(lh,
                                            missingOnRight.startEntry, missingOnRight.endEntry,
                                            shouldCacheEntry, readFromRightCallback, null);
                                }

                                @Override
                                public void readEntriesFailed(ManagedLedgerException exception, Object dummyCtx4) {
                                    entries.forEach(Entry::release);
                                    callback.readEntriesFailed(exception, ctx);
                                }
                            };
                            rangeEntryCache.asyncReadEntry0(lh, missingOnLeft.startEntry, missingOnLeft.endEntry,
                                    shouldCacheEntry, readFromLeftCallback, null);
                        } else if (missingOnLeft != null) {
                            AsyncCallbacks.ReadEntriesCallback readFromLeftCallback =
                                    new AsyncCallbacks.ReadEntriesCallback() {

                                        @Override
                                        public void readEntriesComplete(List<Entry> entriesFromLeft,
                                                                        Object dummyCtx5) {
                                            List<Entry> finalResult =
                                                    new ArrayList<>(entriesFromLeft.size() + entries.size());
                                            finalResult.addAll(entriesFromLeft);
                                            finalResult.addAll(entries);
                                            callback.readEntriesComplete(finalResult, ctx);
                                        }

                                        @Override
                                        public void readEntriesFailed(ManagedLedgerException exception,
                                                                      Object dummyCtx6) {
                                            entries.forEach(Entry::release);
                                            callback.readEntriesFailed(exception, ctx);
                                        }
                                    };
                            rangeEntryCache.asyncReadEntry0(lh, missingOnLeft.startEntry, missingOnLeft.endEntry,
                                    shouldCacheEntry, readFromLeftCallback, null);
                        } else if (missingOnRight != null) {
                            AsyncCallbacks.ReadEntriesCallback readFromRightCallback =
                                    new AsyncCallbacks.ReadEntriesCallback() {

                                        @Override
                                        public void readEntriesComplete(List<Entry> entriesFromRight,
                                                                        Object dummyCtx7) {
                                            List<Entry> finalResult =
                                                    new ArrayList<>(entriesFromRight.size() + entries.size());
                                            finalResult.addAll(entries);
                                            finalResult.addAll(entriesFromRight);
                                            callback.readEntriesComplete(finalResult, ctx);
                                        }

                                        @Override
                                        public void readEntriesFailed(ManagedLedgerException exception,
                                                                      Object dummyCtx8) {
                                            entries.forEach(Entry::release);
                                            callback.readEntriesFailed(exception, ctx);
                                        }
                                    };
                            rangeEntryCache.asyncReadEntry0(lh, missingOnRight.startEntry, missingOnRight.endEntry,
                                    shouldCacheEntry, readFromRightCallback, null);
                        }
                    }

                    @Override
                    public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                        callback.readEntriesFailed(exception, ctx);
                    }
                };
                listenerAdded = pendingRead.addListener(wrappedCallback, ctx, key.startEntry, key.endEntry);
            } else {
                listenerAdded = pendingRead.addListener(callback, ctx, key.startEntry, key.endEntry);
            }


            if (createdByThisThread.get()) {
                CompletableFuture<List<EntryImpl>> readResult = rangeEntryCache.readFromStorage(lh, firstEntry,
                        lastEntry, shouldCacheEntry);
                pendingRead.attach(readResult);
            }
        }
    }
@Test
    public void simpleConcurrentReadMissingRight() throws Exception {

        long firstEntry = 100;
        long endEntry = 199;

        long firstEntrySecondRead = firstEntry;
        long endEntrySecondRead = endEntry + 10;

        boolean shouldCacheEntry = false;

        PreparedReadFromStorage read1 =
                prepareReadFromStorage(lh, rangeEntryCache, firstEntry, endEntry, shouldCacheEntry);

        PreparedReadFromStorage readForRight =
                prepareReadFromStorage(lh, rangeEntryCache, endEntry + 1, endEntrySecondRead, shouldCacheEntry);

        PendingReadsManager pendingReadsManager = new PendingReadsManager(rangeEntryCache);
        CapturingReadEntriesCallback callback = new CapturingReadEntriesCallback();
        pendingReadsManager.readEntries(lh, firstEntry, endEntry, shouldCacheEntry, callback, CTX);

        CapturingReadEntriesCallback callback2 = new CapturingReadEntriesCallback();
        pendingReadsManager.readEntries(lh, firstEntrySecondRead, endEntrySecondRead, shouldCacheEntry, callback2, CTX2);

        // complete the read from BK
        read1.storageReadCompleted();
        // the first read can move forward
        callback.get();

        readForRight.storageReadCompleted();
        callback2.get();

        assertSame(callback.getCtx(), CTX);
        assertSame(callback2.getCtx(), CTX2);

        verifyRange(callback.entries, firstEntry, endEntry);
        verifyRange(callback2.entries, firstEntrySecondRead, endEntrySecondRead);

    }