diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java index 0df07f60d76f4..4167c040ed61c 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java @@ -97,7 +97,9 @@ enum IndividualDeletedEntries { CompletableFuture 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. diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java index 5e761a4fae275..6608a7cea350c 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java +++ b/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 cursorProperties; private final BookKeeper.DigestType digestType; @@ -371,7 +374,28 @@ public void operationFailed(MetaStoreException e) { @Override public CompletableFuture setCursorProperties(Map cursorProperties) { - return computeCursorProperties(lastRead -> cursorProperties); + Map newProperties = + cursorProperties == null ? new HashMap<>() : new HashMap<>(cursorProperties); + + // Prohibit setting of internal properties + Set 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 diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorPropertiesTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorPropertiesTest.java index 2ef8f3eae327c..500de5dd13879 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorPropertiesTest.java +++ b/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 future : futures) { - future.get(); + future.get(10, TimeUnit.SECONDS); } assertEquals(c1.getCursorProperties().get("a"), "2");