oneby-wang commented on code in PR #25101:
URL: https://github.com/apache/pulsar/pull/25101#discussion_r2642281024
##########
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java:
##########
@@ -1385,16 +1390,173 @@ public void markDeleteComplete(Object ctx) {
}
latch.await();
-
assertEquals(c1.getNumberOfEntries(), 0);
+ // Sleep 1s here to wait ledger rollover finished
+ Thread.sleep(1000);
+
// Reopen
- @Cleanup("shutdown")
- ManagedLedgerFactory factory2 = new
ManagedLedgerFactoryImpl(metadataStore, bkc);
+ @Cleanup("shutdown") ManagedLedgerFactory factory2 = new
ManagedLedgerFactoryImpl(metadataStore, bkc);
+ // flaky test case: factory2.open() may throw
MetadataStoreException$BadVersionException, race condition:
+ // 1. my_test_ledger ledger rollover triggers cursor.asyncMarkDelete()
operation.
+ // 2. factory2.open() triggers ledger recovery, read versionA
ManagedLedgerInfo of my_test_ledger ledger.
+ // 3. cursor.asyncMarkDelete() triggers
MetaStoreImpl.asyncUpdateLedgerIds(), update versionB ManagedLedgerInfo
+ // into metaStore.
+ // 4. factory2.open() triggers MetaStoreImpl.asyncUpdateLedgerIds(),
update versionA ManagedLedgerInfo
+ // into metaStore, then throws BadVersionException and moves
my_test_ledger ledger to fenced state.
+ // See PR https://github.com/apache/pulsar/pull/25087.
+ // Recovery open async_mark_delete_blocking_test_ledger ledger,
ledgerId++
ledger = factory2.open("my_test_ledger");
ManagedCursor c2 = ledger.openCursor("c1");
- assertEquals(c2.getMarkDeletedPosition(), lastPosition.get());
+ // Three cases:
+ // 1. cursor recovered with lastPosition markDeletePosition
+ // 2. cursor recovered with (lastPositionLedgerId+1:-1)
markDeletePosition, cursor ledger not rolled over, we
+ // move markDeletePosition to (lastPositionLegderId+2:-1)
+ // 3. cursor recovered with (lastPositionLedgerId+1:-1)
markDeletePosition, cursor ledger rolled over, we
+ // move markDeletePosition to (lastPositionLegderId+3:-1)
+ // See PR https://github.com/apache/pulsar/pull/25087.
+ log.info("c2 markDeletePosition: {}, lastPosition: {}",
c2.getMarkDeletedPosition(), lastPosition);
+ Awaitility.await()
+ .untilAsserted(() ->
assertTrue(c2.getMarkDeletedPosition().compareTo(lastPosition.get()) >= 0));
Review Comment:
Thanks for input, now I know the reason.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
