You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/07/25 14:17:57 UTC
(pulsar) 01/11: [fix][ml] Preserve ledger properties when closing ledger (#26227)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 07d214e41b1738545bc34c7f3d86ca4c8b541b7c
Author: 道君- Tao Jiuming <da...@apache.org>
AuthorDate: Thu Jul 23 12:42:53 2026 +0800
[fix][ml] Preserve ledger properties when closing ledger (#26227)
(cherry picked from commit 79a2c6ff5641fb6fae01576c51d730ed59cdebe1)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 12 ++++++--
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 36 ++++++++++++----------
2 files changed, 28 insertions(+), 20 deletions(-)
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 d5462bd76ebb..0d4530e11e86 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
@@ -1971,9 +1971,15 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
log.debug("[{}] Ledger has been closed id={} entries={}", name, lh.getId(), entriesInLedger);
}
if (entriesInLedger > 0) {
- LedgerInfo info = new LedgerInfo().setLedgerId(lh.getId()).setEntries(entriesInLedger)
- .setSize(lh.getLength()).setTimestamp(clock.millis());
- ledgers.put(lh.getId(), info);
+ ledgers.compute(lh.getId(), (ledgerId, oldInfo) -> {
+ LedgerInfo info = new LedgerInfo();
+ if (oldInfo != null) {
+ info.copyFrom(oldInfo);
+ } else {
+ info.setLedgerId(ledgerId);
+ }
+ return info.setEntries(entriesInLedger).setSize(lh.getLength()).setTimestamp(clock.millis());
+ });
} else {
// The last ledger was empty, so we can discard it
ledgers.remove(lh.getId());
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 ff75e77b9529..b8ae552ff6ab 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
@@ -4668,30 +4668,27 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
private ManagedLedger testSetLedgerProperty0(String name) throws Exception {
ManagedLedgerConfig config = new ManagedLedgerConfig();
config.setMaxEntriesPerLedger(2);
+ config.setMinimumRolloverTime(0, TimeUnit.SECONDS);
ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open(name, config);
ml.addEntry("entry-1".getBytes());
+ long firstLedger = ml.ledgers.firstKey();
+ ml.asyncAddLedgerProperty(firstLedger, "key1", "value1").join();
ml.addEntry("entry-2".getBytes());
ml.addEntry("entry-3".getBytes());
+ long lastLedger = ml.ledgers.lastKey();
+ ml.asyncAddLedgerProperty(lastLedger, "key2", "value2").join();
- Assert.assertEquals(ml.getLedgersInfo().size(), 2);
- long firstLedger = ml.getLedgersInfo().firstKey();
- LedgerInfo firstLedgerInfo = ml.getLedgersInfo().get(firstLedger);
- Assert.assertEquals(firstLedgerInfo.getPropertiesCount(), 0);
- long lastLedger = ml.getLedgersInfo().lastKey();
- LedgerInfo lastLedgerInfo = ml.getLedgersInfo().get(lastLedger);
- Assert.assertEquals(lastLedgerInfo.getPropertiesCount(), 0);
-
- ml.asyncAddLedgerProperty(firstLedger, "key1", "value1").get();
- ml.asyncAddLedgerProperty(lastLedger, "key2", "value2").get();
-
- firstLedgerInfo = ml.getLedgersInfo().get(firstLedger);
- Assert.assertEquals(firstLedgerInfo.getPropertiesCount(), 1);
- Assert.assertEquals(ml.asyncGetLedgerProperty(firstLedger, "key1").get(), "value1");
+ Assert.assertNotEquals(firstLedger, lastLedger);
+ LedgerInfo firstLedgerInfo = ml.ledgers.get(firstLedger);
+ Assert.assertEquals(firstLedgerInfo.getEntries(), 2);
- lastLedgerInfo = ml.getLedgersInfo().get(lastLedger);
- Assert.assertEquals(lastLedgerInfo.getPropertiesCount(), 1);
- Assert.assertEquals(ml.asyncGetLedgerProperty(lastLedger, "key2").get(), "value2");
+ ml.asyncAddLedgerProperty(firstLedger, "key3", "value3").join();
+ ml.asyncAddLedgerProperty(lastLedger, "key4", "value4").join();
+ Assert.assertEquals(ml.asyncGetLedgerProperty(firstLedger, "key1").join(), "value1");
+ Assert.assertEquals(ml.asyncGetLedgerProperty(firstLedger, "key3").join(), "value3");
+ Assert.assertEquals(ml.asyncGetLedgerProperty(lastLedger, "key2").join(), "value2");
+ Assert.assertEquals(ml.asyncGetLedgerProperty(lastLedger, "key4").join(), "value4");
return ml;
}
@@ -4705,6 +4702,11 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
ml.asyncRemoveLedgerProperty(firstLedger, "key1").get();
ml.asyncRemoveLedgerProperty(lastLedger, "key2").get();
+ Assert.assertEquals(ml.getLedgersInfo().get(firstLedger).getPropertiesCount(), 1);
+ Assert.assertEquals(ml.getLedgersInfo().get(lastLedger).getPropertiesCount(), 1);
+
+ ml.asyncRemoveLedgerProperty(firstLedger, "key3").get();
+ ml.asyncRemoveLedgerProperty(lastLedger, "key4").get();
Assert.assertEquals(ml.getLedgersInfo().get(firstLedger).getPropertiesCount(), 0);
Assert.assertEquals(ml.getLedgersInfo().get(lastLedger).getPropertiesCount(), 0);
}