[fix][broker] First entry will be skipped if opening NonDurableCursor while trimmed ledger is adding first entry. (#24738)

Co-authored-by: Yunze Xu <xyzinfernity@163.com> Co-authored-by: Lari Hotari <lhotari@users.noreply.github.com> (cherry picked from commit 7a120f3)

Author: 3 people

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java

@@ -43,11 +43,12 @@ public class NonDurableCursorImpl extends ManagedCursorImpl {
  // Compare with "latest" position marker by using only the ledger id. Since the C++ client is using 48bits to
  // store the entryId, it's not able to pass a Long.max() as entryId. In this case there's no point to require
  // both ledgerId and entryId to be Long.max()
- if (startCursorPosition == null || startCursorPosition.compareTo(ledger.lastConfirmedEntry) > 0) {
+ Pair<PositionImpl, Long> lastPositionCounter = ledger.getLastPositionAndCounter();
+ if (startCursorPosition == null || startCursorPosition.compareTo(lastPositionCounter.getLeft()) > 0) {
  // Start from last entry
  switch (initialPosition) {
  case Latest:
- initializeCursorPosition(ledger.getLastPositionAndCounter());
+ initializeCursorPosition(lastPositionCounter);
  break;
  case Earliest:
  initializeCursorPosition(ledger.getFirstPositionAndCounter());

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

@@ -19,15 +19,15 @@
 package org.apache.bookkeeper.mledger.impl;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.ENTRIES_ADDED_COUNTER_UPDATER;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNotEquals;
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
-
 import com.google.common.collect.Iterables;
-
+import io.netty.buffer.ByteBuf;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
@@ -37,8 +37,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
-
-import io.netty.buffer.ByteBuf;
 import lombok.Cleanup;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
@@ -52,6 +50,11 @@
 import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.common.api.proto.CommandSubscribe;
+import org.awaitility.Awaitility;
+import org.mockito.Mockito;
+import org.mockito.stubbing.Answer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testng.Assert;
@@ -105,6 +108,58 @@ void testOpenNonDurableCursorAtNonExistentMessageId() throws Exception {
  ledger.close();
  }
 
+ @Test(timeOut = 20000)
+ void testOpenNonDurableCursorWhileLedgerIsAddingFirstEntryAfterTrimmed() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig().setMaxEntriesPerLedger(1)
+ .setRetentionTime(0, TimeUnit.MILLISECONDS);
+ config.setMinimumRolloverTime(0, TimeUnit.MILLISECONDS);
+ @Cleanup
+ ManagedLedgerImpl ledgerSpy =
+ Mockito.spy((ManagedLedgerImpl) factory.open("non_durable_cursor_while_ledger_trimmed", config));
+
+ ledgerSpy.addEntry("message1".getBytes());
+
+ ledgerSpy.rollCurrentLedgerIfFull();
+ Awaitility.waitAtMost(5, TimeUnit.SECONDS).until(() ->
+ ledgerSpy.getLedgersInfoAsList().size() > 1
+ );
+ CompletableFuture<Void> trimFuture = new CompletableFuture<>();
+ ledgerSpy.trimConsumedLedgersInBackground(trimFuture);
+ trimFuture.join();
+
+ // Use (currentLedgerId, -1) as startCursorPosition after ledger was trimmed
+ PositionImpl startCursorPosition = PositionImpl.get(ledgerSpy.getCurrentLedger().getId(), -1);
+ assertTrue(startCursorPosition.compareTo(ledgerSpy.lastConfirmedEntry) > 0);
+
+ CountDownLatch getLastPositionLatch = new CountDownLatch(1);
+ CountDownLatch newNonDurableCursorLatch = new CountDownLatch(1);
+ Mockito.when(ledgerSpy.getLastPositionAndCounter()).then((Answer<Pair<Position, Long>>) invocation -> {
+ newNonDurableCursorLatch.countDown();
+ getLastPositionLatch.await();
+ return Pair.of(ledgerSpy.lastConfirmedEntry, ENTRIES_ADDED_COUNTER_UPDATER.get(ledgerSpy));
+ });
+
+ CompletableFuture<ManagedCursor> cursorFuture = new CompletableFuture<ManagedCursor>()
+ .completeAsync(() ->
+ new NonDurableCursorImpl(bkc, ledgerSpy, "my_test_cursor",
+ startCursorPosition, CommandSubscribe.InitialPosition.Latest, false)
+ );
+ PositionImpl oldLastConfirmedEntry = ledgerSpy.lastConfirmedEntry;
+
+ // Wait until NonDurableCursorImpl constructor invokes ManagedLedgerImpl.getLastPositionAndCounter
+ newNonDurableCursorLatch.await();
+ // Add first entry after ledger was trimmed
+ ledgerSpy.addEntry("message2".getBytes());
+ assertTrue(oldLastConfirmedEntry.compareTo(ledgerSpy.lastConfirmedEntry) < 0);
+
+ // Unblock NonDurableCursorImpl constructor
+ getLastPositionLatch.countDown();
+
+ // cursor should read from lastConfirmedEntry
+ ManagedCursor cursor = cursorFuture.join();
+ assertEquals(cursor.getReadPosition(), ledgerSpy.lastConfirmedEntry);
+ }
+
  @Test(timeOut = 20000)
  void testZNodeBypassed() throws Exception {
  ManagedLedger ledger = factory.open("my_test_ledger");