[fix][broker] Pass bytesToRead when reading compacted entries (apache#20850)

Author: coderzc

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java

@@ -350,8 +350,8 @@ protected void readMoreEntries(Consumer consumer) {
                 havePendingRead = true;
                 if (consumer.readCompacted()) {
                     boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
-                    CompactedTopicUtils.readCompactedEntries(topic.getTopicCompactionService(), cursor, messagesToRead,
-                            readFromEarliest, this, consumer);
+                    CompactedTopicUtils.asyncReadCompactedEntries(topic.getTopicCompactionService(), cursor,
+                            messagesToRead, bytesToRead, readFromEarliest, this, true, consumer);
                 } else {
                     ReadEntriesCtx readEntriesCtx =
                             ReadEntriesCtx.create(consumer, consumer.getConsumerEpoch());

pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopic.java

@@ -33,8 +33,8 @@ public interface CompactedTopic {
     /**
      * Read entries from compacted topic.
      *
-     * @deprecated Use {@link CompactedTopicUtils#readCompactedEntries(TopicCompactionService, ManagedCursor,
-     * int, boolean, ReadEntriesCallback, Consumer)} instead.
+     * @deprecated Use {@link CompactedTopicUtils#asyncReadCompactedEntries(TopicCompactionService, ManagedCursor,
+     * int, long, boolean, ReadEntriesCallback, boolean, Consumer)} instead.
      */
     @Deprecated
     void asyncReadEntriesOrWait(ManagedCursor cursor,

pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicUtils.java

@@ -39,9 +39,11 @@
 public class CompactedTopicUtils {
 
     @Beta
-    public static void readCompactedEntries(TopicCompactionService topicCompactionService, ManagedCursor cursor,
-                                            int numberOfEntriesToRead, boolean readFromEarliest,
-                                            AsyncCallbacks.ReadEntriesCallback callback, @Nullable Consumer consumer) {
+    public static void asyncReadCompactedEntries(TopicCompactionService topicCompactionService,
+                                                 ManagedCursor cursor, int numberOfEntriesToRead,
+                                                 long bytesToRead, boolean readFromEarliest,
+                                                 AsyncCallbacks.ReadEntriesCallback callback,
+                                                 boolean wait, @Nullable Consumer consumer) {
         Objects.requireNonNull(topicCompactionService);
         Objects.requireNonNull(cursor);
         checkArgument(numberOfEntriesToRead > 0);
@@ -64,7 +66,13 @@ public static void readCompactedEntries(TopicCompactionService topicCompactionSe
             if (lastCompactedPosition == null
                     || readPosition.compareTo(
                     lastCompactedPosition.getLedgerId(), lastCompactedPosition.getEntryId()) > 0) {
-                cursor.asyncReadEntriesOrWait(numberOfEntriesToRead, callback, readEntriesCtx, PositionImpl.LATEST);
+                if (wait) {
+                    cursor.asyncReadEntriesOrWait(numberOfEntriesToRead, bytesToRead, callback, readEntriesCtx,
+                        PositionImpl.LATEST);
+                } else {
+                    cursor.asyncReadEntries(numberOfEntriesToRead, bytesToRead, callback, readEntriesCtx,
+                        PositionImpl.LATEST);
+                }
                 return CompletableFuture.completedFuture(null);
             }