[fix][broker] Try to fix race condition when setting mark delete properties in ManagedLedgerTest.testTrimmerRaceCondition

Author: oneby-wang

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

@@ -2334,11 +2334,15 @@ protected void internalAsyncMarkDelete(final Position newPosition, Map<String, L
             final MarkDeleteCallback callback, final Object ctx, Runnable alignAcknowledgeStatusAfterPersisted) {
         ledger.mbean.addMarkDeleteOp();
 
-        MarkDeleteEntry mdEntry = new MarkDeleteEntry(newPosition, properties, callback, ctx,
-                alignAcknowledgeStatusAfterPersisted);
-
         // We cannot write to the ledger during the switch, need to wait until the new metadata ledger is available
         synchronized (pendingMarkDeleteOps) {
+            // use given properties or when missing, use the properties from the previous field value
+            MarkDeleteEntry last = pendingMarkDeleteOps.peekLast();
+            Map<String, Long> propertiesToUse =
+                    properties != null ? properties : (last != null ? last.properties : getProperties());
+            MarkDeleteEntry mdEntry = new MarkDeleteEntry(newPosition, propertiesToUse, callback, ctx,
+                    alignAcknowledgeStatusAfterPersisted);
+
             // The state might have changed while we were waiting on the queue mutex
             switch (state) {
             case Closed:

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

@@ -2739,22 +2739,20 @@ public CompletableFuture<Void> maybeUpdateCursorBeforeTrimmingConsumedLedger() {
                 Position finalPosition = lastAckedPosition;
                 log.info("Mark deleting cursor:{} from {} to {} since ledger consumed completely.", cursor,
                         markDeletedPosition, lastAckedPosition);
-                cursor.asyncMarkDelete(lastAckedPosition, cursor.getProperties(),
-                    new MarkDeleteCallback() {
-                        @Override
-                        public void markDeleteComplete(Object ctx) {
-                            log.info("Successfully persisted cursor position for cursor:{} to {}",
-                                    cursor, finalPosition);
-                            future.complete(null);
-                        }
+                cursor.asyncMarkDelete(lastAckedPosition, null, new MarkDeleteCallback() {
+                    @Override
+                    public void markDeleteComplete(Object ctx) {
+                        log.info("Successfully persisted cursor position for cursor:{} to {}", cursor, finalPosition);
+                        future.complete(null);
+                    }
 
-                        @Override
-                        public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
-                            log.warn("Failed to mark delete: {} from {} to {}. ", cursor,
-                                    cursor.getMarkDeletedPosition(), finalPosition, exception);
-                            future.completeExceptionally(exception);
-                        }
-                    }, null);
+                    @Override
+                    public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
+                        log.warn("Failed to mark delete: {} from {} to {}. ", cursor, cursor.getMarkDeletedPosition(),
+                                finalPosition, exception);
+                        future.completeExceptionally(exception);
+                    }
+                }, null);
             } else if (compareResult == 0) {
                 log.debug("No need to reset cursor: {}, last acked position equals to current mark-delete position {}.",
                         cursor, markDeletedPosition);

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

@@ -5204,7 +5204,7 @@ public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
         latch.await();
         assertThat(cursor.getPersistentMarkDeletedPosition()).isGreaterThanOrEqualTo(lastPosition);
         assertThat(ledger.getCursors().getSlowestCursorPosition()).isGreaterThanOrEqualTo(lastPosition);
-        // assertEquals(cursor.getProperties(), properties);
+        assertEquals(cursor.getProperties(), properties);
 
         // 3. Add Entry 2. Triggers second rollover process.
         // This implicitly calls maybeUpdateCursorBeforeTrimmingConsumedLedger due to rollover
@@ -5217,6 +5217,6 @@ public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
         assertEquals(cursor.getPersistentMarkDeletedPosition(), PositionFactory.create(p.getLedgerId(), -1));
 
         // Verify properties are preserved after cursor reset
-        // assertEquals(cursor.getProperties(), properties);
+        assertEquals(cursor.getProperties(), properties);
     }
 }