[ci][feat][broker]Support internal cursor properties

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

@@ -97,7 +97,9 @@ enum IndividualDeletedEntries {
     CompletableFuture<Void> putCursorProperty(String key, String value);
 
     /**
-     * Set all properties associated with the cursor.
+     * Set all properties associated with the cursor,
+     * but internal properties(start with '#pulsar_internal.') are still preserved
+     * and prohibit setting of internal properties by this method.
      *
      * Note: {@link ManagedLedgerException.BadVersionException} will be set in this {@link CompletableFuture},
      * if there are concurrent modification and store data has changed.

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

@@ -96,6 +96,7 @@
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.StringProperty;
 import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.common.util.collections.BitSetRecyclable;
 import org.apache.pulsar.common.util.collections.LongPairRangeSet;
 import org.apache.pulsar.common.util.collections.LongPairRangeSet.LongPairConsumer;
@@ -121,6 +122,8 @@ public class ManagedCursorImpl implements ManagedCursor {
     protected final ManagedLedgerImpl ledger;
     private final String name;
 
+    public static final String CURSOR_INTERNAL_PROPERTY_PREFIX = "#pulsar.internal.";
+
     private volatile Map<String, String> cursorProperties;
     private final BookKeeper.DigestType digestType;
 
@@ -371,7 +374,28 @@ public void operationFailed(MetaStoreException e) {
 
     @Override
     public CompletableFuture<Void> setCursorProperties(Map<String, String> cursorProperties) {
-        return computeCursorProperties(lastRead -> cursorProperties);
+        Map<String, String> newProperties =
+                cursorProperties == null ? new HashMap<>() : new HashMap<>(cursorProperties);
+
+        // Prohibit setting of internal properties
+        Set<String> keys = newProperties.keySet();
+        for (String key : keys) {
+            if (key.startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX)) {
+                return FutureUtil.failedFuture(new IllegalArgumentException(
+                        "The property key can't start with " + CURSOR_INTERNAL_PROPERTY_PREFIX));
+            }
+        }
+
+        return computeCursorProperties(lastRead -> {
+            if (lastRead != null) {
+                lastRead.forEach((k, v) -> {
+                    if (k.startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX)) {
+                        newProperties.put(k, v);
+                    }
+                });
+            }
+            return newProperties;
+        });
     }
 
     @Override

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

@@ -18,9 +18,11 @@
  */
 package org.apache.bookkeeper.mledger.impl;
 
+import static org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.CURSOR_INTERNAL_PROPERTY_PREFIX;
 import static org.apache.bookkeeper.mledger.util.Futures.executeWithRetry;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -37,6 +39,8 @@
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
 import org.apache.pulsar.common.api.proto.CommandSubscribe.InitialPosition;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.testng.Assert;
 import org.testng.annotations.Test;
 
 public class ManagedCursorPropertiesTest extends MockedBookKeeperTestCase {
@@ -232,6 +236,32 @@ void testUpdateCursorProperties() throws Exception {
         ledger.close();
 
         factory2.shutdown();
+
+        // Create a new factory to force a managed ledger close and recovery
+        ManagedLedgerFactory factory3 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+        // Reopen the managed ledger
+        ledger = factory3.open("testUpdateCursorProperties", new ManagedLedgerConfig());
+        c1 = ledger.openCursor("c1");
+
+        c1.putCursorProperty(CURSOR_INTERNAL_PROPERTY_PREFIX + "test", "test").get(10, TimeUnit.SECONDS);
+        c1.putCursorProperty("custom4", "custom4").get(10, TimeUnit.SECONDS);
+        c1.setCursorProperties(cursorPropertiesUpdated).get(10, TimeUnit.SECONDS);
+
+        cursorPropertiesUpdated.put(CURSOR_INTERNAL_PROPERTY_PREFIX + "test", "test");
+
+        try {
+            c1.setCursorProperties(cursorPropertiesUpdated).get(10, TimeUnit.SECONDS);
+            Assert.fail("Should fail");
+        } catch (Exception e) {
+            assertTrue(
+                    FutureUtil.unwrapCompletionException(e).getMessage().contains("The property key can't start with"));
+        }
+
+        assertEquals(c1.getCursorProperties(), cursorPropertiesUpdated);
+
+        ledger.close();
+
+        factory3.shutdown();
     }
 
     @Test
@@ -256,7 +286,7 @@ public void testUpdateCursorPropertiesConcurrent() throws Exception {
                 ManagedLedgerException.BadVersionException.class, 3));
 
         for (CompletableFuture<Void> future : futures) {
-            future.get();
+            future.get(10, TimeUnit.SECONDS);
         }
 
         assertEquals(c1.getCursorProperties().get("a"), "2");