[fix][ml]Fix EOFException after enabled topics offloading (apache#24753)

(cherry picked from commit 3a98763)
(cherry picked from commit 9bc3856)

Author: poorbarcode

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

@@ -45,6 +45,7 @@
 import org.apache.bookkeeper.mledger.offload.jcloud.BackedInputStream;
 import org.apache.bookkeeper.mledger.offload.jcloud.OffloadIndexBlock;
 import org.apache.bookkeeper.mledger.offload.jcloud.OffloadIndexBlockBuilder;
+import org.apache.bookkeeper.mledger.offload.jcloud.OffloadIndexEntry;
 import org.apache.bookkeeper.mledger.offload.jcloud.impl.DataBlockUtils.VersionCheck;
 import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
 import org.apache.pulsar.common.naming.TopicName;
@@ -79,7 +80,8 @@ enum State {
 
     private long lastAccessTimestamp = System.currentTimeMillis();
 
-    private BlobStoreBackedReadHandleImpl(long ledgerId, OffloadIndexBlock index,
+    @VisibleForTesting
+    BlobStoreBackedReadHandleImpl(long ledgerId, OffloadIndexBlock index,
                                           BackedInputStream inputStream, ExecutorService executor,
                                           OffsetsCache entryOffsetsCache) {
         this.ledgerId = ledgerId;
@@ -121,108 +123,212 @@ public CompletableFuture<Void> closeAsync() {
         return promise;
     }
 
-    @Override
-    public CompletableFuture<LedgerEntries> readAsync(long firstEntry, long lastEntry) {
-        if (log.isDebugEnabled()) {
-            log.debug("Ledger {}: reading {} - {} ({} entries}",
-                    getId(), firstEntry, lastEntry, (1 + lastEntry - firstEntry));
+    private class ReadTask implements Runnable {
+        private final long firstEntry;
+        private final long lastEntry;
+        private final CompletableFuture<LedgerEntries> promise;
+        private int seekedAndTryTimes = 0;
+
+        public ReadTask(long firstEntry, long lastEntry, CompletableFuture<LedgerEntries> promise) {
+            this.firstEntry = firstEntry;
+            this.lastEntry = lastEntry;
+            this.promise = promise;
         }
-        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) -> {
-            lastAccessTimestamp = System.currentTimeMillis();
-            PENDING_READ_UPDATER.decrementAndGet(BlobStoreBackedReadHandleImpl.this);
-        });
-        executor.execute(() -> {
+        @Override
+        public void run() {
             if (state == State.Closed) {
                 log.warn("Reading a closed read handler. Ledger ID: {}, Read range: {}-{}",
                         ledgerId, firstEntry, lastEntry);
                 promise.completeExceptionally(new ManagedLedgerException.OffloadReadHandleClosedException());
                 return;
             }
 
-            List<LedgerEntry> entries = new ArrayList<LedgerEntry>();
-            boolean seeked = false;
+            List<LedgerEntry> entryCollector = new ArrayList<LedgerEntry>();
             try {
                 if (firstEntry > lastEntry
-                    || firstEntry < 0
-                    || lastEntry > getLastAddConfirmed()) {
+                        || firstEntry < 0
+                        || lastEntry > getLastAddConfirmed()) {
                     promise.completeExceptionally(new BKException.BKIncorrectParameterException());
                     return;
                 }
                 long entriesToRead = (lastEntry - firstEntry) + 1;
-                long nextExpectedId = firstEntry;
-
-                // checking the data stream has enough data to read to avoid throw EOF exception when reading data.
-                // 12 bytes represent the stream have the length and entryID to read.
-                if (dataStream.available() < 12) {
-                    log.warn("There hasn't enough data to read, current available data has {} bytes,"
-                        + " seek to the first entry {} to avoid EOF exception", inputStream.available(), firstEntry);
-                    seekToEntry(firstEntry);
-                }
+                long expectedEntryId = firstEntry;
+                seekToEntryOffset(firstEntry);
+                seekedAndTryTimes++;
 
                 while (entriesToRead > 0) {
                     long currentPosition = inputStream.getCurrentPosition();
                     int length = dataStream.readInt();
                     if (length < 0) { // hit padding or new block
-                        seekToEntry(nextExpectedId);
+                        seekToEntryOffset(expectedEntryId);
                         continue;
                     }
                     long entryId = dataStream.readLong();
-
-                    if (entryId == nextExpectedId) {
+                    if (entryId == expectedEntryId) {
                         entryOffsetsCache.put(ledgerId, entryId, currentPosition);
                         ByteBuf buf = PulsarByteBufAllocator.DEFAULT.buffer(length, length);
-                        entries.add(LedgerEntryImpl.create(ledgerId, entryId, length, buf));
+                        entryCollector.add(LedgerEntryImpl.create(ledgerId, entryId, length, buf));
                         int toWrite = length;
                         while (toWrite > 0) {
                             toWrite -= buf.writeBytes(dataStream, toWrite);
                         }
                         entriesToRead--;
-                        nextExpectedId++;
-                    } else if (entryId > nextExpectedId && entryId < lastEntry) {
-                        log.warn("The read entry {} is not the expected entry {} but in the range of {} - {},"
-                            + " seeking to the right position", entryId, nextExpectedId, nextExpectedId, lastEntry);
-                        seekToEntry(nextExpectedId);
-                    } else if (entryId < nextExpectedId
-                        && !index.getIndexEntryForEntry(nextExpectedId).equals(index.getIndexEntryForEntry(entryId))) {
-                        log.warn("Read an unexpected entry id {} which is smaller than the next expected entry id {}"
-                        + ", seeking to the right position", entryId, nextExpectedId);
-                        seekToEntry(nextExpectedId);
-                    } else if (entryId > lastEntry) {
-                        // in the normal case, the entry id should increment in order. But if there has random access in
-                        // the read method, we should allow to seek to the right position and the entry id should
-                        // never over to the last entry again.
-                        if (!seeked) {
-                            seekToEntry(nextExpectedId);
-                            seeked = true;
-                            continue;
-                        }
-                        log.info("Expected to read {}, but read {}, which is greater than last entry {}",
-                            nextExpectedId, entryId, lastEntry);
-                        throw new BKException.BKUnexpectedConditionException();
+                        expectedEntryId++;
                     } else {
-                        long ignore = inputStream.skip(length);
+                        handleUnexpectedEntryId(expectedEntryId, entryId);
                     }
                 }
-
-                promise.complete(LedgerEntriesImpl.create(entries));
+                promise.complete(LedgerEntriesImpl.create(entryCollector));
             } catch (Throwable t) {
-                log.error("Failed to read entries {} - {} from the offloader in ledger {}",
-                    firstEntry, lastEntry, ledgerId, t);
+                log.error("Failed to read entries {} - {} from the offloader in ledger {}, current position of input"
+                        + " stream is {}", firstEntry, lastEntry, ledgerId, inputStream.getCurrentPosition(), t);
                 if (t instanceof KeyNotFoundException) {
                     promise.completeExceptionally(new BKException.BKNoSuchLedgerExistsException());
                 } else {
                     promise.completeExceptionally(t);
                 }
-                entries.forEach(LedgerEntry::close);
+                entryCollector.forEach(LedgerEntry::close);
+            }
+        }
+
+        // in the normal case, the entry id should increment in order. But if there has random access in
+        // the read method, we should allow to seek to the right position and the entry id should
+        // never over to the last entry again.
+        private void handleUnexpectedEntryId(long expectedId, long actEntryId) throws Exception {
+            LedgerMetadata ledgerMetadata = getLedgerMetadata();
+            OffloadIndexEntry offsetOfExpectedId = index.getIndexEntryForEntry(expectedId);
+            OffloadIndexEntry offsetOfActId = actEntryId <= getLedgerMetadata().getLastEntryId() && actEntryId >= 0
+                    ? index.getIndexEntryForEntry(actEntryId) : null;
+            String logLine = String.format("Failed to read [ %s ~ %s ] of the ledger %s."
+                    + " Because got a incorrect entry id %s, the offset is %s."
+                    + " The expected entry id is %s, the offset is %s."
+                    + " Have seeked and retry read times: %s. LAC is %s.",
+                    firstEntry, lastEntry, ledgerId,
+                    actEntryId, offsetOfActId == null ? "null because it does not exist"
+                            : String.valueOf(offsetOfActId),
+                    expectedId, String.valueOf(offsetOfExpectedId),
+                    seekedAndTryTimes, ledgerMetadata != null ? ledgerMetadata.getLastEntryId() : "unknown");
+            // If it still fails after tried entries count times, throw the exception.
+            long maxTryTimes = Math.max(3, (lastEntry - firstEntry + 1) >> 2);
+            if (seekedAndTryTimes > maxTryTimes) {
+                log.error(logLine);
+                throw new BKException.BKUnexpectedConditionException();
+            } else {
+                log.warn(logLine);
+            }
+            seekToEntryOffset(expectedId);
+            seekedAndTryTimes++;
+        }
+
+        private void skipPreviousEntry(long startEntryId, long expectedEntryId) throws IOException, BKException {
+            long nextExpectedEntryId = startEntryId;
+            while (nextExpectedEntryId < expectedEntryId) {
+                long offset = inputStream.getCurrentPosition();
+                int len = dataStream.readInt();
+                if (len < 0) {
+                    LedgerMetadata ledgerMetadata = getLedgerMetadata();
+                    OffloadIndexEntry offsetOfExpectedId = index.getIndexEntryForEntry(expectedEntryId);
+                    log.error("Failed to read [ {} ~ {} ] of the ledger {}."
+                        + " Because failed to skip a previous entry {}, len: {}, got a negative len."
+                        + " The expected entry id is {}, the offset is {}."
+                        + " Have seeked and retry read times: {}. LAC is {}.",
+                        firstEntry, lastEntry, ledgerId,
+                        nextExpectedEntryId, len,
+                        expectedEntryId, String.valueOf(offsetOfExpectedId),
+                        seekedAndTryTimes, ledgerMetadata != null ? ledgerMetadata.getLastEntryId() : "unknown");
+                    throw new BKException.BKUnexpectedConditionException();
+                }
+                long entryId = dataStream.readLong();
+                if (entryId == nextExpectedEntryId) {
+                    long skipped = inputStream.skip(len);
+                    if (skipped != len) {
+                        LedgerMetadata ledgerMetadata = getLedgerMetadata();
+                        OffloadIndexEntry offsetOfExpectedId = index.getIndexEntryForEntry(expectedEntryId);
+                        log.error("Failed to read [ {} ~ {} ] of the ledger {}."
+                            + " Because failed to skip a previous entry {}, offset: {}, len: {}, there is no more data."
+                            + " The expected entry id is {}, the offset is {}."
+                            + " Have seeked and retry read times: {}. LAC is {}.",
+                            firstEntry, lastEntry, ledgerId,
+                            entryId, offset, len,
+                            expectedEntryId, String.valueOf(offsetOfExpectedId),
+                            seekedAndTryTimes, ledgerMetadata != null ? ledgerMetadata.getLastEntryId() : "unknown");
+                        throw new BKException.BKUnexpectedConditionException();
+                    }
+                    nextExpectedEntryId++;
+                } else {
+                    LedgerMetadata ledgerMetadata = getLedgerMetadata();
+                    OffloadIndexEntry offsetOfExpectedId = index.getIndexEntryForEntry(expectedEntryId);
+                    log.error("Failed to read [ {} ~ {} ] of the ledger {}."
+                        + " Because got a incorrect entry id {},."
+                        + " The expected entry id is {}, the offset is {}."
+                        + " Have seeked and retry read times: {}. LAC is {}.",
+                        firstEntry, lastEntry, ledgerId,
+                        entryId, expectedEntryId, String.valueOf(offsetOfExpectedId),
+                        seekedAndTryTimes, ledgerMetadata != null ? ledgerMetadata.getLastEntryId() : "unknown");
+                    throw new BKException.BKUnexpectedConditionException();
+                }
+            }
+        }
+
+        private void seekToEntryOffset(long expectedEntryId) throws IOException, BKException {
+            // 1. Try to find the precise index.
+            // 1-1. Precise cached indexes.
+            Long cachedPreciseIndex = entryOffsetsCache.getIfPresent(ledgerId, expectedEntryId);
+            if (cachedPreciseIndex != null) {
+                inputStream.seek(cachedPreciseIndex);
+                return;
+            }
+            // 1-2. Precise persistent indexes.
+            OffloadIndexEntry indexOfNearestEntry = index.getIndexEntryForEntry(expectedEntryId);
+            if (indexOfNearestEntry.getEntryId() == expectedEntryId) {
+                inputStream.seek(indexOfNearestEntry.getDataOffset());
+                return;
             }
+            // 2. Try to use the previous index. Since the entry-0 must have a precise index, we can skip to check
+            //    whether "expectedEntryId" is larger than 0;
+            Long cachedPreviousKnownOffset = entryOffsetsCache.getIfPresent(ledgerId, expectedEntryId - 1);
+            if (cachedPreviousKnownOffset != null) {
+                inputStream.seek(cachedPreviousKnownOffset);
+                skipPreviousEntry(expectedEntryId - 1, expectedEntryId);
+                return;
+            }
+            // 3. Use the persistent index of the nearest entry that is smaller than "expectedEntryId".
+            //    Because it is a sparse index, some entries need to be skipped.
+            if (indexOfNearestEntry.getEntryId() < expectedEntryId) {
+                inputStream.seek(indexOfNearestEntry.getDataOffset());
+                skipPreviousEntry(indexOfNearestEntry.getEntryId(), expectedEntryId);
+            } else {
+                LedgerMetadata ledgerMetadata = getLedgerMetadata();
+                log.error("Failed to read [ {} ~ {} ] of the ledger {}."
+                    + " Because got a incorrect index {} of the entry {}, which is greater than expected."
+                    + " Have seeked and retry read times: {}. LAC is {}.",
+                    firstEntry, lastEntry, ledgerId,
+                    String.valueOf(indexOfNearestEntry), expectedEntryId,
+                    seekedAndTryTimes, ledgerMetadata != null ? ledgerMetadata.getLastEntryId() : "unknown");
+                throw new BKException.BKUnexpectedConditionException();
+            }
+        }
+    }
+
+    @Override
+    public CompletableFuture<LedgerEntries> readAsync(long firstEntry, long lastEntry) {
+        if (log.isDebugEnabled()) {
+            log.debug("Ledger {}: reading {} - {} ({} entries}",
+                    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) -> {
+            lastAccessTimestamp = System.currentTimeMillis();
+            PENDING_READ_UPDATER.decrementAndGet(BlobStoreBackedReadHandleImpl.this);
         });
+        executor.execute(new ReadTask(firstEntry, lastEntry, promise));
         return promise;
     }
 
@@ -238,6 +344,11 @@ private void seekToEntry(long nextExpectedId) throws IOException {
         }
     }
 
+    private void seekToEntry(OffloadIndexEntry offloadIndexEntry) throws IOException {
+        long dataOffset = offloadIndexEntry.getDataOffset();
+        inputStream.seek(dataOffset);
+    }
+
     @Override
     public CompletableFuture<LedgerEntries> readUnconfirmedAsync(long firstEntry, long lastEntry) {
         return readAsync(firstEntry, lastEntry);

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

@@ -18,6 +18,7 @@
  */
 package org.apache.bookkeeper.mledger.offload.jcloud.impl;
 
+import java.util.Objects;
 import org.apache.bookkeeper.mledger.offload.jcloud.OffloadIndexEntry;
 
 /**
@@ -65,5 +66,19 @@ public String toString() {
         return String.format("[eid:%d, part:%d, offset:%d, doffset:%d]",
                 entryId, partId, offset, getDataOffset());
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof OffloadIndexEntryImpl that)) {
+            return false;
+        }
+        return entryId == that.entryId && partId == that.partId && offset == that.offset
+                && blockHeaderSize == that.blockHeaderSize;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(entryId, partId, offset, blockHeaderSize);
+    }
 }