[improve][broker] Don't rollover empty ledgers based on inactivity (#21893)
### Motivation
When `managedLedgerInactiveLedgerRolloverTimeSeconds` is set, let's say to `300` (5 minutes), the ledger will also get rolled in the case when no new entries (messages) were added to the ledger. This doesn't make sense.
Empty ledgers are deleted, but having this extra churn is causing extra load on brokers, bookies, and metadata stores (zookeeper).
### Modifications
Skip rolling the ledger if it is empty.
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 82a3a84..a4bfb5d 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
@@ -4459,7 +4459,8 @@
@Override
public boolean checkInactiveLedgerAndRollOver() {
long currentTimeMs = System.currentTimeMillis();
- if (inactiveLedgerRollOverTimeMs > 0 && currentTimeMs > (lastAddEntryTimeMs + inactiveLedgerRollOverTimeMs)) {
+ if (currentLedgerEntries > 0 && inactiveLedgerRollOverTimeMs > 0 && currentTimeMs > (lastAddEntryTimeMs
+ + inactiveLedgerRollOverTimeMs)) {
log.info("[{}] Closing inactive ledger, last-add entry {}", name, lastAddEntryTimeMs);
if (STATE_UPDATER.compareAndSet(this, State.LedgerOpened, State.ClosingLedger)) {
LedgerHandle currentLedger = this.currentLedger;
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 8430afb..4c92911 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
@@ -3899,6 +3899,30 @@
}
@Test
+ public void testDontRollOverEmptyInactiveLedgers() throws Exception {
+ int inactiveLedgerRollOverTimeMs = 5;
+ ManagedLedgerFactoryConfig factoryConf = new ManagedLedgerFactoryConfig();
+ @Cleanup("shutdown")
+ ManagedLedgerFactory factory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setInactiveLedgerRollOverTime(inactiveLedgerRollOverTimeMs, TimeUnit.MILLISECONDS);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("rollover_inactive", config);
+ ManagedCursor cursor = ledger.openCursor("c1");
+
+ long ledgerId = ledger.currentLedger.getId();
+
+ Thread.sleep(inactiveLedgerRollOverTimeMs * 5);
+ ledger.checkInactiveLedgerAndRollOver();
+
+ Thread.sleep(inactiveLedgerRollOverTimeMs * 5);
+ ledger.checkInactiveLedgerAndRollOver();
+
+ assertEquals(ledger.currentLedger.getId(), ledgerId);
+
+ ledger.close();
+ }
+
+ @Test
public void testOffloadTaskCancelled() throws Exception {
@Cleanup("shutdown")
ManagedLedgerFactory factory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
@@ -4094,6 +4118,7 @@
ManagedLedgerConfig config = new ManagedLedgerConfig();
config.setInactiveLedgerRollOverTime(10, TimeUnit.MILLISECONDS);
ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open(mlName, config);
+ ml.addEntry("entry".getBytes(UTF_8));
MutableBoolean isRolledOver = new MutableBoolean(false);
retryStrategically((test) -> {
