[improve][broker] Improve CompactedTopicImpl lock (apache#20697)

Author: coderzc

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

@@ -60,7 +60,7 @@ public class CompactedTopicImpl implements CompactedTopic {
 
     private final BookKeeper bk;
 
-    private PositionImpl compactionHorizon = null;
+    private volatile PositionImpl compactionHorizon = null;
     private volatile CompletableFuture<CompactedTopicContext> compactedTopicContext = null;
 
     public CompactedTopicImpl(BookKeeper bk) {
@@ -91,7 +91,6 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
                                        int numberOfEntriesToRead,
                                        boolean isFirstRead,
                                        ReadEntriesCallback callback, Consumer consumer) {
-        synchronized (this) {
             PositionImpl cursorPosition;
             if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
                 cursorPosition = PositionImpl.EARLIEST;
@@ -101,32 +100,25 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
 
             // TODO: redeliver epoch link https://github.com/apache/pulsar/issues/13690
             ReadEntriesCtx readEntriesCtx = ReadEntriesCtx.create(consumer, DEFAULT_CONSUMER_EPOCH);
-            if (compactionHorizon == null
-                || compactionHorizon.compareTo(cursorPosition) < 0) {
+
+            final PositionImpl currentCompactionHorizon = compactionHorizon;
+
+            if (currentCompactionHorizon == null
+                || currentCompactionHorizon.compareTo(cursorPosition) < 0) {
                 cursor.asyncReadEntriesOrWait(numberOfEntriesToRead, callback, readEntriesCtx, PositionImpl.LATEST);
             } else {
                 compactedTopicContext.thenCompose(
                     (context) -> findStartPoint(cursorPosition, context.ledger.getLastAddConfirmed(), context.cache)
                         .thenCompose((startPoint) -> {
                             // do not need to read the compaction ledger if it is empty.
                             // the cursor just needs to be set to the compaction horizon
-                            if (startPoint == COMPACT_LEDGER_EMPTY) {
-                                cursor.seek(compactionHorizon.getNext());
+                            if (startPoint == COMPACT_LEDGER_EMPTY || startPoint == NEWER_THAN_COMPACTED) {
+                                cursor.seek(currentCompactionHorizon.getNext());
                                 callback.readEntriesComplete(Collections.emptyList(), readEntriesCtx);
                                 return CompletableFuture.completedFuture(null);
-                            }
-                            if (startPoint == NEWER_THAN_COMPACTED && compactionHorizon.compareTo(cursorPosition) < 0) {
-                                cursor.asyncReadEntriesOrWait(numberOfEntriesToRead, callback, readEntriesCtx,
-                                        PositionImpl.LATEST);
-                                return CompletableFuture.completedFuture(null);
                             } else {
                                 long endPoint = Math.min(context.ledger.getLastAddConfirmed(),
                                                          startPoint + numberOfEntriesToRead);
-                                if (startPoint == NEWER_THAN_COMPACTED) {
-                                    cursor.seek(compactionHorizon.getNext());
-                                    callback.readEntriesComplete(Collections.emptyList(), readEntriesCtx);
-                                    return CompletableFuture.completedFuture(null);
-                                }
                                 return readEntries(context.ledger, startPoint, endPoint)
                                     .thenAccept((entries) -> {
                                         Entry lastEntry = entries.get(entries.size() - 1);
@@ -142,15 +134,14 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
                         }))
                     .exceptionally((exception) -> {
                         if (exception.getCause() instanceof NoSuchElementException) {
-                            cursor.seek(compactionHorizon.getNext());
+                            cursor.seek(currentCompactionHorizon.getNext());
                             callback.readEntriesComplete(Collections.emptyList(), readEntriesCtx);
                         } else {
                             callback.readEntriesFailed(new ManagedLedgerException(exception), readEntriesCtx);
                         }
                         return null;
                     });
             }
-        }
     }
 
     static CompletableFuture<Long> findStartPoint(PositionImpl p,
@@ -318,7 +309,7 @@ private static int comparePositionAndMessageId(PositionImpl p, MessageIdData m)
             .compare(p.getEntryId(), m.getEntryId()).result();
     }
 
-    public synchronized Optional<Position> getCompactionHorizon() {
+    public Optional<Position> getCompactionHorizon() {
         return Optional.ofNullable(this.compactionHorizon);
     }