(pulsar) branch branch-3.0 updated: [fix][ml] No rollover inactive ledgers when metadata service invalid (#22284)

[technoboy]

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

technoboy pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-3.0 by this push:
     new b9bf0a83b5c [fix][ml] No rollover inactive ledgers when metadata 
service invalid (#22284)
b9bf0a83b5c is described below

commit b9bf0a83b5ccf89e036cf77a1c4cb764acf59bba
Author: houxiaoyu <houxia...@apache.org>
AuthorDate: Sat Mar 30 21:38:55 2024 +0800

    [fix][ml] No rollover inactive ledgers when metadata service invalid 
(#22284)
    
    ### Motivation
    
    We should not rollover inactive ledgers when metadata service is invailable.
    
    ### Modifications
    
    Checking metadata service is vailable when schedule 
`checkInactiveLedgerAndRollOver`
---
 .../bookkeeper/mledger/impl/ManagedLedgerImpl.java |  7 ++++---
 .../bookkeeper/mledger/impl/ManagedLedgerTest.java | 24 ++++++++++++++++++++++
 2 files changed, 28 insertions(+), 3 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 da9b1fde27a..8a7b376a7f5 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
@@ -4464,9 +4464,10 @@ public class ManagedLedgerImpl implements ManagedLedger, 
CreateCallback {
 
     @Override
     public boolean checkInactiveLedgerAndRollOver() {
-        long currentTimeMs = System.currentTimeMillis();
-        if (currentLedgerEntries > 0 && inactiveLedgerRollOverTimeMs > 0 && 
currentTimeMs > (lastAddEntryTimeMs
-                + inactiveLedgerRollOverTimeMs)) {
+        if (factory.isMetadataServiceAvailable()
+                && currentLedgerEntries > 0
+                && inactiveLedgerRollOverTimeMs > 0
+                && System.currentTimeMillis() > (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 a08b51cf29f..f918ffdc755 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
@@ -3922,6 +3922,30 @@ public class ManagedLedgerTest extends 
MockedBookKeeperTestCase {
         ledger.close();
     }
 
+    @Test
+    public void testDontRollOverInactiveLedgersWhenMetadataServiceInvalid() 
throws Exception {
+        int inactiveLedgerRollOverTimeMs = 5;
+        @Cleanup("shutdown")
+        ManagedLedgerFactoryImpl factory = spy(new 
ManagedLedgerFactoryImpl(metadataStore, bkc));
+        // mock metadata service invalid
+        when(factory.isMetadataServiceAvailable()).thenReturn(false);
+        ManagedLedgerConfig config = new ManagedLedgerConfig();
+        config.setInactiveLedgerRollOverTime(inactiveLedgerRollOverTimeMs, 
TimeUnit.MILLISECONDS);
+        ManagedLedgerImpl ledger = (ManagedLedgerImpl) 
factory.open("rollover_inactive", config);
+
+        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 {
         ManagedLedgerFactory factory = new 
ManagedLedgerFactoryImpl(metadataStore, bkc);