@Beta
    public static void asyncReadCompactedEntries(TopicCompactionService topicCompactionService,
                                                 ManagedCursor cursor, int maxEntries,
                                                 long bytesToRead, Position maxReadPosition,
                                                 boolean readFromEarliest, AsyncCallbacks.ReadEntriesCallback callback,
                                                 boolean wait, @Nullable Consumer consumer) {
        Objects.requireNonNull(topicCompactionService);
        Objects.requireNonNull(cursor);
        checkArgument(maxEntries > 0);
        Objects.requireNonNull(callback);

        final Position readPosition;
        if (readFromEarliest) {
            readPosition = PositionFactory.EARLIEST;
        } else {
            readPosition = cursor.getReadPosition();
        }

        // TODO: redeliver epoch link https://github.com/apache/pulsar/issues/13690
        PersistentDispatcherSingleActiveConsumer.ReadEntriesCtx readEntriesCtx =
                PersistentDispatcherSingleActiveConsumer.ReadEntriesCtx.create(consumer, DEFAULT_CONSUMER_EPOCH);

        CompletableFuture<Position> lastCompactedPositionFuture = topicCompactionService.getLastCompactedPosition();

        lastCompactedPositionFuture.thenCompose(lastCompactedPosition -> {
            if (lastCompactedPosition == null
                    || readPosition.compareTo(
                    lastCompactedPosition.getLedgerId(), lastCompactedPosition.getEntryId()) > 0) {
                if (wait) {
                    cursor.asyncReadEntriesOrWait(maxEntries, bytesToRead, callback, readEntriesCtx, maxReadPosition);
                } else {
                    cursor.asyncReadEntries(maxEntries, bytesToRead, callback, readEntriesCtx, maxReadPosition);
                }
                return CompletableFuture.completedFuture(null);
            }

            ManagedCursorImpl managedCursor = (ManagedCursorImpl) cursor;
            int numberOfEntriesToRead = managedCursor.applyMaxSizeCap(maxEntries, bytesToRead);

            return topicCompactionService.readCompactedEntries(readPosition, numberOfEntriesToRead)
                    .thenAccept(entries -> {
                        if (CollectionUtils.isEmpty(entries)) {
                            Position seekToPosition = lastCompactedPosition.getNext();
                            if (readPosition.compareTo(seekToPosition.getLedgerId(), seekToPosition.getEntryId()) > 0) {
                                seekToPosition = readPosition;
                            }
                            cursor.seek(seekToPosition);
                            callback.readEntriesComplete(Collections.emptyList(), readEntriesCtx);
                            return;
                        }

                        long entriesSize = 0;
                        for (Entry entry : entries) {
                            entriesSize += entry.getLength();
                        }
                        managedCursor.updateReadStats(entries.size(), entriesSize);

                        Entry lastEntry = entries.get(entries.size() - 1);
                        cursor.seek(lastEntry.getPosition().getNext(), true);
                        callback.readEntriesComplete(entries, readEntriesCtx);
                    });
        }).exceptionally((exception) -> {
            exception = FutureUtil.unwrapCompletionException(exception);
            callback.readEntriesFailed(ManagedLedgerException.getManagedLedgerException(exception), readEntriesCtx);
            return null;
        });
    }
@Test
    public void testReadCompactedEntriesWithEmptyEntries() throws ExecutionException, InterruptedException {
        Position lastCompactedPosition = PositionFactory.create(1, 100);
        TopicCompactionService service = Mockito.mock(TopicCompactionService.class);
        Mockito.doReturn(CompletableFuture.completedFuture(Collections.emptyList()))
                .when(service).readCompactedEntries(Mockito.any(), Mockito.intThat(argument -> argument > 0));
        Mockito.doReturn(CompletableFuture.completedFuture(lastCompactedPosition)).when(service)
                .getLastCompactedPosition();


        Position initPosition = PositionFactory.create(1, 90);
        AtomicReference<Position> readPositionRef = new AtomicReference<>(initPosition.getNext());
        ManagedCursorImpl cursor = Mockito.mock(ManagedCursorImpl.class);
        Mockito.doReturn(readPositionRef.get()).when(cursor).getReadPosition();
        Mockito.doReturn(1).when(cursor).applyMaxSizeCap(Mockito.anyInt(), Mockito.anyLong());
        Mockito.doAnswer(invocation -> {
            readPositionRef.set(invocation.getArgument(0));
            return null;
        }).when(cursor).seek(Mockito.any());

        CompletableFuture<List<Entry>> completableFuture = new CompletableFuture<>();
        final AtomicReference<Throwable> throwableRef = new AtomicReference<>();
        AsyncCallbacks.ReadEntriesCallback readEntriesCallback = new AsyncCallbacks.ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                completableFuture.complete(entries);
            }

            @Override
            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                completableFuture.completeExceptionally(exception);
                throwableRef.set(exception);
            }
        };

        CompactedTopicUtils.asyncReadCompactedEntries(service, cursor, 1, 100,
                PositionFactory.LATEST, false, readEntriesCallback, false, null);

        List<Entry> entries = completableFuture.get();
        Assert.assertTrue(entries.isEmpty());
        Assert.assertNull(throwableRef.get());
        Assert.assertEquals(readPositionRef.get(), lastCompactedPosition.getNext());
    }