Make testing simpler

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryTest.java

@@ -20,12 +20,16 @@
 
 import static org.testng.Assert.assertEquals;
 
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 import org.apache.bookkeeper.mledger.ManagedCursor;
 import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
 import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
 import org.apache.bookkeeper.mledger.ManagedLedgerInfo.CursorInfo;
 import org.apache.bookkeeper.mledger.ManagedLedgerInfo.MessageRangeInfo;
 import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+import org.awaitility.Awaitility;
+import org.testng.Assert;
 import org.testng.annotations.Test;
 
 public class ManagedLedgerFactoryTest extends MockedBookKeeperTestCase {
@@ -71,4 +75,41 @@ public void testGetManagedLedgerInfoWithClose() throws Exception {
         assertEquals(mri.to.entryId, 0);
     }
 
+    /**
+     * see: https://github.com/apache/pulsar/pull/18688
+     */
+    @Test
+    public void testConcurrentCloseLedgerAndSwitchLedgerForReproduceIssue() throws Exception {
+        String managedLedgerName = "lg_" + UUID.randomUUID().toString().replaceAll("-", "_");
+
+        ManagedLedgerConfig config = new ManagedLedgerConfig();
+        config.setThrottleMarkDelete(1);
+        config.setMaximumRolloverTime(Integer.MAX_VALUE, TimeUnit.SECONDS);
+        config.setMaxEntriesPerLedger(5);
+
+        // call "switch ledger" and "managedLedger.close" concurrently.
+        ManagedLedgerImpl managedLedger1 = (ManagedLedgerImpl) factory.open(managedLedgerName, config);
+        waitManagedLedgerStateEquals(managedLedger1, ManagedLedgerImpl.State.LedgerOpened);
+        managedLedger1.close();
+
+        ManagedLedgerImpl managedLedger2 = (ManagedLedgerImpl) factory.open(managedLedgerName, config);
+        waitManagedLedgerStateEquals(managedLedger2, ManagedLedgerImpl.State.LedgerOpened);
+
+        // Mock the task create ledger complete now, it will change the state to another value which not is Closed.
+        managedLedger1.createComplete(1, null, null);
+        managedLedger1.close();
+
+        // Verify managedLedger2 is still there.
+        Assert.assertFalse(factory.ledgers.isEmpty());
+        Assert.assertEquals(factory.ledgers.get(managedLedger2.getName()).join(), managedLedger2);
+
+        // cleanup.
+        managedLedger2.close();
+    }
+
+    private void waitManagedLedgerStateEquals(ManagedLedgerImpl managedLedger, ManagedLedgerImpl.State expectedStat){
+        Awaitility.await().untilAsserted(() ->
+                Assert.assertTrue(managedLedger.getState() == expectedStat));
+    }
+
 }