[pulsar] branch master updated: Fix : The frist position in managedLedger is ahead of the last (#4853)

[sijie]

This is an automated email from the ASF dual-hosted git repository.

sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 46bfb2c  Fix : The frist position in managedLedger is ahead of the 
last (#4853)
46bfb2c is described below

commit 46bfb2c21b9f96866238c79b886458dcde6685c8
Author: Xiaobing Fang <bin...@qq.com>
AuthorDate: Fri Aug 2 21:01:45 2019 +0800

    Fix : The frist position in managedLedger is ahead of the last (#4853)
    
    Fixes #4852
    
    Bug:
    
    After create an empty ledger by sub/unsub and trim ledgers, the first 
position in managedLedger is ahead of the last when create a consumer without 
reopen the managed ledger.
    
    Reason：
    
    Last position will be update when opening managedLedger.
    Because when creating an empty Ledger without reopening managedLedger, the 
last position is not be updated. Since the ledger pointed of the last position 
has been deleted, the first position point to the new ledger.
    
    Fix:
    
    The first position should fall back to last position.
---
 .../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 10 +++++++-
 .../bookkeeper/mledger/impl/ManagedLedgerTest.java | 27 ++++++++++++++++++++++
 2 files changed, 36 insertions(+), 1 deletion(-)

diff --git 
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
 
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 0555936..1095a90 100644
--- 
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ 
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -19,6 +19,7 @@
 package org.apache.bookkeeper.mledger.impl;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
 import static java.lang.Math.min;
 import static org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.FALSE;
 import static org.apache.bookkeeper.mledger.util.SafeRun.safeRun;
@@ -2765,7 +2766,14 @@ public class ManagedLedgerImpl implements ManagedLedger, 
CreateCallback {
 
     PositionImpl getFirstPosition() {
         Long ledgerId = ledgers.firstKey();
-        return ledgerId == null ? null : new PositionImpl(ledgerId, -1);
+        if (ledgerId == null) {
+            return null;
+        }
+        if (ledgerId > lastConfirmedEntry.getLedgerId()) {
+            checkState(ledgers.get(ledgerId).getEntries() == 0);
+            ledgerId = lastConfirmedEntry.getLedgerId();
+        }
+        return new PositionImpl(ledgerId, -1);
     }
 
     PositionImpl getLastPosition() {
diff --git 
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
 
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
index 40726c9..d54f112 100644
--- 
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
+++ 
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
@@ -1693,6 +1693,33 @@ public class ManagedLedgerTest extends 
MockedBookKeeperTestCase {
         ml.close();
     }
 
+    /**
+     * Set retention time = 0 and create a empty ledger,
+     * first position can't higher than last after trim ledgers.
+     */
+    @Test
+    public void testRetention0WithEmptyLedger() throws Exception {
+        ManagedLedgerFactory factory = new ManagedLedgerFactoryImpl(bkc, 
bkc.getZkHandle());
+        ManagedLedgerConfig config = new ManagedLedgerConfig();
+        config.setRetentionTime(0, TimeUnit.MINUTES);
+        config.setMaxEntriesPerLedger(1);
+
+        ManagedLedgerImpl ml = (ManagedLedgerImpl) 
factory.open("deletion_after_retention_test_ledger", config);
+        ManagedCursor c1 = ml.openCursor("c1noretention");
+        ml.addEntry("message1".getBytes());
+        c1.skipEntries(1, IndividualDeletedEntries.Exclude);
+        ml.close();
+
+        // reopen ml
+        ml = (ManagedLedgerImpl) 
factory.open("deletion_after_retention_test_ledger", config);
+        c1 = ml.openCursor("c1noretention");
+        ml.deleteCursor(c1.getName());
+        
ml.internalTrimConsumedLedgers(CompletableFuture.completedFuture(null));
+
+        assertTrue(ml.getFirstPosition().ledgerId <= 
ml.lastConfirmedEntry.ledgerId);
+        ml.close();
+    }
+
     @Test
     public void testInfiniteRetention() throws Exception {
         ManagedLedgerFactory factory = new ManagedLedgerFactoryImpl(bkc, 
bkc.getZkHandle());