[improve][ml]Release idle offloaded read handle only the ref count is 0 (apache#24381)

(cherry picked from commit 21406b5)
(cherry picked from commit 639521b)

Author: poorbarcode

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerHandle.java

@@ -26,4 +26,8 @@ public interface OffloadedLedgerHandle {
     default long lastAccessTimestamp() {
         return -1;
     }
+
+    default int getPendingRead() {
+        return 0;
+    }
 }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

@@ -2647,10 +2647,10 @@ List<Long> internalEvictOffloadedLedgers() {
         ledgerCache.forEach((ledgerId, ledger) -> {
             if (ledger.isDone() && !ledger.isCompletedExceptionally()) {
                 ReadHandle readHandle = ledger.join();
-                if (readHandle instanceof OffloadedLedgerHandle) {
-                   long lastAccessTimestamp = ((OffloadedLedgerHandle) readHandle).lastAccessTimestamp();
-                   if (lastAccessTimestamp >= 0) {
-                       long delta = now - lastAccessTimestamp;
+                if (readHandle instanceof OffloadedLedgerHandle offloadedLedgerHandle) {
+                   int pendingRead = offloadedLedgerHandle.getPendingRead();
+                   if (pendingRead == 0) {
+                       long delta = now - offloadedLedgerHandle.lastAccessTimestamp();
                        if (delta >= inactiveOffloadedLedgerEvictionTimeMs) {
                            log.info("[{}] Offloaded ledger {} can be released ({} ms elapsed since last access)",
                                        name, ledgerId, delta);
@@ -2659,6 +2659,10 @@ List<Long> internalEvictOffloadedLedgers() {
                            log.debug("[{}] Offloaded ledger {} cannot be released ({} ms elapsed since last access)",
                                    name, ledgerId, delta);
                        }
+                        } else if (pendingRead < 0) {
+                            log.error("[{}] Offloaded ledger {} went to a wrong state because its pending read is a"
+                                + " negative value {}. Please raise an issue to https://github.com/apache/pulsar", name,
+                                ledgerId, pendingRead);
                    }
                 }
             }

tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java

@@ -29,6 +29,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.bookkeeper.client.BKException;
 import org.apache.bookkeeper.client.api.LastConfirmedAndEntry;
@@ -56,6 +57,9 @@
 public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedgerHandle {
     private static final Logger log = LoggerFactory.getLogger(BlobStoreBackedReadHandleImpl.class);
 
+    protected static final AtomicIntegerFieldUpdater<BlobStoreBackedReadHandleImpl> PENDING_READ_UPDATER =
+            AtomicIntegerFieldUpdater.newUpdater(BlobStoreBackedReadHandleImpl.class, "pendingRead");
+
     private final long ledgerId;
     private final OffloadIndexBlock index;
     private final BackedInputStream inputStream;
@@ -71,6 +75,8 @@ enum State {
 
     private volatile State state = null;
 
+    private volatile int pendingRead;
+
     private long lastAccessTimestamp = System.currentTimeMillis();
 
     private BlobStoreBackedReadHandleImpl(long ledgerId, OffloadIndexBlock index,
@@ -122,8 +128,17 @@ public CompletableFuture<LedgerEntries> readAsync(long firstEntry, long lastEntr
                     getId(), firstEntry, lastEntry, (1 + lastEntry - firstEntry));
         }
         CompletableFuture<LedgerEntries> promise = new CompletableFuture<>();
+
+        // Ledger handles will be only marked idle when "pendingRead" is "0", it is not needed to update
+        // "lastAccessTimestamp" if "pendingRead" is larger than "0".
+        // Rather than update "lastAccessTimestamp" when starts a reading, updating it when a reading task is finished
+        // is better.
+        PENDING_READ_UPDATER.incrementAndGet(this);
+        promise.whenComplete((__, ex) -> {
+            PENDING_READ_UPDATER.decrementAndGet(BlobStoreBackedReadHandleImpl.this);
+            lastAccessTimestamp = System.currentTimeMillis();
+        });
         executor.execute(() -> {
-            touch();
             if (state == State.Closed) {
                 log.warn("Reading a closed read handler. Ledger ID: {}, Read range: {}-{}",
                         ledgerId, firstEntry, lastEntry);
@@ -212,7 +227,6 @@ public CompletableFuture<LedgerEntries> readAsync(long firstEntry, long lastEntr
     }
 
     private void seekToEntry(long nextExpectedId) throws IOException {
-        touch();
         Long knownOffset = entryOffsetsCache.getIfPresent(ledgerId, nextExpectedId);
         if (knownOffset != null) {
             inputStream.seek(knownOffset);
@@ -323,7 +337,8 @@ public long lastAccessTimestamp() {
         return lastAccessTimestamp;
     }
 
-    private void touch() {
-        lastAccessTimestamp = System.currentTimeMillis();
+    @Override
+    public int getPendingRead() {
+        return PENDING_READ_UPDATER.get(this);
     }
 }