[fix][ml] Don't estimate number of entries when ledgers are empty, return 1 instead (#24125)

(cherry picked from commit 228a98f)

Author: lhotari

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

@@ -19,7 +19,6 @@
 package org.apache.bookkeeper.mledger.impl;
 
 import static org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImpl.BOOKKEEPER_READ_OVERHEAD_PER_ENTRY;
-import static org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImpl.DEFAULT_ESTIMATED_ENTRY_SIZE;
 import java.util.Collection;
 import java.util.Map;
 import java.util.NavigableMap;
@@ -97,8 +96,7 @@ static int internalEstimateEntryCountByBytesSize(int maxEntries, long maxSizeByt
 
         long estimatedEntryCount = 0;
         long remainingBytesSize = maxSizeBytes;
-        // Start with a default estimated average size per entry, including any overhead
-        long currentAvgSize = DEFAULT_ESTIMATED_ENTRY_SIZE + BOOKKEEPER_READ_OVERHEAD_PER_ENTRY;
+        long currentAvgSize = 0;
         // Get a collection of ledger info starting from the read position
         Collection<MLDataFormats.ManagedLedgerInfo.LedgerInfo> ledgersAfterReadPosition =
                 ledgersInfo.tailMap(readPosition.getLedgerId(), true).values();
@@ -159,7 +157,26 @@ static int internalEstimateEntryCountByBytesSize(int maxEntries, long maxSizeByt
 
         // Add any remaining bytes to the estimated entry count considering the current average entry size
         if (remainingBytesSize > 0 && estimatedEntryCount < maxEntries) {
-            estimatedEntryCount += remainingBytesSize / currentAvgSize;
+            // need to find the previous non-empty ledger to find the average size
+            if (currentAvgSize == 0) {
+                Collection<MLDataFormats.ManagedLedgerInfo.LedgerInfo> ledgersBeforeReadPosition =
+                        ledgersInfo.headMap(readPosition.getLedgerId(), false).descendingMap().values();
+                for (MLDataFormats.ManagedLedgerInfo.LedgerInfo ledgerInfo : ledgersBeforeReadPosition) {
+                    long ledgerTotalSize = ledgerInfo.getSize();
+                    long ledgerTotalEntries = ledgerInfo.getEntries();
+                    // Skip processing ledgers that have no entries or size
+                    if (ledgerTotalEntries == 0 || ledgerTotalSize == 0) {
+                        continue;
+                    }
+                    // Update the average entry size based on the current ledger's size and entry count
+                    currentAvgSize = Math.max(1, ledgerTotalSize / ledgerTotalEntries)
+                            + BOOKKEEPER_READ_OVERHEAD_PER_ENTRY;
+                    break;
+                }
+            }
+            if (currentAvgSize > 0) {
+                estimatedEntryCount += remainingBytesSize / currentAvgSize;
+            }
         }
 
         // Ensure at least one entry is always returned as the result

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCountEstimatorTest.java

@@ -19,7 +19,6 @@
 package org.apache.bookkeeper.mledger.impl;
 
 import static org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImpl.BOOKKEEPER_READ_OVERHEAD_PER_ENTRY;
-import static org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImpl.DEFAULT_ESTIMATED_ENTRY_SIZE;
 import static org.testng.Assert.assertEquals;
 import java.util.HashSet;
 import java.util.NavigableMap;
@@ -31,7 +30,6 @@
 import org.testng.annotations.Test;
 
 public class EntryCountEstimatorTest {
-
     private NavigableMap<Long, MLDataFormats.ManagedLedgerInfo.LedgerInfo> ledgersInfo;
     private Position readPosition;
     private Long lastLedgerId;
@@ -202,11 +200,39 @@ public void testWithOnlyLastLedgerWhichIsEmpty() {
         beforeLastKey.forEach(ledgersInfo::remove);
         lastLedgerTotalEntries = 0;
         lastLedgerTotalSize = 0;
+        int result = estimateEntryCountByBytesSize(Integer.MAX_VALUE);
+        // expect that result is 1 because the estimation couldn't be done
+        assertEquals(result, 1);
+    }
+
+    @Test
+    public void testWithOnlySecondLastLedgerAndEmptyLastLedger() {
+        readPosition = PositionImpl.LATEST;
+        // remove all but the second last and last ledger
+        long secondLastLedgerId = ledgersInfo.lowerKey(lastLedgerId);
+        Set<Long> beforeSecondLastKey = new HashSet<>(ledgersInfo.headMap(secondLastLedgerId).keySet());
+        beforeSecondLastKey.forEach(ledgersInfo::remove);
+        lastLedgerTotalEntries = 0;
+        lastLedgerTotalSize = 0;
+        long expectedEntries = 50;
+        long requiredSize =
+                expectedEntries * (2000 / 150 + BOOKKEEPER_READ_OVERHEAD_PER_ENTRY);
+        int result = estimateEntryCountByBytesSize(requiredSize);
+        assertEquals(result, expectedEntries);
+    }
+
+    @Test
+    public void testWithMultipleEmptyLedgers() {
+        readPosition = PositionImpl.LATEST;
+        long secondLastLedgerId = ledgersInfo.lowerKey(lastLedgerId);
+        MLDataFormats.ManagedLedgerInfo.LedgerInfo secondLastLedgerInfo = ledgersInfo.get(secondLastLedgerId);
+        // make the second last ledger empty
+        ledgersInfo.put(secondLastLedgerId, secondLastLedgerInfo.toBuilder().setEntries(0).setSize(0).build());
+        lastLedgerTotalEntries = 0;
+        lastLedgerTotalSize = 0;
         long expectedEntries = 50;
-        // when last is empty, DEFAULT_ESTIMATED_ENTRY_SIZE + BOOKKEEPER_READ_OVERHEAD_PER_ENTRY is used
-        // for the average size per entry
         long requiredSize =
-                expectedEntries * (DEFAULT_ESTIMATED_ENTRY_SIZE + BOOKKEEPER_READ_OVERHEAD_PER_ENTRY);
+                expectedEntries * (3000 / 200 + BOOKKEEPER_READ_OVERHEAD_PER_ENTRY);
         int result = estimateEntryCountByBytesSize(requiredSize);
         assertEquals(result, expectedEntries);
     }