You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/10/09 14:19:07 UTC
[pulsar] branch master updated: fix ttl expiry does not take effect
(#12266)
This is an automated email from the ASF dual-hosted git repository.
penghui 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 aa408ff fix ttl expiry does not take effect (#12266)
aa408ff is described below
commit aa408fff0244b4e4a4d1a8b1475b50704c422a63
Author: Aloys <lo...@gmail.com>
AuthorDate: Sat Oct 9 22:18:16 2021 +0800
fix ttl expiry does not take effect (#12266)
Fixes #12265
### Motivation
fix ttl expiry does not take effect
### Modifications
When `ManagedLedgerImpl.getPositionAfterN`
1. check whether the `currentLeder` contains entry when count the entries of `currentLeder`
2. check the result and `LAC` first before return
### Verifying this change
add test in `ManagedLedgerTest.testGetPositionAfterN`
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 9 +++-
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 52 ++++++++++++++++++++++
2 files changed, 60 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 47f38a9..5dd98cb 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
@@ -3171,7 +3171,11 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
// for previous ledgers, LedgerInfo in ZK has the number of entries
if (currentLedger != null && currentLedgerId == currentLedger.getId()) {
lastLedger = true;
- totalEntriesInCurrentLedger = lastConfirmedEntry.getEntryId() + 1;
+ if (currentLedgerEntries > 0) {
+ totalEntriesInCurrentLedger = lastConfirmedEntry.getEntryId() + 1;
+ } else {
+ totalEntriesInCurrentLedger = 0;
+ }
} else {
totalEntriesInCurrentLedger = ledgers.get(currentLedgerId).getEntries();
}
@@ -3199,6 +3203,9 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
PositionImpl positionToReturn = getPreviousPosition(PositionImpl.get(currentLedgerId, currentEntryId));
+ if (positionToReturn.compareTo(lastConfirmedEntry) > 0) {
+ positionToReturn = lastConfirmedEntry;
+ }
if (log.isDebugEnabled()) {
log.debug("getPositionAfterN: Start position {}:{}, startIncluded: {}, Return position {}:{}",
startPosition.getLedgerId(), startPosition.getEntryId(), startRange, positionToReturn.getLedgerId(),
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 f8776fd..4358c2f 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
@@ -2182,6 +2182,58 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
assertTrue(mlInfo.stream().allMatch(ledgerInfo -> ledgerInfo.hasTimestamp()));
}
+
+
+ @Test
+ public void testGetPositionAfterN() throws Exception {
+ ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
+ managedLedgerConfig.setMaxEntriesPerLedger(5);
+ ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) factory.open("testGetPositionAfterN", managedLedgerConfig);
+
+ // open cursor to prevent ledger to be deleted when ledger rollover
+ ManagedCursorImpl managedCursor = (ManagedCursorImpl) managedLedger.openCursor("cursor");
+ Position positionMarkDelete = null;
+ for (int i = 0; i < 10; i ++) {
+ if (i == 3) {
+ positionMarkDelete = managedLedger.addEntry(("entry-" + i).getBytes(Encoding));
+ continue;
+ }
+ managedLedger.addEntry(("entry-" + i).getBytes(Encoding));
+ }
+
+ managedCursor.markDelete(positionMarkDelete);
+
+ //trigger ledger rollover and wait for the new ledger created
+ managedLedger.rollCurrentLedgerIfFull();
+ Awaitility.await().untilAsserted(() -> assertEquals(managedLedger.getLedgersInfo().size(), 3));
+ assertEquals(5, managedLedger.getLedgersInfoAsList().get(0).getEntries());
+ assertEquals(5, managedLedger.getLedgersInfoAsList().get(1).getEntries());
+ assertEquals(0, managedLedger.getLedgersInfoAsList().get(2).getEntries());
+ log.info("### ledgers {}", managedLedger.getLedgersInfo());
+
+ long firstLedger = managedLedger.getLedgersInfo().firstKey();
+ long secondLedger = managedLedger.getLedgersInfoAsList().get(1).getLedgerId();
+
+ PositionImpl startPosition = new PositionImpl(firstLedger, 0);
+
+ PositionImpl targetPosition = managedLedger.getPositionAfterN(startPosition, 1, ManagedLedgerImpl.PositionBound.startExcluded);
+ assertEquals(targetPosition.getLedgerId(), firstLedger);
+ assertEquals(targetPosition.getEntryId(), 1);
+
+ targetPosition = managedLedger.getPositionAfterN(startPosition, 4, ManagedLedgerImpl.PositionBound.startExcluded);
+ assertEquals(targetPosition.getLedgerId(), firstLedger);
+ assertEquals(targetPosition.getEntryId(), 4);
+
+ // test for expiry situation
+ PositionImpl searchPosition = managedLedger.getNextValidPosition((PositionImpl) managedCursor.getMarkDeletedPosition());
+ long length = managedCursor.getNumberOfEntriesInStorage();
+ // return the last confirm entry position if searchPosition is exceed the last confirm entry
+ targetPosition = managedLedger.getPositionAfterN(searchPosition, length, ManagedLedgerImpl.PositionBound.startExcluded);
+ log.info("Target position is {}", targetPosition);
+ assertEquals(targetPosition.getLedgerId(), secondLedger);
+ assertEquals(targetPosition.getEntryId(), 4);
+ }
+
@Test
public void testEstimatedBacklogSize() throws Exception {
ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("testEstimatedBacklogSize");